feat: Added support to read parquet metadata files from S3 (#5777)

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

@@ -87,10 +87,14 @@ public boolean isCompatibleWith(@NotNull final SeekableChannelContext channelCon
  return wrappedProvider.isCompatibleWith(channelContext);
  }
 
+ @Override
+ public boolean exists(@NotNull final URI uri) {
+ return wrappedProvider.exists(uri);
+ }
+
  @Override
  public SeekableByteChannel getReadChannel(@NotNull final SeekableChannelContext channelContext,
- @NotNull final URI uri)
- throws IOException {
+ @NotNull final URI uri) throws IOException {
  final String uriString = uri.toString();
  final KeyedObjectHashMap<String, PerPathPool> channelPool = channelPools.get(ChannelType.Read);
  final CachedChannel result = tryGetPooledChannel(uriString, channelPool);

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

@@ -33,6 +33,11 @@ public boolean isCompatibleWith(@Nullable final SeekableChannelContext channelCo
  return true;
  }
 
+ @Override
+ public boolean exists(@NotNull final URI uri) {
+ return Files.exists(Path.of(uri));
+ }
+
  @Override
  public SeekableByteChannel getReadChannel(@Nullable final SeekableChannelContext channelContext,
  @NotNull final URI uri)

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

@@ -58,6 +58,14 @@ default SeekableChannelContext makeSingleUseContext() {
  */
  boolean isCompatibleWith(@NotNull SeekableChannelContext channelContext);
 
+ /**
+ * Returns true if the given URI exists in the underlying storage.
+ *
+ * @param uri the URI to check
+ * @return true if the URI exists
+ */
+ boolean exists(@NotNull URI uri);
+
  default SeekableByteChannel getReadChannel(@NotNull SeekableChannelContext channelContext, @NotNull String uriStr)
  throws IOException {
  return getReadChannel(channelContext, convertToURI(uriStr, false));

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

@@ -208,6 +208,11 @@ public boolean isCompatibleWith(@NotNull SeekableChannelContext channelContext)
  return channelContext == SeekableChannelContext.NULL;
  }
 
+ @Override
+ public boolean exists(@NotNull URI uri) {
+ throw new UnsupportedOperationException("exists");
+ }
+
  @Override
  public SeekableByteChannel getReadChannel(@NotNull SeekableChannelContext channelContext,
  @NotNull String path) {

engine/table/src/test/java/io/deephaven/engine/table/impl/QueryTableAggregationTest.java

@@ -3976,7 +3976,7 @@ public void testMultiPartitionSymbolTableBy() throws IOException {
  t4.updateView("Date=`2021-07-21`", "Num=400")).moveColumnsUp("Date", "Num");
 
  final Table loaded = ParquetTools.readTable(
- new ParquetKeyValuePartitionedLayout(testRootFile.toURI(), 2, ParquetInstructions.EMPTY),
+ ParquetKeyValuePartitionedLayout.create(testRootFile.toURI(), 2, ParquetInstructions.EMPTY, null),
  ParquetInstructions.EMPTY);
 
  // verify the sources are identical

extensions/parquet/base/src/main/java/io/deephaven/parquet/base/ColumnChunkReaderImpl.java

@@ -10,7 +10,6 @@
 import io.deephaven.parquet.compress.DeephavenCompressorAdapterFactory;
 import io.deephaven.util.channel.SeekableChannelContext.ContextHolder;
 import io.deephaven.util.datastructures.SoftCachingFunction;
-import org.apache.commons.io.FilenameUtils;
 import org.apache.parquet.bytes.BytesInput;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.column.Dictionary;
@@ -28,14 +27,12 @@
 import java.io.UncheckedIOException;
 import java.net.URI;
 import java.nio.channels.SeekableByteChannel;
-import java.nio.file.Path;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.function.Function;
 
-import static io.deephaven.base.FileUtils.convertToURI;
+import static io.deephaven.parquet.base.ParquetUtils.resolve;
 import static io.deephaven.parquet.base.ColumnPageReaderImpl.getDecompressorHolder;
-import static io.deephaven.parquet.base.ParquetFileReader.FILE_URI_SCHEME;
 import static org.apache.parquet.format.Encoding.PLAIN_DICTIONARY;
 import static org.apache.parquet.format.Encoding.RLE_DICTIONARY;
 
@@ -83,12 +80,10 @@ final class ColumnChunkReaderImpl implements ColumnChunkReader {
  this.dictionarySupplier = new SoftCachingFunction<>(this::getDictionary);
  this.numRows = numRows;
  this.version = version;
- if (columnChunk.isSetFile_path() && FILE_URI_SCHEME.equals(rootURI.getScheme())) {
- final String relativePath = FilenameUtils.separatorsToSystem(columnChunk.getFile_path());
- this.columnChunkURI = convertToURI(Path.of(rootURI).resolve(relativePath), false);
+ if (columnChunk.isSetFile_path()) {
+ columnChunkURI = resolve(rootURI, columnChunk.getFile_path());
  } else {
- // TODO(deephaven-core#5066): Add support for reading metadata files from non-file URIs
- this.columnChunkURI = rootURI;
+ columnChunkURI = rootURI;
  }
  // Construct the reader object but don't read the offset index yet
  this.offsetIndexReader = (columnChunk.isSetOffset_index_offset())

extensions/parquet/base/src/main/java/io/deephaven/parquet/base/ParquetUtils.java

@@ -3,26 +3,34 @@
 //
 package io.deephaven.parquet.base;
 
+import io.deephaven.UncheckedDeephavenException;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.File;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 
+import static io.deephaven.base.FileUtils.URI_SEPARATOR;
 import static io.deephaven.base.FileUtils.URI_SEPARATOR_CHAR;
 
 public final class ParquetUtils {
 
+ public static final String PARQUET_FILE_EXTENSION = ".parquet";
+
  public static final String METADATA_FILE_NAME = "_metadata";
  public static final String COMMON_METADATA_FILE_NAME = "_common_metadata";
- public static final String PARQUET_FILE_EXTENSION = ".parquet";
  public static final String METADATA_FILE_URI_SUFFIX = URI_SEPARATOR_CHAR + METADATA_FILE_NAME;
  public static final String COMMON_METADATA_FILE_URI_SUFFIX = URI_SEPARATOR_CHAR + COMMON_METADATA_FILE_NAME;
- public static final String METADATA_FILE_SUFFIX = File.separatorChar + METADATA_FILE_NAME;
- public static final String COMMON_METADATA_FILE_SUFFIX = File.separatorChar + COMMON_METADATA_FILE_NAME;
+ private static final String METADATA_FILE_SUFFIX = File.separatorChar + METADATA_FILE_NAME;
+ private static final String COMMON_METADATA_FILE_SUFFIX = File.separatorChar + COMMON_METADATA_FILE_NAME;
+
  private static final String MAGIC_STR = "PAR1";
  public static final byte[] MAGIC = MAGIC_STR.getBytes(StandardCharsets.US_ASCII);
 
+ private static final String WINDOWS_FILE_SEPARATOR = "\\";
+
  /**
  * The number of bytes to buffer before flushing while writing parquet files and metadata files.
  */
@@ -42,17 +50,23 @@ public static String getPerFileMetadataKey(final String filePath) {
  }
 
  /**
- * This method verifies if the source points to a parquet file or a metadata file. Provided source can be a local
- * file path or a URI. Also, it can point to a parquet file, metadata file or a directory.
+ * This method verifies if the source points to a parquet file. Provided source can be a local file path or a URI.
  */
  public static boolean isParquetFile(@NotNull final String source) {
- boolean ret = source.endsWith(PARQUET_FILE_EXTENSION)
- || source.endsWith(METADATA_FILE_URI_SUFFIX)
- || source.endsWith(COMMON_METADATA_FILE_URI_SUFFIX);
+ return source.endsWith(PARQUET_FILE_EXTENSION);
+ }
+
+ /**
+ * This method verifies if the source points to a metadata file. Provided source can be a local file path or a URI.
+ */
+ public static boolean isMetadataFile(@NotNull final String source) {
+ if (source.endsWith(METADATA_FILE_URI_SUFFIX) || source.endsWith(COMMON_METADATA_FILE_URI_SUFFIX)) {
+ return true;
+ }
  if (File.separatorChar != URI_SEPARATOR_CHAR) {
- ret = ret || source.endsWith(METADATA_FILE_SUFFIX) || source.endsWith(COMMON_METADATA_FILE_SUFFIX);
+ return source.endsWith(METADATA_FILE_SUFFIX) || source.endsWith(COMMON_METADATA_FILE_SUFFIX);
  }
- return ret;
+ return false;
  }
 
  /**
@@ -74,4 +88,18 @@ public static boolean isVisibleParquetFile(@NotNull final Path rootDir, @NotNull
  }
  return true;
  }
+
+ /**
+ * Resolve a relative path against a base URI. The path can be from Windows or Unix systems.
+ */
+ public static URI resolve(final URI base, final String relativePath) {
+ final URI relativeURI;
+ try {
+ // Sanitize the relative path before resolving it to avoid issues with separators and special characters
+ relativeURI = new URI(null, null, relativePath.replace(WINDOWS_FILE_SEPARATOR, URI_SEPARATOR), null);
+ } catch (final URISyntaxException e) {
+ throw new UncheckedDeephavenException("Failed to create URI from relative path: " + relativePath, e);
+ }
+ return base.resolve(relativeURI);
+ }
 }