Added support to read partitioned parquet files from S3

Util/channel/src/main/java/io/deephaven/util/channel/CachedChannelProvider.java

@@ -19,6 +19,7 @@
 import java.nio.channels.SeekableByteChannel;
 import java.nio.file.Path;
 import java.util.*;
+import java.util.function.Predicate;
 
 /**
  * {@link SeekableChannelsProvider Channel provider} that will cache a bounded number of unused channels.
@@ -109,6 +110,12 @@ public SeekableByteChannel getWriteChannel(@NotNull final Path path, final boole
                                                                // end no matter what.
     }
 
+    @Override
+    public List<URI> getChildURIListFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
+            throws IOException {
+        return wrappedProvider.getChildURIListFromDirectory(directoryURI, uriFilter);
+    }
+
     @Nullable
     private synchronized CachedChannel tryGetPooledChannel(@NotNull final String pathKey,
             @NotNull final KeyedObjectHashMap<String, PerPathPool> channelPool) {

Util/channel/src/main/java/io/deephaven/util/channel/SeekableChannelsProvider.java

@@ -14,6 +14,8 @@
 import java.nio.channels.SeekableByteChannel;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.List;
+import java.util.function.Predicate;
 
 public interface SeekableChannelsProvider extends SafeCloseable {
 
@@ -106,4 +108,17 @@ default SeekableByteChannel getWriteChannel(@NotNull final String path, final bo
     }
 
     SeekableByteChannel getWriteChannel(@NotNull Path path, boolean append) throws IOException;
+
+    /**
+     * Get a list of URIs of files contained inside the given directory. The URIs are expected to be the direct children
+     * of the directory, and not recursively include files from subdirectories.
+     *
+     * @param directoryURI The URI of the directory to list
+     * @param uriFilter A filter to apply to the URIs in the directory
+     */
+    default List<URI> getChildURIListFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
+            throws IOException {
+        throw new UnsupportedOperationException("TODO Add this support for all providers");
+    }
+
 }

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/ParquetTools.java

@@ -914,7 +914,7 @@ public static Table readKeyValuePartitionedTable(
      * Callers wishing to be more explicit and skip the inference step may prefer to call
      * {@link #readFlatPartitionedTable(File, ParquetInstructions, TableDefinition)}.
      *
-     * @param directory the source of {@link ParquetTableLocationKey location keys} to include
+     * @param directory the directory to search for .parquet files
      * @param readInstructions the instructions for customizations while reading
      * @return the table
      * @see #readPartitionedTable(TableLocationKeyFinder, ParquetInstructions)
@@ -926,11 +926,32 @@ public static Table readFlatPartitionedTable(
         return readPartitionedTable(new ParquetFlatPartitionedLayout(directory, readInstructions), readInstructions);
     }
 
+    /**
+     * Creates a partitioned table via the flat parquet files from the root {@code directory}, inferring the table
+     * definition from those files.
+     *
+     * <p>
+     * Callers wishing to be more explicit and skip the inference step may prefer to call
+     * {@link #readFlatPartitionedTable(File, ParquetInstructions, TableDefinition)}.
+     *
+     * @param directory the path or URI for the directory to search for .parquet files
+     * @param readInstructions the instructions for customizations while reading
+     * @return the table
+     * @see #readPartitionedTable(TableLocationKeyFinder, ParquetInstructions)
+     * @see ParquetFlatPartitionedLayout#ParquetFlatPartitionedLayout(File, ParquetInstructions)
+     */
+    public static Table readFlatPartitionedTable(
+            @NotNull final String directory,
+            @NotNull final ParquetInstructions readInstructions) {
+        return readPartitionedTable(new ParquetFlatPartitionedLayout(convertToURI(directory), readInstructions),
+                readInstructions);
+    }
+
     /**
      * Creates a partitioned table via the flat parquet files from the root {@code directory} using the provided
      * {@code tableDefinition}.
      *
-     * @param directory the source of {@link ParquetTableLocationKey location keys} to include
+     * @param directory the directory to search for .parquet files
      * @param readInstructions the instructions for customizations while reading
      * @param tableDefinition the table definition
      * @return the table
@@ -945,6 +966,25 @@ public static Table readFlatPartitionedTable(
                 tableDefinition);
     }
 
+    /**
+     * Creates a partitioned table via the flat parquet files from the root {@code directory} using the provided
+     * {@code tableDefinition}.
+     *
+     * @param directory the path or URI for the directory to search for .parquet files
+     * @param readInstructions the instructions for customizations while reading
+     * @param tableDefinition the table definition
+     * @return the table
+     * @see #readPartitionedTable(TableLocationKeyFinder, ParquetInstructions, TableDefinition)
+     * @see ParquetFlatPartitionedLayout#ParquetFlatPartitionedLayout(File, ParquetInstructions)
+     */
+    public static Table readFlatPartitionedTable(
+            @NotNull final String directory,
+            @NotNull final ParquetInstructions readInstructions,
+            @NotNull final TableDefinition tableDefinition) {
+        return readPartitionedTable(new ParquetFlatPartitionedLayout(convertToURI(directory), readInstructions),
+                readInstructions, tableDefinition);
+    }
+
     /**
      * Creates a single table via the parquet {@code file} using the table definition derived from that {@code file}.
      *

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/layout/ParquetFileHelper.java

@@ -2,6 +2,7 @@
 
 import io.deephaven.parquet.table.ParquetTableWriter;
 
+import java.net.URI;
 import java.nio.file.Path;
 
 final class ParquetFileHelper {
@@ -12,4 +13,11 @@ static boolean fileNameMatches(final Path path) {
         final String fileName = path.getFileName().toString();
         return fileName.endsWith(ParquetTableWriter.PARQUET_FILE_EXTENSION) && fileName.charAt(0) != '.';
     }
+
+    static boolean fileNameMatches(final URI uri) {
+        // TODO Test this for file paths
+        final String path = uri.getPath();
+        final String fileName = path.substring(path.lastIndexOf('/') + 1);
+        return fileName.endsWith(ParquetTableWriter.PARQUET_FILE_EXTENSION) && fileName.charAt(0) != '.';
+    }
 }

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/layout/ParquetFlatPartitionedLayout.java

@@ -7,14 +7,15 @@
 import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder;
 import io.deephaven.parquet.table.ParquetInstructions;
 import io.deephaven.parquet.table.location.ParquetTableLocationKey;
+import io.deephaven.util.channel.SeekableChannelsProvider;
+import io.deephaven.util.channel.SeekableChannelsProviderLoader;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.File;
 import java.io.IOException;
-import java.nio.file.DirectoryStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
+import java.net.URI;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -23,21 +24,30 @@
  */
 public final class ParquetFlatPartitionedLayout implements TableLocationKeyFinder<ParquetTableLocationKey> {
 
-    private static ParquetTableLocationKey locationKey(Path path, @NotNull final ParquetInstructions readInstructions) {
-        return new ParquetTableLocationKey(path.toFile(), 0, null, readInstructions);
+    private static ParquetTableLocationKey locationKey(URI uri, @NotNull final ParquetInstructions readInstructions) {
+        return new ParquetTableLocationKey(uri, 0, null, readInstructions);
     }
 
-    private final File tableRootDirectory;
-    private final Map<Path, ParquetTableLocationKey> cache;
+    private final URI tableRootDirectory;
+    private final Map<URI, ParquetTableLocationKey> cache;
     private final ParquetInstructions readInstructions;
 
     /**
-     * @param tableRootDirectory The directory to search for .parquet files.
+     * @param tableRootDirectoryFile The directory to search for .parquet files.
      * @param readInstructions the instructions for customizations while reading
      */
-    public ParquetFlatPartitionedLayout(@NotNull final File tableRootDirectory,
+    public ParquetFlatPartitionedLayout(@NotNull final File tableRootDirectoryFile,
             @NotNull final ParquetInstructions readInstructions) {
-        this.tableRootDirectory = tableRootDirectory;
+        this(tableRootDirectoryFile.toURI(), readInstructions);
+    }
+
+    /**
+     * @param tableRootDirectoryURI The directory URI to search for .parquet files.
+     * @param readInstructions the instructions for customizations while reading
+     */
+    public ParquetFlatPartitionedLayout(@NotNull final URI tableRootDirectoryURI,
+            @NotNull final ParquetInstructions readInstructions) {
+        this.tableRootDirectory = tableRootDirectoryURI;
         this.cache = new HashMap<>();
         this.readInstructions = readInstructions;
     }
@@ -47,19 +57,25 @@ public String toString() {
     }
 
     @Override
-    public synchronized void findKeys(@NotNull final Consumer<ParquetTableLocationKey> locationKeyObserver) {
-        try (final DirectoryStream<Path> parquetFileStream =
-                Files.newDirectoryStream(tableRootDirectory.toPath(), ParquetFileHelper::fileNameMatches)) {
-            for (final Path parquetFilePath : parquetFileStream) {
-                ParquetTableLocationKey locationKey = cache.get(parquetFilePath);
-                if (locationKey == null) {
-                    locationKey = locationKey(parquetFilePath, readInstructions);
-                    if (!locationKey.verifyFileReader()) {
-                        continue;
+    public void findKeys(@NotNull final Consumer<ParquetTableLocationKey> locationKeyObserver) {
+        final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
+                tableRootDirectory, readInstructions.getSpecialInstructions());
+        try {
+            final List<URI> parquetURIs = provider.getChildURIListFromDirectory(tableRootDirectory,
+                    ParquetFileHelper::fileNameMatches);
+            synchronized (this) {
+                // Iterate over the URI stream and add the location keys to the cache
+                parquetURIs.forEach(parquetFileURI -> {
+                    ParquetTableLocationKey locationKey = cache.get(parquetFileURI);
+                    if (locationKey == null) {
+                        locationKey = locationKey(parquetFileURI, readInstructions);
+                        if (!locationKey.verifyFileReader()) {
+                            return;
+                        }
+                        cache.put(parquetFileURI, locationKey);
                     }
-                    cache.put(parquetFilePath, locationKey);
-                }
-                locationKeyObserver.accept(locationKey);
+                    locationKeyObserver.accept(locationKey);
+                });
             }
         } catch (final IOException e) {
             throw new TableDataException("Error finding parquet locations under " + tableRootDirectory, e);

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/layout/ParquetSingleFileLayout.java

@@ -15,25 +15,25 @@
  * Parquet {@link TableLocationKeyFinder location finder} that will discover a single file.
  */
 public final class ParquetSingleFileLayout implements TableLocationKeyFinder<ParquetTableLocationKey> {
-    private final URI parquetFileUri;
+    private final URI parquetFileURI;
     private final ParquetInstructions readInstructions;
 
     /**
-     * @param parquetFileUri URI of single parquet file to find
+     * @param parquetFileURI URI of single parquet file to find
      * @param readInstructions the instructions for customizations while reading
      */
-    public ParquetSingleFileLayout(@NotNull final URI parquetFileUri,
+    public ParquetSingleFileLayout(@NotNull final URI parquetFileURI,
             @NotNull final ParquetInstructions readInstructions) {
-        this.parquetFileUri = parquetFileUri;
+        this.parquetFileURI = parquetFileURI;
         this.readInstructions = readInstructions;
     }
 
     public String toString() {
-        return ParquetSingleFileLayout.class.getSimpleName() + '[' + parquetFileUri + ']';
+        return ParquetSingleFileLayout.class.getSimpleName() + '[' + parquetFileURI + ']';
     }
 
     @Override
     public void findKeys(@NotNull final Consumer<ParquetTableLocationKey> locationKeyObserver) {
-        locationKeyObserver.accept(new ParquetTableLocationKey(parquetFileUri, 0, null, readInstructions));
+        locationKeyObserver.accept(new ParquetTableLocationKey(parquetFileURI, 0, null, readInstructions));
     }
 }

extensions/parquet/table/src/test/java/io/deephaven/parquet/table/ParquetTableReadWriteTest.java

@@ -658,6 +658,25 @@ public void readSampleParquetFilesFromS3Test2() {
                 readInstructions, tableDefinition).select();
     }
 
+    @Test
+    public void readFlatPartitionedParquetFromS3() {
+        Assume.assumeTrue("Skipping test because s3 testing disabled.", ENABLE_S3_TESTING);
+        final S3Instructions s3Instructions = S3Instructions.builder()
+                .regionName("us-east-1")
+                .readAheadCount(1)
+                .fragmentSize(5 * 1024 * 1024)
+                .maxConcurrentRequests(50)
+                .maxCacheSize(32)
+                .readTimeout(Duration.ofSeconds(60))
+                .credentials(Credentials.defaultCredentials())
+                .build();
+        final ParquetInstructions readInstructions = new ParquetInstructions.Builder()
+                .setSpecialInstructions(s3Instructions)
+                .build();
+        ParquetTools.readFlatPartitionedTable("s3://dh-s3-parquet-test1/flatPartitionedParquet/",
+                readInstructions).select();
+    }
+
     @Test
     public void stringDictionaryTest() {
         final int nullPos = -5;

extensions/s3/build.gradle

@@ -15,7 +15,6 @@ dependencies {
     implementation platform('software.amazon.awssdk:bom:2.23.19')
     implementation 'software.amazon.awssdk:s3'
     implementation 'software.amazon.awssdk:aws-crt-client'
-    //implementation 'software.amazon.awssdk:netty-nio-client'
 
     compileOnly depAnnotations