Added a soft reference based shared cache for S3 reads

engine/table/src/main/java/io/deephaven/engine/table/impl/QueryTable.java

@@ -236,7 +236,7 @@ public interface MemoizableOperation<T extends DynamicNode & NotificationStepRec
             Configuration.getInstance().getIntegerWithDefault("QueryTable.parallelWhereSegments", -1);
 
     /**
-     * You can chose to enable or disable the column parallel select and update.
+     * You can choose to enable or disable the column parallel select and update.
      */
     static boolean ENABLE_PARALLEL_SELECT_AND_UPDATE =
             Configuration.getInstance().getBooleanWithDefault("QueryTable.enableParallelSelectAndUpdate", true);

extensions/parquet/compression/src/main/java/io/deephaven/parquet/compress/DeephavenCompressorAdapterFactory.java

@@ -69,7 +69,7 @@ private static DeephavenCompressorAdapterFactory createInstance() {
                 // Use the Parquet LZ4_RAW codec, which internally uses aircompressor
                 Lz4RawCodec.class, CompressionCodecName.LZ4_RAW,
 
-                // The rest of these are aircompressor codecs which have fast / pure java implementations
+                // The rest of these are aircompressor codecs that have fast / pure java implementations
                 JdkGzipCodec.class, CompressionCodecName.GZIP,
                 LzoCodec.class, CompressionCodecName.LZO,
                 Lz4Codec.class, CompressionCodecName.LZ4,

extensions/s3/build.gradle

@@ -13,6 +13,9 @@ dependencies {
     implementation project(':Configuration')
     implementation project(':log-factory')
 
+    // TODO Included this to get CleanupReferenceProcessorInstance, where should I move CleanupReferenceProcessorInstance?
+    implementation project(':engine-updategraph')
+
     implementation platform('software.amazon.awssdk:bom:2.23.19')
     implementation 'software.amazon.awssdk:s3'
     implementation 'software.amazon.awssdk:aws-crt-client'

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

@@ -0,0 +1,65 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.extensions.s3;
+
+import io.deephaven.hash.KeyedObjectHashMap;
+import io.deephaven.hash.KeyedObjectKey;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import software.amazon.awssdk.services.s3.S3Uri;
+import io.deephaven.extensions.s3.S3ChannelContext.Request;
+
+/**
+ * This class uses a ({@link KeyedObjectHashMap}) to cache {@link Request} objects based on their URI and fragment
+ * index.
+ */
+final class KeyedHashMapBasedRequestCache implements S3RequestCache {
+
+    private final KeyedObjectHashMap<Request.ID, Request> requests;
+
+    KeyedHashMapBasedRequestCache() {
+        requests = new KeyedObjectHashMap<>(new KeyedObjectKey.Basic<>() {
+            @Override
+            public Request.ID getKey(@NotNull final Request request) {
+                return request.getId();
+            }
+        });
+    }
+
+    @Override
+    @Nullable
+    public Request getRequest(@NotNull final S3Uri uri, final long fragmentIndex) {
+        final Request request = requests.get(new Request.ID(uri, fragmentIndex));
+        return request == null ? null : request.acquire();
+    }
+
+    @Override
+    @NotNull
+    public Request getOrCreateRequest(@NotNull final S3Uri uri, final long fragmentIndex,
+            @NotNull final S3ChannelContext context) {
+        // TODO Do you think the acquiring part should be done by the caller or here?
+        // I kept it here because acquire could potentially fail and the caller would have to call again in a loop,
+        // which
+        // felt unnecessary compared to a guaranteed acquire.
+        return requests.compute(new Request.ID(uri, fragmentIndex), (key, existingRequest) -> {
+            if (existingRequest != null) {
+                final Request acquired = existingRequest.acquire();
+                if (acquired != null) {
+                    return acquired;
+                }
+            }
+            final Request newRequest = Request.createAndAcquire(fragmentIndex, context);
+            // TODO Do you think the init part should be done by the context, inside createAndAcquire or here?
+            // Kept it here for now because caller doesn't know whether request is new or not. So should call init or
+            // not. Maybe we can make init more idempotent and the context would call it always.
+            newRequest.init();
+            return newRequest;
+        });
+    }
+
+    @Override
+    public void remove(@NotNull final Request request) {
+        requests.removeValue(request);
+    }
+}