Add s3.signer-type to native S3 filesystem

Author: anusudarsan

docs/src/main/sphinx/object-storage/file-system-s3.md

@@ -34,6 +34,10 @@ support:
   - S3 storage class to use while writing data. Defaults to `STANDARD`. Other allowed
     values are: `STANDARD_IA`, `INTELLIGENT_TIERING`, `REDUCED_REDUNDANCY`, `ONEZONE_IA`,
     `GLACIER`, `DEEP_ARCHIVE`, `OUTPOSTS`, `GLACIER_IR`, `SNOW`, `EXPRESS_ONEZONE`.
+* - `s3.signer-type`
+  - AWS signing protocol to use for authenticating S3 requests. Supported values are: 
+    `AwsS3V4Signer`, `Aws4Signer`, `AsyncAws4Signer`, `Aws4UnsignedPayloadSigner`, 
+    `EventStreamAws4Signer`.
 * - `s3.exclusive-create`
   - Whether conditional write is supported by the S3-compatible storage. Defaults to `true`.
 * - `s3.canned-acl`
@@ -386,13 +390,15 @@ the following edits to your catalog configuration:
    * - `hive.s3.path-style-access`
      - `s3.path-style-access`
      -
+   * - `hive.s3.signer-type`
+     - `s3.signer-type`
+     -
   :::
 
 1. Remove the following legacy configuration properties if they exist in your
    catalog configuration:
 
       * `hive.s3.storage-class`
-      * `hive.s3.signer-type`
       * `hive.s3.signer-class`
       * `hive.s3.staging-directory`
       * `hive.s3.pin-client-to-current-region`

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3FileSystemConfig.java

@@ -27,6 +27,7 @@
 import jakarta.validation.constraints.Min;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Size;
+import software.amazon.awssdk.core.signer.Signer;
 import software.amazon.awssdk.retries.api.RetryStrategy;
 import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
 import software.amazon.awssdk.services.s3.model.StorageClass;
@@ -79,6 +80,27 @@ public static StorageClass toStorageClass(StorageClassType storageClass)
         }
     }
 
+    public enum SignerType
+    {
+        AwsS3V4Signer,
+        Aws4Signer,
+        AsyncAws4Signer,
+        Aws4UnsignedPayloadSigner,
+        EventStreamAws4Signer;
+
+        @SuppressWarnings("deprecation")
+        public Signer create()
+        {
+            return switch (this) {
+                case AwsS3V4Signer -> software.amazon.awssdk.auth.signer.AwsS3V4Signer.create();
+                case Aws4Signer -> software.amazon.awssdk.auth.signer.Aws4Signer.create();
+                case AsyncAws4Signer -> software.amazon.awssdk.auth.signer.AsyncAws4Signer.create();
+                case Aws4UnsignedPayloadSigner -> software.amazon.awssdk.auth.signer.Aws4UnsignedPayloadSigner.create();
+                case EventStreamAws4Signer -> software.amazon.awssdk.auth.signer.EventStreamAws4Signer.create();
+            };
+        }
+    }
+
     public enum ObjectCannedAcl
     {
         NONE,
@@ -134,6 +156,7 @@ public static RetryStrategy getRetryStrategy(RetryMode retryMode)
     private String sseKmsKeyId;
     private String sseCustomerKey;
     private boolean useWebIdentityTokenCredentialsProvider;
+    private SignerType signerType;
     private DataSize streamingPartSize = DataSize.of(32, MEGABYTE);
     private boolean requesterPays;
     private Integer maxConnections = 500;
@@ -395,6 +418,19 @@ public boolean isSseWithCustomerKeyConfigValid()
         return true;
     }
 
+    public Optional<SignerType> getSignerType()
+    {
+        return Optional.ofNullable(signerType);
+    }
+
+    @ConfigDescription("AWS signing protocol to use while authenticating S3 requests")
+    @Config("s3.signer-type")
+    public S3FileSystemConfig setSignerType(SignerType signerType)
+    {
+        this.signerType = signerType;
+        return this;
+    }
+
     @NotNull
     @MinDataSize("5MB")
     @MaxDataSize("256MB")

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3FileSystemLoader.java

@@ -55,6 +55,7 @@
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.Executors.newCachedThreadPool;
 import static software.amazon.awssdk.core.checksums.ResponseChecksumValidation.WHEN_REQUIRED;
+import static software.amazon.awssdk.core.client.config.SdkAdvancedClientOption.SIGNER;
 
 final class S3FileSystemLoader
         implements Function<Location, TrinoFileSystemFactory>
@@ -277,7 +278,7 @@ private static StsClient createStsClient(S3FileSystemConfig config, Optional<Aws
 
     private static ClientOverrideConfiguration createOverrideConfiguration(OpenTelemetry openTelemetry, S3FileSystemConfig config, MetricPublisher metricPublisher)
     {
-        return ClientOverrideConfiguration.builder()
+        ClientOverrideConfiguration.Builder builder = ClientOverrideConfiguration.builder()
                 .addExecutionInterceptor(AwsSdkTelemetry.builder(openTelemetry)
                         .setCaptureExperimentalSpanAttributes(true)
                         .setRecordIndividualHttpError(true)
@@ -286,8 +287,9 @@ private static ClientOverrideConfiguration createOverrideConfiguration(OpenTelem
                         .maxAttempts(config.getMaxErrorRetries())
                         .build())
                 .appId(config.getApplicationId())
-                .addMetricPublisher(metricPublisher)
-                .build();
+                .addMetricPublisher(metricPublisher);
+        config.getSignerType().ifPresent(signer -> builder.putAdvancedOption(SIGNER, signer.create()));
+        return builder.build();
     }
 
     private static SdkHttpClient createHttpClient(S3FileSystemConfig config)

lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3FileSystemAwsS3.java

@@ -89,6 +89,7 @@ protected S3FileSystemFactory createS3FileSystemFactory()
                 .setRegion(region)
                 .setEndpoint(endpoint)
                 .setSupportsExclusiveCreate(true)
+                .setSignerType(S3FileSystemConfig.SignerType.AwsS3V4Signer)
                 .setStreamingPartSize(DataSize.valueOf("5.5MB")), new S3FileSystemStats());
     }
 

lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3FileSystemConfig.java

@@ -32,6 +32,7 @@
 import static io.airlift.units.DataSize.Unit.MEGABYTE;
 import static io.trino.filesystem.s3.S3FileSystemConfig.RetryMode.LEGACY;
 import static io.trino.filesystem.s3.S3FileSystemConfig.RetryMode.STANDARD;
+import static io.trino.filesystem.s3.S3FileSystemConfig.SignerType.Aws4Signer;
 import static io.trino.filesystem.s3.S3FileSystemConfig.StorageClassType.STANDARD_IA;
 import static java.util.concurrent.TimeUnit.MINUTES;
 
@@ -52,6 +53,7 @@ public void testDefaults()
                 .setStsEndpoint(null)
                 .setStsRegion(null)
                 .setStorageClass(StorageClassType.STANDARD)
+                .setSignerType(null)
                 .setCannedAcl(ObjectCannedAcl.NONE)
                 .setSseType(S3SseType.NONE)
                 .setRetryMode(LEGACY)
@@ -93,6 +95,7 @@ public void testExplicitPropertyMappings()
                 .put("s3.sts.endpoint", "sts.example.com")
                 .put("s3.sts.region", "us-west-2")
                 .put("s3.storage-class", "STANDARD_IA")
+                .put("s3.signer-type", "Aws4Signer")
                 .put("s3.canned-acl", "BUCKET_OWNER_FULL_CONTROL")
                 .put("s3.retry-mode", "STANDARD")
                 .put("s3.max-error-retries", "12")
@@ -131,6 +134,7 @@ public void testExplicitPropertyMappings()
                 .setStsEndpoint("sts.example.com")
                 .setStsRegion("us-west-2")
                 .setStorageClass(STANDARD_IA)
+                .setSignerType(Aws4Signer)
                 .setCannedAcl(ObjectCannedAcl.BUCKET_OWNER_FULL_CONTROL)
                 .setStreamingPartSize(DataSize.of(42, MEGABYTE))
                 .setRetryMode(STANDARD)

lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3FileSystemS3Mock.java

@@ -78,6 +78,7 @@ protected S3FileSystemFactory createS3FileSystemFactory()
                 .setRegion(Region.US_EAST_1.id())
                 .setPathStyleAccess(true)
                 .setStreamingPartSize(DataSize.valueOf("5.5MB"))
+                .setSignerType(S3FileSystemConfig.SignerType.AwsS3V4Signer)
                 .setSupportsExclusiveCreate(false), new S3FileSystemStats());
     }