New Physical IO Implementation

.github/workflows/gradle-integration-test.yml

@@ -20,12 +20,13 @@ env:
   S3_TEST_BUCKET : ${{ vars.S3_TEST_BUCKET }}
   S3_TEST_PREFIX : ${{ vars.S3_TEST_PREFIX }}
   ROLE_TO_ASSUME: ${{ secrets.S3_TEST_ASSUME_ROLE_ARN }}
+  CUSTOMER_KEY: ${{ secrets.CUSTOMER_KEY }}
 
 jobs:
   build:
     name: Integration Tests
     runs-on: codebuild-s3-analytics-accelerator-eu-west-1-${{ github.run_id }}-${{ github.run_attempt }}
-    environment: integration-tests
+    environment: ${{ github.event_name == 'pull_request_target' && 'integration-tests' || '' }}
     permissions:
       contents: read
       id-token: write

README.md

@@ -71,6 +71,21 @@ When the `S3SeekableInputStreamFactory` is no longer required to create new stre
 s3SeekableInputStreamFactory.close();
 ```
 
+### Accessing SSE_C encrypted objects
+
+To access SSE_C encrypted objects using AAL, set the customer key which was used to encrypt the object in the ```OpenStreamInformation``` object and pass the openStreamInformation object in the stream. The customer key must be base64 encoded.
+
+```
+ OpenStreamInformation openStreamInformation =
+        OpenStreamInformation.builder()
+            .encryptionSecrets(
+                EncryptionSecrets.builder().sseCustomerKey(Optional.of(base64EncodedCustomerKey)).build())
+            .build();
+
+ S3SeekableInputStream s3SeekableInputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(bucket, key), openStreamInformation);
+
+```
+
 ### Using with Hadoop
 
 If you are using Analytics Accelerator Library for Amazon S3 with Hadoop, you need to set the stream type to `analytics` in the Hadoop configuration. An example configuration is as follows:

buildSrc/src/main/kotlin/buildlogic.java-common-conventions.gradle.kts

@@ -109,6 +109,7 @@ tasks.named<Test>("test") {
 }
 
 tasks.test {
+    maxHeapSize = "2G"
     // Report is generated and verification is run after tests
     finalizedBy(tasks.jacocoTestReport, tasks.jacocoTestCoverageVerification)
 }

common/src/main/java/software/amazon/s3/analyticsaccelerator/request/EncryptionSecrets.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package software.amazon.s3.analyticsaccelerator.request;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+import java.util.Optional;
+import lombok.Builder;
+import lombok.Getter;
+
+/**
+ * Contains encryption secrets for Server-Side Encryption with Customer-Provided Keys (SSE-C). This
+ * class manages the customer-provided encryption key used for SSE-C operations with Amazon S3.
+ */
+@Getter
+public class EncryptionSecrets {
+
+  /**
+   * The customer-provided encryption key for SSE-C operations. When present, this key will be used
+   * for server-side encryption. The key must be Base64 encoded and exactly 256 bits (32 bytes) when
+   * decoded.
+   */
+  private final Optional<String> ssecCustomerKey;
+
+  /**
+   * The Base64-encoded MD5 hash of the customer key. This hash is automatically calculated from the
+   * customer key and is used by Amazon S3 to verify the integrity of the encryption key during
+   * transmission. Will be null if no customer key is provided.
+   */
+  private final String ssecCustomerKeyMd5;
+
+  /**
+   * Constructs an EncryptionSecrets instance with the specified SSE-C customer key.
+   *
+   * <p>This constructor processes the SSE-C (Server-Side Encryption with Customer-Provided Keys)
+   * encryption key and calculates its MD5 hash as required by Amazon S3. The process involves:
+   *
+   * <ol>
+   *   <li>Accepting a Base64-encoded encryption key
+   *   <li>Decoding the Base64 key back to bytes
+   *   <li>Computing the MD5 hash of these bytes
+   *   <li>Encoding the MD5 hash in Base64 format
+   * </ol>
+   *
+   * @param sseCustomerKey An Optional containing the Base64-encoded encryption key, or empty if no
+   *     encryption is needed
+   */
+  @Builder
+  public EncryptionSecrets(Optional<String> sseCustomerKey) {
+    this.ssecCustomerKey = sseCustomerKey;
+    this.ssecCustomerKeyMd5 =
+        sseCustomerKey
+            .map(
+                key -> {
+                  try {
+                    MessageDigest md = MessageDigest.getInstance("MD5");
+                    return Base64.getEncoder()
+                        .encodeToString(md.digest(Base64.getDecoder().decode(key)));
+                  } catch (NoSuchAlgorithmException e) {
+                    throw new IllegalStateException("MD5 algorithm not available", e);
+                  }
+                })
+            .orElse(null);
+  }
+}

common/src/main/java/software/amazon/s3/analyticsaccelerator/request/ReadMode.java

@@ -15,12 +15,35 @@
  */
 package software.amazon.s3.analyticsaccelerator.request;
 
+import lombok.AllArgsConstructor;
+
 /**
  * Enum to help with the annotation of reads. We mark reads SYNC when they were triggered by a
  * synchronous read or ASYNC when they were to do logical or physical prefetching.
  */
+@AllArgsConstructor
 public enum ReadMode {
-  SYNC,
-  ASYNC,
-  SMALL_OBJECT_PREFETCH;
+  SYNC(true),
+  ASYNC(true),
+  SMALL_OBJECT_PREFETCH(true),
+  SEQUENTIAL_FILE_PREFETCH(true),
+  DICTIONARY_PREFETCH(false),
+  COLUMN_PREFETCH(false),
+  REMAINING_COLUMN_PREFETCH(false),
+  PREFETCH_TAIL(false),
+  READ_VECTORED(false);
+
+  private final boolean allowRequestExtension;
+
+  /**
+   * Should requests be extended for this read mode?
+   *
+   * <p>When the read is from the parquet prefetcher or readVectored(), we know the exact ranges we
+   * want to read, so in this case don't extend the ranges.
+   *
+   * @return true if requests should be extended
+   */
+  public boolean allowRequestExtension() {
+    return allowRequestExtension;
+  }
 }

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/MetricKey.java

@@ -35,7 +35,13 @@ public enum MetricKey {
   /**
    * Tracks the number of cache misses. Incremented when requested block is not found in the cache
    */
-  CACHE_MISS("CacheMiss");
+  CACHE_MISS("CacheMiss"),
+
+  /** Counts number of GET requests made. */
+  GET_REQUEST_COUNT("GetRequestCount"),
+
+  /** Counts number of HEAD requests made. */
+  HEAD_REQUEST_COUNT("HeadRequestCount");
 
   /** The string name representation of the metric. */
   private final String name;

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java

@@ -18,6 +18,7 @@
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Getter;
+import software.amazon.s3.analyticsaccelerator.request.EncryptionSecrets;
 import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
 import software.amazon.s3.analyticsaccelerator.request.StreamAuditContext;
 
@@ -41,6 +42,7 @@ public class OpenStreamInformation {
   private final StreamAuditContext streamAuditContext;
   private final ObjectMetadata objectMetadata;
   private final InputPolicy inputPolicy;
+  private final EncryptionSecrets encryptionSecrets;
 
   /** Default set of settings for {@link OpenStreamInformation} */
   public static final OpenStreamInformation DEFAULT = OpenStreamInformation.builder().build();

common/src/test/java/software/amazon/s3/analyticsaccelerator/common/MetricKeyTest.java

@@ -27,14 +27,18 @@ public void testMetricKeyNames() {
     assertEquals("MemoryUsage", MetricKey.MEMORY_USAGE.getName());
     assertEquals("CacheHit", MetricKey.CACHE_HIT.getName());
     assertEquals("CacheMiss", MetricKey.CACHE_MISS.getName());
+    assertEquals("GetRequestCount", MetricKey.GET_REQUEST_COUNT.getName());
+    assertEquals("HeadRequestCount", MetricKey.HEAD_REQUEST_COUNT.getName());
   }
 
   @Test
   public void testEnumValues() {
     MetricKey[] values = MetricKey.values();
-    assertEquals(3, values.length);
+    assertEquals(5, values.length);
     assertEquals(MetricKey.MEMORY_USAGE, values[0]);
     assertEquals(MetricKey.CACHE_HIT, values[1]);
     assertEquals(MetricKey.CACHE_MISS, values[2]);
+    assertEquals(MetricKey.GET_REQUEST_COUNT, values[3]);
+    assertEquals(MetricKey.HEAD_REQUEST_COUNT, values[4]);
   }
 }

common/src/test/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformationTest.java

@@ -17,13 +17,25 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Optional;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
+import software.amazon.s3.analyticsaccelerator.request.EncryptionSecrets;
 import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
 import software.amazon.s3.analyticsaccelerator.request.StreamAuditContext;
 
 public class OpenStreamInformationTest {
 
+  private static final String CUSTOMER_KEY = "32-bytes-long-key-for-testing-123";
+
+  /**
+   * To generate the base64 encoded md5 value for a customer key use the cli command echo -n
+   * "customer_key" | base64 | base64 -d | openssl md5 -binary | base64
+   */
+  private static final String EXPECTED_BASE64_MD5 = "R+k8pqEVUmkxDfaH5MqIdw==";
+
   @Test
   public void testDefaultInstance() {
     OpenStreamInformation info = OpenStreamInformation.DEFAULT;
@@ -32,24 +44,36 @@ public void testDefaultInstance() {
     assertNull(info.getStreamAuditContext(), "Default streamContext should be null");
     assertNull(info.getObjectMetadata(), "Default objectMetadata should be null");
     assertNull(info.getInputPolicy(), "Default inputPolicy should be null");
+    assertNull(info.getEncryptionSecrets(), "Default encryptionSecrets should be null");
   }
 
   @Test
   public void testBuilderWithAllFields() {
     StreamAuditContext mockContext = Mockito.mock(StreamAuditContext.class);
     ObjectMetadata mockMetadata = Mockito.mock(ObjectMetadata.class);
     InputPolicy mockPolicy = Mockito.mock(InputPolicy.class);
+    String base64Key =
+        Base64.getEncoder().encodeToString(CUSTOMER_KEY.getBytes(StandardCharsets.UTF_8));
+    EncryptionSecrets secrets =
+        EncryptionSecrets.builder().sseCustomerKey(Optional.of(base64Key)).build();
 
     OpenStreamInformation info =
         OpenStreamInformation.builder()
             .streamAuditContext(mockContext)
             .objectMetadata(mockMetadata)
             .inputPolicy(mockPolicy)
+            .encryptionSecrets(secrets)
             .build();
 
     assertSame(mockContext, info.getStreamAuditContext(), "StreamContext should match");
     assertSame(mockMetadata, info.getObjectMetadata(), "ObjectMetadata should match");
     assertSame(mockPolicy, info.getInputPolicy(), "InputPolicy should match");
+    assertEquals(
+        base64Key,
+        info.getEncryptionSecrets().getSsecCustomerKey().get(),
+        "Customer key should match");
+    assertNotNull(info.getEncryptionSecrets().getSsecCustomerKeyMd5(), "MD5 should not be null");
+    assertEquals(EXPECTED_BASE64_MD5, info.getEncryptionSecrets().getSsecCustomerKeyMd5());
   }
 
   @Test
@@ -103,4 +127,47 @@ public void testNullFields() {
     assertNull(info.getObjectMetadata(), "ObjectMetadata should be null");
     assertNull(info.getInputPolicy(), "InputPolicy should be null");
   }
+
+  @Test
+  public void testDefaultInstanceEncryptionSecrets() {
+    OpenStreamInformation info = OpenStreamInformation.DEFAULT;
+    assertNull(info.getEncryptionSecrets(), "Default encryptionSecrets should be null");
+  }
+
+  @Test
+  public void testBuilderWithEncryptionSecrets() {
+    // Create a sample base64 encoded key
+    String base64Key =
+        Base64.getEncoder().encodeToString(CUSTOMER_KEY.getBytes(StandardCharsets.UTF_8));
+    EncryptionSecrets secrets =
+        EncryptionSecrets.builder().sseCustomerKey(Optional.of(base64Key)).build();
+
+    OpenStreamInformation info = OpenStreamInformation.builder().encryptionSecrets(secrets).build();
+
+    assertNotNull(info.getEncryptionSecrets(), "EncryptionSecrets should not be null");
+    assertTrue(
+        info.getEncryptionSecrets().getSsecCustomerKey().isPresent(),
+        "Customer key should be present");
+    assertEquals(
+        base64Key,
+        info.getEncryptionSecrets().getSsecCustomerKey().get(),
+        "Customer key should match");
+    assertNotNull(info.getEncryptionSecrets().getSsecCustomerKeyMd5(), "MD5 should not be null");
+    assertEquals(EXPECTED_BASE64_MD5, info.getEncryptionSecrets().getSsecCustomerKeyMd5());
+  }
+
+  @Test
+  public void testBuilderWithEmptyEncryptionSecrets() {
+    EncryptionSecrets secrets =
+        EncryptionSecrets.builder().sseCustomerKey(Optional.empty()).build();
+
+    OpenStreamInformation info = OpenStreamInformation.builder().encryptionSecrets(secrets).build();
+
+    assertNotNull(info.getEncryptionSecrets(), "EncryptionSecrets should not be null");
+    assertFalse(
+        info.getEncryptionSecrets().getSsecCustomerKey().isPresent(),
+        "Customer key should be empty");
+    assertNull(
+        info.getEncryptionSecrets().getSsecCustomerKeyMd5(), "MD5 should be null for empty key");
+  }
 }

input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ConcurrencyCorrectnessTest.java

@@ -25,7 +25,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 /**
- * This tests concurrency and thread safety of teh shared state. While the DAT InputStream itself is
+ * This tests concurrency and thread safety of teh shared state. While the AAL InputStream itself is
  * not thread-safe, the shared state that multiple streams access and manipulate should be.
  */
 public class ConcurrencyCorrectnessTest extends IntegrationTestBase {