Allow using S3 CRT client for native S3 filesystem

lib/trino-filesystem-s3/pom.xml

@@ -141,6 +141,11 @@
             <artifactId>aws-core</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>aws-crt-client</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>http-client-spi</artifactId>

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

@@ -177,6 +177,7 @@ public static RetryStrategy getRetryStrategy(RetryMode retryMode)
     private boolean supportsExclusiveCreate = true;
     private boolean crossRegionAccessEnabled;
     private String applicationId = "Trino";
+    private boolean enableCrtClient;
 
     public String getAwsAccessKey()
     {
@@ -650,4 +651,17 @@ public S3FileSystemConfig setApplicationId(String applicationId)
         this.applicationId = applicationId;
         return this;
     }
+
+    public boolean isEnableCrtClient()
+    {
+        return enableCrtClient;
+    }
+
+    @Config("s3.use-crt-client")
+    @ConfigDescription("Use AWS SDK for Java 2.x CRT client for S3 operations")
+    public S3FileSystemConfig setEnableCrtClient(boolean enableCrtClient)
+    {
+        this.enableCrtClient = enableCrtClient;
+        return this;
+    }
 }

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

@@ -14,6 +14,7 @@
 package io.trino.filesystem.s3;
 
 import com.google.inject.Inject;
+import io.airlift.units.Duration;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.instrumentation.awssdk.v2_2.AwsSdkTelemetry;
 import io.trino.filesystem.Location;
@@ -29,6 +30,7 @@
 import software.amazon.awssdk.http.SdkHttpClient;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
 import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.http.crt.AwsCrtHttpClient;
 import software.amazon.awssdk.metrics.MetricPublisher;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.LegacyMd5Plugin;
@@ -294,6 +296,33 @@ private static ClientOverrideConfiguration createOverrideConfiguration(OpenTelem
 
     private static SdkHttpClient createHttpClient(S3FileSystemConfig config)
     {
+        if (config.isEnableCrtClient()) {
+            AwsCrtHttpClient.Builder builder = AwsCrtHttpClient
+                    .builder()
+                    .maxConcurrency(config.getMaxConnections());
+
+            config.getSocketConnectTimeout()
+                    .map(Duration::toJavaTime)
+                    .ifPresent(builder::connectionTimeout);
+
+            config.getConnectionMaxIdleTime()
+                    .map(Duration::toJavaTime)
+                    .ifPresent(builder::connectionMaxIdleTime);
+
+            if (config.getHttpProxy() != null) {
+                builder.proxyConfiguration(proxyBuilder -> {
+                    proxyBuilder.host(config.getHttpProxy().getHost());
+                    proxyBuilder.port(config.getHttpProxy().getPort());
+                    proxyBuilder.scheme(config.isHttpProxySecure() ? "https" : "http");
+                    proxyBuilder.username(config.getHttpProxyUsername());
+                    proxyBuilder.password(config.getHttpProxyPassword());
+                    proxyBuilder.nonProxyHosts(config.getNonProxyHosts());
+                });
+            }
+
+            return builder.build();
+        }
+
         ApacheHttpClient.Builder client = ApacheHttpClient.builder()
                 .maxConnections(config.getMaxConnections())
                 .tcpKeepAlive(config.getTcpKeepAlive());

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

@@ -111,6 +111,9 @@ public int read()
     public int read(byte[] bytes, int offset, int length)
             throws IOException
     {
+        if (offset < 0) {
+            throw new IndexOutOfBoundsException("Offset cannot be negative: " + offset);
+        }
         ensureOpen();
         seekStream(false);
 

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

@@ -90,7 +90,8 @@ protected S3FileSystemFactory createS3FileSystemFactory()
                 .setEndpoint(endpoint)
                 .setSupportsExclusiveCreate(true)
                 .setSignerType(S3FileSystemConfig.SignerType.AwsS3V4Signer)
-                .setStreamingPartSize(DataSize.valueOf("5.5MB")), new S3FileSystemStats());
+                .setStreamingPartSize(DataSize.valueOf("5.5MB"))
+                .setEnableCrtClient(true), new S3FileSystemStats());
     }
 
     @Test

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

@@ -77,7 +77,8 @@ public void testDefaults()
                 .setHttpProxyPreemptiveBasicProxyAuth(false)
                 .setSupportsExclusiveCreate(true)
                 .setCrossRegionAccessEnabled(false)
-                .setApplicationId("Trino"));
+                .setApplicationId("Trino")
+                .setEnableCrtClient(false));
     }
 
     @Test
@@ -120,6 +121,7 @@ public void testExplicitPropertyMappings()
                 .put("s3.exclusive-create", "false")
                 .put("s3.application-id", "application id")
                 .put("s3.cross-region-access", "true")
+                .put("s3.use-crt-client", "true")
                 .buildOrThrow();
 
         S3FileSystemConfig expected = new S3FileSystemConfig()
@@ -158,7 +160,8 @@ public void testExplicitPropertyMappings()
                 .setHttpProxyPreemptiveBasicProxyAuth(true)
                 .setSupportsExclusiveCreate(false)
                 .setCrossRegionAccessEnabled(true)
-                .setApplicationId("application id");
+                .setApplicationId("application id")
+                .setEnableCrtClient(true);
 
         assertFullMapping(properties, expected);
     }

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

@@ -80,6 +80,7 @@ protected S3FileSystemFactory createS3FileSystemFactory()
                 .setAwsAccessKey(Minio.MINIO_ACCESS_KEY)
                 .setAwsSecretKey(Minio.MINIO_SECRET_KEY)
                 .setSupportsExclusiveCreate(true)
+                .setEnableCrtClient(true)
                 .setStreamingPartSize(DataSize.valueOf("5.5MB")), new S3FileSystemStats());
     }