Added support to read partitioned parquet files from S3

Base/src/main/java/io/deephaven/base/FileUtils.java

@@ -264,9 +264,16 @@ public static URI convertToURI(final String source, final boolean isDirectory) {
         if (source.isEmpty()) {
             throw new IllegalArgumentException("Cannot convert empty source to URI");
         }
-        final URI uri;
+        URI uri;
         try {
             uri = new URI(source);
+            // Replace two or more consecutive slashes in the path with a single slash
+            final String path = uri.getPath();
+            if (path.contains("//")) {
+                final String canonicalizedPath = uri.getPath().replaceAll("//+", "/");
+                uri = new URI(uri.getScheme(), uri.getUserInfo(), uri.getHost(), uri.getPort(), canonicalizedPath,
+                        uri.getQuery(), uri.getFragment());
+            }
         } catch (final URISyntaxException e) {
             // If the URI is invalid, assume it's a file path
             return convertToURI(new File(source), isDirectory);

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

@@ -91,9 +91,8 @@ default SeekableByteChannel getWriteChannel(@NotNull final String path, final bo
      * Returns a stream of URIs, the elements of which are the entries in the directory. The listing is non-recursive.
      * 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.
+     * Also, the caller is responsible for closing the stream, preferably using a try-with-resources block.
      *
-     * @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
@@ -104,10 +103,7 @@ default SeekableByteChannel getWriteChannel(@NotNull final String path, final bo
      * Returns a stream of URIs, the elements of which are all the files in the file tree rooted at the given starting
      * directory. 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.
+     * processor. Also, the caller is responsible for closing the stream, preferably using a try-with-resources block.
      *
      * @param directory the URI of the directory to walk
      * @return The {@link Stream} of {@link URI}s

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

@@ -11,7 +11,6 @@
 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;
@@ -35,7 +34,7 @@ public abstract class URIStreamKeyValuePartitionLayout<TLK extends TableLocation
     private static final String URI_SEPARATOR = "/";
 
     protected final URI tableRootDirectory;
-    private final Supplier<KeyValuePartitionLayout.LocationTableBuilder> locationTableBuilderFactory;
+    private final Supplier<LocationTableBuilder> locationTableBuilderFactory;
     private final int maxPartitioningLevels;
 
     /**
@@ -47,7 +46,7 @@ public abstract class URIStreamKeyValuePartitionLayout<TLK extends TableLocation
      * @param maxPartitioningLevels Maximum partitioning levels to traverse. Must be {@code >= 0}. {@code 0} means only
      *        look at files in {@code tableRootDirectory} and find no partitions.
      */
-    public URIStreamKeyValuePartitionLayout(
+    protected URIStreamKeyValuePartitionLayout(
             @NotNull final URI tableRootDirectory,
             @NotNull final Supplier<LocationTableBuilder> locationTableBuilderFactory,
             @NotNull final BiFunction<URI, Map<String, Comparable<?>>, TLK> keyFactory,
@@ -78,10 +77,6 @@ protected final void findKeys(@NotNull final Stream<URI> uriStream,
             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.setTrue();
             }
@@ -99,34 +94,33 @@ private void getPartitions(@NotNull final URI relativePath,
             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(URI_SEPARATOR);
-        for (int sdi = 0; sdi < subDirs.length - 1; sdi++) {
-            final String dirName = subDirs[sdi];
-            if (dirName.isEmpty()) {
-                // Ignore empty directory names
-                continue;
+        final int numPartitioningCol = subDirs.length - 1;
+        if (!registered) {
+            if (numPartitioningCol > maxPartitioningLevels) {
+                throw new TableDataException("Too many partitioning levels at " + relativePathString + ", count = " +
+                        numPartitioningCol + ", maximum expected are " + maxPartitioningLevels);
+            }
+        } else {
+            if (numPartitioningCol > partitionKeys.size()) {
+                throw new TableDataException("Too many partitioning levels at " + relativePathString + " (expected "
+                        + partitionKeys.size() + ") based on earlier leaf nodes in the tree.");
             }
+        }
+        for (int partitioningColIndex = 0; partitioningColIndex < numPartitioningCol; partitioningColIndex++) {
+            final String dirName = subDirs[partitioningColIndex];
             final String[] components = dirName.split("=", 2);
             if (components.length != 2) {
                 throw new TableDataException("Unexpected directory name format (not key=value) at "
-                        + new File(tableRootDirectory.getPath(), relativePathString));
+                        + relativePathString);
             }
-            // We use an empty set to allow duplicate partition keys across files
             final String columnKey = NameValidator.legalizeColumnName(components[0], takenNames);
             takenNames.add(columnKey);
             if (registered) {
-                // 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 leaf nodes in the tree.");
-                }
-                if (!partitionKeys.get(partitioningColumnIndex).equals(columnKey)) {
+                if (!partitionKeys.get(partitioningColIndex).equals(columnKey)) {
                     throw new TableDataException(String.format(
                             "Column name mismatch at column index %d: expected %s found %s at %s",
-                            partitioningColumnIndex, partitionKeys.get(partitioningColumnIndex), columnKey,
+                            partitioningColIndex, partitionKeys.get(partitioningColIndex), columnKey,
                             relativePathString));
                 }
             } else {
@@ -135,7 +129,6 @@ private void getPartitions(@NotNull final URI relativePath,
             }
             final String columnValue = components[1];
             partitionValues.add(columnValue);
-            partitioningColumnIndex++;
         }
     }
 }

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

@@ -1244,7 +1244,7 @@ private static Table readTableInternal(
         }
         if (source.endsWith(METADATA_FILE_NAME) || source.endsWith(COMMON_METADATA_FILE_NAME)) {
             throw new UncheckedDeephavenException("We currently do not support reading parquet metadata files " +
-                    "from non local file systems");
+                    "from non local storage");
         }
         // Both flat partitioned and key-value partitioned data can be read under key-value partitioned layout
         return readPartitionedTable(new ParquetKeyValuePartitionedLayout(sourceURI, MAX_PARTITIONING_LEVELS_INFERENCE,

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

@@ -64,19 +64,18 @@ public String toString() {
 
     @Override
     public void findKeys(@NotNull final Consumer<ParquetTableLocationKey> locationKeyObserver) {
-        final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
+        final Predicate<URI> uriFilter;
+        if (FILE_URI_SCHEME.equals(tableRootDirectory.getScheme())) {
+            uriFilter = uri -> {
+                final String filename = new File(uri).getName();
+                return filename.endsWith(ParquetUtils.PARQUET_FILE_EXTENSION) && filename.charAt(0) != '.';
+            };
+        } else {
+            uriFilter = uri -> uri.getPath().endsWith(ParquetUtils.PARQUET_FILE_EXTENSION);
+        }
+        try (final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
                 tableRootDirectory, readInstructions.getSpecialInstructions());
-        final boolean isFileURI = FILE_URI_SCHEME.equals(tableRootDirectory.getScheme());
-        try (final Stream<URI> stream = provider.list(tableRootDirectory)) {
-            final Predicate<URI> uriFilter;
-            if (isFileURI) {
-                uriFilter = uri -> {
-                    final String filename = new File(uri).getName();
-                    return filename.endsWith(ParquetUtils.PARQUET_FILE_EXTENSION) && filename.charAt(0) != '.';
-                };
-            } else {
-                uriFilter = uri -> uri.getPath().endsWith(ParquetUtils.PARQUET_FILE_EXTENSION);
-            }
+                final Stream<URI> stream = provider.list(tableRootDirectory)) {
             stream.filter(uriFilter).forEach(uri -> {
                 synchronized (ParquetFlatPartitionedLayout.this) {
                     ParquetTableLocationKey locationKey = cache.get(uri);

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

@@ -11,6 +11,7 @@
 import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder;
 import io.deephaven.engine.table.impl.locations.local.LocationTableBuilderDefinition;
 import io.deephaven.engine.table.impl.locations.local.URIStreamKeyValuePartitionLayout;
+import io.deephaven.engine.table.impl.locations.local.KeyValuePartitionLayout;
 import io.deephaven.parquet.base.ParquetUtils;
 import io.deephaven.parquet.table.ParquetInstructions;
 import io.deephaven.parquet.table.location.ParquetTableLocationKey;
@@ -32,11 +33,11 @@
 import static io.deephaven.parquet.base.ParquetUtils.isVisibleParquetFile;
 
 /**
- * Key-Value partitioned layout for Parquet data.
+ * {@link KeyValuePartitionLayout} for Parquet data.
  * 
  * @implNote
  *           <ul>
- *           <li>Unless table definition is provided, type inference for partitioning column uses
+ *           <li>Unless a {@link TableDefinition} is provided, type inference for partitioning column uses
  *           {@link CsvTools#readCsv(java.io.InputStream) CsvTools.readCsv} as a conversion tool, and hence follows the
  *           same rules.</li>
  *           <li>Column names will be legalized via {@link NameValidator#legalizeColumnName(String, Set)
@@ -87,17 +88,16 @@ public ParquetKeyValuePartitionedLayout(
 
     @Override
     public final void findKeys(@NotNull final Consumer<ParquetTableLocationKey> locationKeyObserver) {
-        final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
+        final Predicate<URI> uriFilter;
+        if (FILE_URI_SCHEME.equals(tableRootDirectory.getScheme())) {
+            final Path rootDir = Path.of(tableRootDirectory);
+            uriFilter = uri -> isVisibleParquetFile(rootDir, Path.of(uri));
+        } else {
+            uriFilter = uri -> uri.getPath().endsWith(ParquetUtils.PARQUET_FILE_EXTENSION);
+        }
+        try (final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
                 tableRootDirectory, readInstructions.getSpecialInstructions());
-        final boolean isFileURI = FILE_URI_SCHEME.equals(tableRootDirectory.getScheme());
-        try (final Stream<URI> uriStream = provider.walk(tableRootDirectory)) {
-            final Predicate<URI> uriFilter;
-            if (isFileURI) {
-                final Path rootDir = Path.of(tableRootDirectory);
-                uriFilter = uri -> isVisibleParquetFile(rootDir, Path.of(uri));
-            } else {
-                uriFilter = uri -> uri.getPath().endsWith(ParquetUtils.PARQUET_FILE_EXTENSION);
-            }
+                final Stream<URI> uriStream = provider.walk(tableRootDirectory)) {
             final Stream<URI> filteredStream = uriStream.filter(uriFilter);
             findKeys(filteredStream, locationKeyObserver);
         } catch (final IOException e) {

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

@@ -1803,6 +1803,16 @@ public void basicWriteAndReadFromFileURITests() {
         } catch (final RuntimeException e) {
             assertTrue(e instanceof UnsupportedOperationException);
         }
+
+        // Read from absolute path with additional "/" in the path
+        final String additionalSlashPath = rootFile.getAbsolutePath() + "/////" + filename;
+        final Table fromDisk5 = ParquetTools.readTable(additionalSlashPath);
+        assertTableEquals(tableToSave, fromDisk5);
+
+        // Read from URI with additional "/" in the path
+        final String additionalSlashURI = "file:////" + additionalSlashPath;
+        final Table fromDisk6 = ParquetTools.readTable(additionalSlashURI);
+        assertTableEquals(tableToSave, fromDisk6);
     }
 
     /**

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

@@ -4,7 +4,6 @@
 package io.deephaven.extensions.s3;
 
 import com.google.auto.service.AutoService;
-import io.deephaven.base.verify.Require;
 import io.deephaven.util.channel.SeekableChannelsProvider;
 import io.deephaven.util.channel.SeekableChannelsProviderPlugin;
 import org.jetbrains.annotations.NotNull;
@@ -28,12 +27,11 @@ public boolean isCompatible(@NotNull final URI uri, @Nullable final Object confi
     @Override
     public SeekableChannelsProvider createProvider(@NotNull final URI uri, @Nullable final Object config) {
         if (!isCompatible(uri, config)) {
-            if (!(config instanceof S3Instructions)) {
-                throw new IllegalArgumentException("Must provide S3Instructions to read files from S3");
-            }
             throw new IllegalArgumentException("Arguments not compatible, provided uri " + uri);
         }
-        final S3Instructions s3Instructions = (S3Instructions) Require.neqNull(config, "config");
-        return new S3SeekableChannelProvider(s3Instructions);
+        if (!(config instanceof S3Instructions)) {
+            throw new IllegalArgumentException("Must provide S3Instructions to read files from S3");
+        }
+        return new S3SeekableChannelProvider((S3Instructions) config);
     }
 }

extensions/trackedfile/src/main/java/io/deephaven/extensions/trackedfile/TrackedSeekableChannelsProviderPlugin.java

@@ -28,11 +28,11 @@ public boolean isCompatible(@NotNull final URI uri, @Nullable final Object objec
     @Override
     public SeekableChannelsProvider createProvider(@NotNull final URI uri, @Nullable final Object object) {
         if (!isCompatible(uri, object)) {
-            if (object != null) {
-                throw new IllegalArgumentException("Arguments not compatible, provided non null object");
-            }
             throw new IllegalArgumentException("Arguments not compatible, provided uri " + uri);
         }
+        if (object != null) {
+            throw new IllegalArgumentException("Arguments not compatible, provided non null object");
+        }
         return new TrackedSeekableChannelsProvider(TrackedFileHandleFactory.getInstance());
     }
 }