Added support to read partitioned parquet files from S3

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

@@ -16,7 +16,6 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
-import java.util.function.Consumer;
 import java.util.stream.Stream;
 
 public class LocalFSChannelProvider implements SeekableChannelsProvider {

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

@@ -92,6 +92,9 @@ default SeekableByteChannel getWriteChannel(@NotNull final String path, final bo
      * Note that the URIs supplied by the stream will be file URIs (not ending with "/") irrespective of whether the URI
      * corresponds to a file or a directory. The caller should manage file vs. directory handling in the processor.
      *
+     * @apiNote This method must be used within a try-with-resources statement or similar control structure to ensure
+     *          that the stream's open resources are closed promptly after the stream's operations have completed.
+     *
      * @param directory the URI of the directory to list
      * @return The {@link Stream} of {@link URI}s
      */
@@ -103,6 +106,9 @@ default SeekableByteChannel getWriteChannel(@NotNull final String path, final bo
      * whether the URI corresponds to a file or a directory. The caller should manage file vs. directory handling in the
      * processor.
      *
+     * @apiNote This method must be used within a try-with-resources statement or similar control structure to ensure
+     *          that the stream's open resources are closed promptly after the stream's operations have completed.
+     *
      * @param directory the URI of the directory to walk
      * @return The {@link Stream} of {@link URI}s
      */

Util/channel/src/test/java/io/deephaven/util/channel/CachedChannelProviderTest.java

@@ -221,12 +221,12 @@ public SeekableByteChannel getWriteChannel(@NotNull Path path, boolean append) {
 
         @Override
         public final Stream<URI> list(@NotNull final URI directory) {
-            return Stream.empty();
+            throw new UnsupportedOperationException("list");
         }
 
         @Override
         public final Stream<URI> walk(@NotNull final URI directory) {
-            return Stream.empty();
+            throw new UnsupportedOperationException("walk");
         }
 
         @Override

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/local/FileKeyValuePartitionLayout.java

@@ -68,17 +68,19 @@ public FileKeyValuePartitionLayout(
         this.maxPartitioningLevels = Require.geqZero(maxPartitioningLevels, "maxPartitioningLevels");
     }
 
+    @Override
     public String toString() {
         return FileKeyValuePartitionLayout.class.getSimpleName() + '[' + tableRootDirectory + ']';
     }
 
     @Override
     public void findKeys(@NotNull final Consumer<TLK> locationKeyObserver) {
-        final Deque<Path> targetFiles = new ArrayDeque<>();
+        final Queue<Path> targetFiles = new ArrayDeque<>();
         final LocationTableBuilder locationTableBuilder = locationTableBuilderFactory.get();
         try {
             Files.walkFileTree(tableRootDirectory.toPath(), EnumSet.of(FileVisitOption.FOLLOW_LINKS),
                     maxPartitioningLevels + 1, new SimpleFileVisitor<>() {
+                        final Set<String> takenNames = new HashSet<>();
                         final List<String> partitionKeys = new ArrayList<>();
                         final List<String> partitionValues = new ArrayList<>();
                         boolean registered;
@@ -100,9 +102,8 @@ public FileVisitResult preVisitDirectory(
                                     throw new TableDataException(
                                             "Unexpected directory name format (not key=value) at " + dir);
                                 }
-                                // We use an empty set to allow duplicate partition keys across files
-                                final String columnKey =
-                                        NameValidator.legalizeColumnName(components[0], Collections.emptySet());
+                                final String columnKey = NameValidator.legalizeColumnName(components[0], takenNames);
+                                takenNames.add(columnKey);
                                 final int columnIndex = columnCount - 1;
                                 if (columnCount > partitionKeys.size()) {
                                     partitionKeys.add(columnKey);
@@ -129,6 +130,7 @@ public FileVisitResult visitFile(
                                 locationTableBuilder.acceptLocation(partitionValues);
                                 targetFiles.add(file);
                             }
+                            takenNames.clear();
                             return FileVisitResult.CONTINUE;
                         }
 

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/local/KeyValuePartitionLayout.java

@@ -16,8 +16,7 @@
 import java.util.function.Consumer;
 
 /**
- * Base class for {@link TableLocationKeyFinder location finders} that perform file system traversal to infer
- * partitions.
+ * Base class for {@link TableLocationKeyFinder location finders} that traverse file hierarchy to infer partitions.
  *
  * @param <TLK> The type of {@link TableLocationKey} to be generated
  * @param <TARGET_FILE_TYPE> The type of files used to generate location keys, like a {@link URI} or a {@link Path}
@@ -71,8 +70,9 @@ public KeyValuePartitionLayout(
         this.keyFactory = keyFactory;
     }
 
+    @Override
     public String toString() {
-        return KeyValuePartitionLayout.class.getSimpleName();
+        return getClass().getSimpleName();
     }
 
     /**
@@ -84,7 +84,7 @@ public String toString() {
      */
     final void buildLocationKeys(
             @NotNull final Table locationTable,
-            @NotNull final Deque<TARGET_FILE_TYPE> targetFiles,
+            @NotNull final Queue<TARGET_FILE_TYPE> targetFiles,
             @NotNull final Consumer<TLK> locationKeyObserver) {
         final Map<String, Comparable<?>> partitions = new LinkedHashMap<>();
         // Note that we allow the location table to define partition priority order.

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/local/URIStreamKeyValuePartitionLayout.java

@@ -8,18 +8,18 @@
 import io.deephaven.engine.table.Table;
 import io.deephaven.engine.table.impl.locations.TableDataException;
 import io.deephaven.engine.table.impl.locations.TableLocationKey;
+import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.File;
 import java.net.URI;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
-import java.util.Deque;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -32,10 +32,11 @@
 public abstract class URIStreamKeyValuePartitionLayout<TLK extends TableLocationKey>
         extends KeyValuePartitionLayout<TLK, URI> {
 
+    private static final String URI_SEPARATOR = "/";
+
     protected final URI tableRootDirectory;
     private final Supplier<KeyValuePartitionLayout.LocationTableBuilder> locationTableBuilderFactory;
     private final int maxPartitioningLevels;
-    private final String fileSeparator;
 
     /**
      * @param tableRootDirectory The directory to traverse from
@@ -48,15 +49,13 @@ public abstract class URIStreamKeyValuePartitionLayout<TLK extends TableLocation
      */
     public URIStreamKeyValuePartitionLayout(
             @NotNull final URI tableRootDirectory,
-            @NotNull final Supplier<KeyValuePartitionLayout.LocationTableBuilder> locationTableBuilderFactory,
+            @NotNull final Supplier<LocationTableBuilder> locationTableBuilderFactory,
             @NotNull final BiFunction<URI, Map<String, Comparable<?>>, TLK> keyFactory,
-            final int maxPartitioningLevels,
-            @NotNull final String fileSeparator) {
+            final int maxPartitioningLevels) {
         super(keyFactory);
         this.tableRootDirectory = tableRootDirectory;
         this.locationTableBuilderFactory = locationTableBuilderFactory;
         this.maxPartitioningLevels = Require.geqZero(maxPartitioningLevels, "maxPartitioningLevels");
-        this.fileSeparator = fileSeparator;
     }
 
     @Override
@@ -69,19 +68,22 @@ public String toString() {
      */
     protected final void findKeys(@NotNull final Stream<URI> uriStream,
             @NotNull final Consumer<TLK> locationKeyObserver) {
-        final KeyValuePartitionLayout.LocationTableBuilder locationTableBuilder = locationTableBuilderFactory.get();
-        final Deque<URI> targetURIs = new ArrayDeque<>();
-        final Set<String> takenNames = new HashSet<>();
+        final LocationTableBuilder locationTableBuilder = locationTableBuilderFactory.get();
+        final Queue<URI> targetURIs = new ArrayDeque<>();
         final List<String> partitionKeys = new ArrayList<>();
-        final boolean[] registered = {false}; // Hack to make the variable final
+        final MutableBoolean registered = new MutableBoolean(false);
         uriStream.forEachOrdered(uri -> {
             final Collection<String> partitionValues = new ArrayList<>();
             final URI relativePath = tableRootDirectory.relativize(uri);
-            getPartitions(relativePath, partitionKeys, partitionValues, takenNames, registered[0]);
-            if (!registered[0]) {
-                // Use the first path to find the partition keys and then use the same partition keys for the rest
+            getPartitions(relativePath, partitionKeys, partitionValues, registered.booleanValue());
+            if (registered.isFalse()) {
+                // Use the first path to find the partition keys and use the same for the rest
+                if (partitionKeys.size() > maxPartitioningLevels) {
+                    throw new TableDataException("Too many partitioning levels at " + uri + ", count = " +
+                            partitionKeys.size() + ", maximum expected are " + maxPartitioningLevels);
+                }
                 locationTableBuilder.registerPartitionKeys(partitionKeys);
-                registered[0] = true;
+                registered.setTrue();
             }
             // Use the partition values from each path to build the location table
             locationTableBuilder.acceptLocation(partitionValues);
@@ -94,14 +96,14 @@ protected final void findKeys(@NotNull final Stream<URI> uriStream,
     private void getPartitions(@NotNull final URI relativePath,
             @NotNull final List<String> partitionKeys,
             @NotNull final Collection<String> partitionValues,
-            @NotNull final Set<String> takenNames,
             final boolean registered) {
+        final Set<String> takenNames = new HashSet<>();
         final String relativePathString = relativePath.getPath();
         int partitioningColumnIndex = 0;
         // Split the path to get the subdirectory names
-        final String[] subDirs = relativePathString.split(fileSeparator);
-        for (int i = 0; i < subDirs.length - 1; i++) {
-            final String dirName = subDirs[i];
+        final String[] subDirs = relativePathString.split(URI_SEPARATOR);
+        for (int sdi = 0; sdi < subDirs.length - 1; sdi++) {
+            final String dirName = subDirs[sdi];
             if (dirName.isEmpty()) {
                 // Ignore empty directory names
                 continue;
@@ -111,18 +113,15 @@ private void getPartitions(@NotNull final URI relativePath,
                 throw new TableDataException("Unexpected directory name format (not key=value) at "
                         + new File(tableRootDirectory.getPath(), relativePathString));
             }
-            if (partitioningColumnIndex == maxPartitioningLevels) {
-                throw new TableDataException("Too many partitioning levels at " + relativePathString + ", maximum " +
-                        "expected partitioning levels are " + maxPartitioningLevels);
-            }
             // We use an empty set to allow duplicate partition keys across files
-            final String columnKey = NameValidator.legalizeColumnName(components[0], Collections.emptySet());
+            final String columnKey = NameValidator.legalizeColumnName(components[0], takenNames);
+            takenNames.add(columnKey);
             if (registered) {
-                // We have already seen another parquet file in the tree, so compare the
-                // partitioning levels against the previous ones
+                // We have already seen another leaf node in the tree, so compare the partitioning levels against the
+                // previous ones
                 if (partitioningColumnIndex >= partitionKeys.size()) {
                     throw new TableDataException("Too many partitioning levels at " + relativePathString + " (expected "
-                            + partitionKeys.size() + ") based on earlier parquet files in the tree.");
+                            + partitionKeys.size() + ") based on earlier leaf nodes in the tree.");
                 }
                 if (!partitionKeys.get(partitioningColumnIndex).equals(columnKey)) {
                     throw new TableDataException(String.format(
@@ -131,7 +130,7 @@ private void getPartitions(@NotNull final URI relativePath,
                             relativePathString));
                 }
             } else {
-                // This is the first parquet file in the tree, so accumulate the partitioning levels
+                // This is the first leaf node in the tree, so accumulate the partitioning levels
                 partitionKeys.add(columnKey);
             }
             final String columnValue = components[1];

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

@@ -22,7 +22,6 @@
 import java.io.IOException;
 import java.net.URI;
 import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
@@ -64,8 +63,7 @@ public ParquetKeyValuePartitionedLayout(
         super(tableRootDirectory,
                 () -> new LocationTableBuilderDefinition(tableDefinition),
                 (uri, partitions) -> new ParquetTableLocationKey(uri, 0, partitions, readInstructions),
-                Math.toIntExact(tableDefinition.getColumnStream().filter(ColumnDefinition::isPartitioning).count()),
-                FILE_URI_SCHEME.equals(tableRootDirectory.getScheme()) ? File.separator : "/");
+                Math.toIntExact(tableDefinition.getColumnStream().filter(ColumnDefinition::isPartitioning).count()));
         this.readInstructions = readInstructions;
     }
 
@@ -83,8 +81,7 @@ public ParquetKeyValuePartitionedLayout(
         super(tableRootDirectory,
                 () -> new LocationTableBuilderCsv(tableRootDirectory),
                 (uri, partitions) -> new ParquetTableLocationKey(uri, 0, partitions, readInstructions),
-                maxPartitioningLevels,
-                FILE_URI_SCHEME.equals(tableRootDirectory.getScheme()) ? File.separator : "/");
+                maxPartitioningLevels);
         this.readInstructions = readInstructions;
     }
 

extensions/s3/src/main/java/io/deephaven/extensions/s3/S3SeekableChannelProvider.java

@@ -134,7 +134,7 @@ public Stream<URI> list(@NotNull final URI directory) {
     @Override
     public Stream<URI> walk(@NotNull final URI directory) {
         if (log.isDebugEnabled()) {
-            log.debug("requesting list of child URIs for directory: {}", directory);
+            log.debug("requesting walking the tree rooted at directory: {}", directory);
         }
         return createStream(directory, true);
     }