S3 storage classes config support

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

@@ -30,6 +30,8 @@ support:
   - Required region name for S3.
 * - `s3.path-style-access`
   - Use path-style access for all requests to S3
+* - `s3.storage-class`
+  - S3 storage class to use while writing data. Defaults to STANDARD.
 * - `s3.exclusive-create`
   - Whether conditional write is supported by the S3-compatible storage. Defaults to `true`.
 * - `s3.canned-acl`

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

@@ -15,6 +15,7 @@
 
 import io.trino.filesystem.s3.S3FileSystemConfig.ObjectCannedAcl;
 import io.trino.filesystem.s3.S3FileSystemConfig.S3SseType;
+import io.trino.filesystem.s3.S3FileSystemConfig.StorageClassType;
 import io.trino.spi.security.ConnectorIdentity;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
@@ -37,6 +38,7 @@ record S3Context(
         boolean requesterPays,
         S3SseContext s3SseContext,
         Optional<AwsCredentialsProvider> credentialsProviderOverride,
+        StorageClassType storageClass,
         ObjectCannedAcl cannedAcl,
         boolean exclusiveWriteSupported)
 {
@@ -56,7 +58,7 @@ public RequestPayer requestPayer()
 
     public S3Context withKmsKeyId(String kmsKeyId)
     {
-        return new S3Context(partSize, requesterPays, S3SseContext.withKmsKeyId(kmsKeyId), credentialsProviderOverride, cannedAcl, exclusiveWriteSupported);
+        return new S3Context(partSize, requesterPays, S3SseContext.withKmsKeyId(kmsKeyId), credentialsProviderOverride, storageClass, cannedAcl, exclusiveWriteSupported);
     }
 
     public S3Context withCredentials(ConnectorIdentity identity)
@@ -73,7 +75,7 @@ public S3Context withCredentials(ConnectorIdentity identity)
 
     public S3Context withSseCustomerKey(String key)
     {
-        return new S3Context(partSize, requesterPays, S3SseContext.withSseCustomerKey(key), credentialsProviderOverride, cannedAcl, exclusiveWriteSupported);
+        return new S3Context(partSize, requesterPays, S3SseContext.withSseCustomerKey(key), credentialsProviderOverride, storageClass, cannedAcl, exclusiveWriteSupported);
     }
 
     public S3Context withCredentialsProviderOverride(AwsCredentialsProvider credentialsProviderOverride)
@@ -83,6 +85,7 @@ public S3Context withCredentialsProviderOverride(AwsCredentialsProvider credenti
                 requesterPays,
                 s3SseContext,
                 Optional.of(credentialsProviderOverride),
+                storageClass,
                 cannedAcl,
                 exclusiveWriteSupported);
     }

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

@@ -29,6 +29,7 @@
 import jakarta.validation.constraints.Size;
 import software.amazon.awssdk.retries.api.RetryStrategy;
 import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
+import software.amazon.awssdk.services.s3.model.StorageClass;
 
 import java.util.Optional;
 import java.util.Set;
@@ -46,6 +47,22 @@ public enum S3SseType
         NONE, S3, KMS, CUSTOMER
     }
 
+    public enum StorageClassType
+    {
+        STANDARD,
+        STANDARD_IA,
+        INTELLIGENT_TIERING;
+
+        public static StorageClass toStorageClass(StorageClassType storageClass)
+        {
+            return switch (storageClass) {
+                case STANDARD -> StorageClass.STANDARD;
+                case STANDARD_IA -> StorageClass.STANDARD_IA;
+                case INTELLIGENT_TIERING -> StorageClass.INTELLIGENT_TIERING;
+            };
+        }
+    }
+
     public enum ObjectCannedAcl
     {
         NONE,
@@ -91,6 +108,7 @@ public static RetryStrategy getRetryStrategy(RetryMode retryMode)
     private String endpoint;
     private String region;
     private boolean pathStyleAccess;
+    private StorageClassType storageClass = StorageClassType.STANDARD;
     private String iamRole;
     private String roleSessionName = "trino-filesystem";
     private String externalId;
@@ -182,6 +200,19 @@ public S3FileSystemConfig setPathStyleAccess(boolean pathStyleAccess)
         return this;
     }
 
+    public StorageClassType getStorageClass()
+    {
+        return storageClass;
+    }
+
+    @Config("s3.storage-class")
+    @ConfigDescription("The S3 storage class to use when writing the data")
+    public S3FileSystemConfig setStorageClass(StorageClassType storageClass)
+    {
+        this.storageClass = storageClass;
+        return this;
+    }
+
     public String getIamRole()
     {
         return iamRole;

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

@@ -97,6 +97,7 @@ private S3FileSystemLoader(Optional<S3SecurityMappingProvider> mappingProvider,
                         config.getSseKmsKeyId(),
                         config.getSseCustomerKey()),
                 Optional.empty(),
+                config.getStorageClass(),
                 config.getCannedAcl(),
                 config.isSupportsExclusiveCreate());
     }

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

@@ -27,6 +27,7 @@
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.RequestPayer;
 import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.StorageClass;
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartResponse;
 
@@ -46,6 +47,7 @@
 import static com.google.common.base.Verify.verify;
 import static io.trino.filesystem.s3.S3FileSystemConfig.ObjectCannedAcl.getCannedAcl;
 import static io.trino.filesystem.s3.S3FileSystemConfig.S3SseType.NONE;
+import static io.trino.filesystem.s3.S3FileSystemConfig.StorageClassType.toStorageClass;
 import static io.trino.filesystem.s3.S3SseCUtils.encoded;
 import static io.trino.filesystem.s3.S3SseCUtils.md5Checksum;
 import static io.trino.filesystem.s3.S3SseRequestConfigurator.setEncryptionSettings;
@@ -68,6 +70,7 @@ final class S3OutputStream
     private final S3Context context;
     private final int partSize;
     private final RequestPayer requestPayer;
+    private final StorageClass storageClass;
     private final ObjectCannedACL cannedAcl;
     private final boolean exclusiveCreate;
     private final Optional<EncryptionKey> key;
@@ -97,6 +100,7 @@ public S3OutputStream(AggregatedMemoryContext memoryContext, Executor uploadExec
         this.context = requireNonNull(context, "context is null");
         this.partSize = context.partSize();
         this.requestPayer = context.requestPayer();
+        this.storageClass = toStorageClass(context.storageClass());
         this.cannedAcl = getCannedAcl(context.cannedAcl());
         this.key = requireNonNull(key, "key is null");
 
@@ -214,6 +218,7 @@ private void flushBuffer(boolean finished)
                     .requestPayer(requestPayer)
                     .bucket(location.bucket())
                     .key(location.key())
+                    .storageClass(storageClass)
                     .contentLength((long) bufferSize)
                     .applyMutation(builder -> {
                         if (exclusiveCreate) {
@@ -304,6 +309,7 @@ private CompletedPart uploadPage(byte[] data, int length)
                     .requestPayer(requestPayer)
                     .bucket(location.bucket())
                     .key(location.key())
+                    .storageClass(storageClass)
                     .applyMutation(builder ->
                         key.ifPresentOrElse(
                                 encryption ->

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

@@ -19,6 +19,7 @@
 import io.airlift.units.Duration;
 import io.trino.filesystem.s3.S3FileSystemConfig.ObjectCannedAcl;
 import io.trino.filesystem.s3.S3FileSystemConfig.S3SseType;
+import io.trino.filesystem.s3.S3FileSystemConfig.StorageClassType;
 import jakarta.validation.constraints.AssertTrue;
 import org.junit.jupiter.api.Test;
 
@@ -31,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.StorageClassType.STANDARD_IA;
 import static java.util.concurrent.TimeUnit.MINUTES;
 
 public class TestS3FileSystemConfig
@@ -49,6 +51,7 @@ public void testDefaults()
                 .setExternalId(null)
                 .setStsEndpoint(null)
                 .setStsRegion(null)
+                .setStorageClass(StorageClassType.STANDARD)
                 .setCannedAcl(ObjectCannedAcl.NONE)
                 .setSseType(S3SseType.NONE)
                 .setRetryMode(LEGACY)
@@ -88,6 +91,7 @@ public void testExplicitPropertyMappings()
                 .put("s3.external-id", "myid")
                 .put("s3.sts.endpoint", "sts.example.com")
                 .put("s3.sts.region", "us-west-2")
+                .put("s3.storage-class", "STANDARD_IA")
                 .put("s3.canned-acl", "BUCKET_OWNER_FULL_CONTROL")
                 .put("s3.retry-mode", "STANDARD")
                 .put("s3.max-error-retries", "12")
@@ -124,6 +128,7 @@ public void testExplicitPropertyMappings()
                 .setExternalId("myid")
                 .setStsEndpoint("sts.example.com")
                 .setStsRegion("us-west-2")
+                .setStorageClass(STANDARD_IA)
                 .setCannedAcl(ObjectCannedAcl.BUCKET_OWNER_FULL_CONTROL)
                 .setStreamingPartSize(DataSize.of(42, MEGABYTE))
                 .setRetryMode(STANDARD)