Working state, can be optimized further

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

@@ -111,9 +111,9 @@ public SeekableByteChannel getWriteChannel(@NotNull final Path path, final boole
     }
 
     @Override
-    public List<URI> getURIStreamFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
+    public List<URI> getChildURIListFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
             throws IOException {
-        return wrappedProvider.getURIStreamFromDirectory(directoryURI, uriFilter);
+        return wrappedProvider.getChildURIListFromDirectory(directoryURI, uriFilter);
     }
 
     @Nullable

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

@@ -116,7 +116,7 @@ default SeekableByteChannel getWriteChannel(@NotNull final String path, final bo
      * @param directoryURI The URI of the directory to list
      * @param uriFilter A filter to apply to the URIs in the directory
      */
-    default List<URI> getURIStreamFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
+    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

@@ -982,8 +982,7 @@ public static Table readFlatPartitionedTable(
             @NotNull final ParquetInstructions readInstructions,
             @NotNull final TableDefinition tableDefinition) {
         return readPartitionedTable(new ParquetFlatPartitionedLayout(convertToURI(directory), readInstructions),
-                readInstructions,
-                tableDefinition);
+                readInstructions, tableDefinition);
     }
 
     /**

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

@@ -61,7 +61,7 @@ public void findKeys(@NotNull final Consumer<ParquetTableLocationKey> locationKe
         final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
                 tableRootDirectory, readInstructions.getSpecialInstructions());
         try {
-            final List<URI> parquetURIs = provider.getURIStreamFromDirectory(tableRootDirectory,
+            final List<URI> parquetURIs = provider.getChildURIListFromDirectory(tableRootDirectory,
                     ParquetFileHelper::fileNameMatches);
             synchronized (this) {
                 // Iterate over the URI stream and add the location keys to the cache

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

@@ -2,6 +2,7 @@
 
 import io.deephaven.base.reference.PooledObjectReference;
 import io.deephaven.util.channel.SeekableChannelContext;
+import org.jetbrains.annotations.Nullable;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.slf4j.Logger;
@@ -28,20 +29,26 @@
 import java.util.function.BiConsumer;
 
 /**
- * Context object used to store read-ahead buffers for efficiently reading from S3.
+ * Context object used to store read-ahead buffers for efficiently reading from S3. A single context object can only be
+ * associated with a single URI at a time.
  */
 final class S3ChannelContext implements SeekableChannelContext {
     private static final Logger log = LoggerFactory.getLogger(S3ChannelContext.class);
-    private static final long UNINITIALIZED_SIZE = -1;
+    static final long UNINITIALIZED_SIZE = -1;
+    private static final long UNINITIALIZED_NUM_FRAGMENTS = -1;
 
     private final S3AsyncClient client;
-    final S3Instructions instructions;
+    private final S3Instructions instructions;
     private final BufferPool bufferPool;
 
+    /**
+     * The URI associated with this context. A single context object can only be associated with a single URI at a time.
+     * But it can be re-associated with a different URI after closing.
+     */
     private S3Uri uri;
 
     /**
-     * Used to cache recently fetched fragments for faster lookup
+     * Used to cache recently fetched fragments from the {@link #uri} for faster lookup.
      */
     private final Request[] requests;
 
@@ -60,13 +67,19 @@ final class S3ChannelContext implements SeekableChannelContext {
         this.instructions = Objects.requireNonNull(instructions);
         this.bufferPool = Objects.requireNonNull(bufferPool);
         requests = new Request[instructions.maxCacheSize()];
+        uri = null;
         size = UNINITIALIZED_SIZE;
+        numFragments = UNINITIALIZED_NUM_FRAGMENTS;
         if (log.isDebugEnabled()) {
             log.debug("creating context: {}", ctxStr());
         }
     }
 
-    void verifyOrSetUri(S3Uri uri) {
+    S3Uri getUri() {
+        return uri;
+    }
+
+    void verifyOrSetUri(@Nullable final S3Uri uri) {
         if (this.uri == null) {
             this.uri = Objects.requireNonNull(uri);
         } else if (!this.uri.equals(uri)) {
@@ -84,7 +97,7 @@ void verifyOrSetSize(long size) {
         }
     }
 
-    public long size() throws IOException {
+    long size() throws IOException {
         ensureSize();
         return size;
     }
@@ -137,6 +150,10 @@ public void close() {
                 requests[i] = null;
             }
         }
+        // Reset the internal state
+        uri = null;
+        size = UNINITIALIZED_SIZE;
+        numFragments = UNINITIALIZED_NUM_FRAGMENTS;
     }
 
     // --------------------------------------------------------------------------------------------------

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

@@ -17,15 +17,17 @@
 import java.nio.channels.SeekableByteChannel;
 import java.util.Objects;
 
+import static io.deephaven.extensions.s3.S3ChannelContext.UNINITIALIZED_SIZE;
+
 
 /**
  * {@link SeekableByteChannel} class used to fetch objects from S3 buckets using an async client with the ability to
  * read ahead and cache fragments of the object.
  */
 final class S3SeekableByteChannel implements SeekableByteChannel, CachedChannelProvider.ContextHolder {
 
-    private static final long UNINITIALIZED_SIZE = -1;
     private static final long CLOSED_SENTINEL = -1;
+    private static final int INIT_POSITION = 0;
 
     private final S3Uri uri;
 
@@ -41,6 +43,7 @@ final class S3SeekableByteChannel implements SeekableByteChannel, CachedChannelP
     S3SeekableByteChannel(S3Uri uri) {
         this.uri = Objects.requireNonNull(uri);
         this.size = UNINITIALIZED_SIZE;
+        this.position = INIT_POSITION;
     }
 
     /**
@@ -56,7 +59,10 @@ public void setContext(@Nullable final SeekableChannelContext channelContext) {
         }
         this.context = (S3ChannelContext) channelContext;
         if (this.context != null) {
-            this.context.verifyOrSetUri(uri);
+            if (this.context.getUri() != uri) {
+                this.context.close();
+                this.context.verifyOrSetUri(uri);
+            }
             if (size != UNINITIALIZED_SIZE) {
                 context.verifyOrSetSize(size);
             }

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

@@ -112,7 +112,7 @@ public SeekableByteChannel getWriteChannel(@NotNull final Path path, final boole
     }
 
     @Override
-    public List<URI> getURIStreamFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
+    public List<URI> getChildURIListFromDirectory(@NotNull URI directoryURI, @NotNull Predicate<URI> uriFilter)
             throws IOException {
         final S3Uri s3DirectoryURI = s3AsyncClient.utilities().parseUri(directoryURI);
         final String bucketName = s3DirectoryURI.bucket().orElseThrow();

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

@@ -77,7 +77,7 @@ public SeekableByteChannel getWriteChannel(@NotNull final Path filePath, final b
     }
 
     @Override
-    public List<URI> getURIStreamFromDirectory(@NotNull final URI directoryURI,
+    public List<URI> getChildURIListFromDirectory(@NotNull final URI directoryURI,
             @NotNull final Predicate<URI> uriFilter) throws IOException {
         try (final Stream<Path> childFileStream = Files.list(Path.of(directoryURI))) {
             return childFileStream

py/server/deephaven/parquet.py

@@ -181,7 +181,7 @@ def read(
             if file_layout == ParquetFileLayout.SINGLE_FILE:
                 j_table = _JParquetTools.readSingleFileTable(path, read_instructions, j_table_definition)
             elif file_layout == ParquetFileLayout.FLAT_PARTITIONED:
-                j_table = _JParquetTools.readFlatPartitionedTable(_JFile(path), read_instructions, j_table_definition)
+                j_table = _JParquetTools.readFlatPartitionedTable(path, read_instructions, j_table_definition)
             elif file_layout == ParquetFileLayout.KV_PARTITIONED:
                 j_table = _JParquetTools.readKeyValuePartitionedTable(_JFile(path), read_instructions, j_table_definition)
             elif file_layout == ParquetFileLayout.METADATA_PARTITIONED:
@@ -194,7 +194,7 @@ def read(
             elif file_layout == ParquetFileLayout.SINGLE_FILE:
                 j_table = _JParquetTools.readSingleFileTable(path, read_instructions)
             elif file_layout == ParquetFileLayout.FLAT_PARTITIONED:
-                j_table = _JParquetTools.readFlatPartitionedTable(_JFile(path), read_instructions)
+                j_table = _JParquetTools.readFlatPartitionedTable(path, read_instructions)
             elif file_layout == ParquetFileLayout.KV_PARTITIONED:
                 j_table = _JParquetTools.readKeyValuePartitionedTable(_JFile(path), read_instructions)
             elif file_layout == ParquetFileLayout.METADATA_PARTITIONED: