feat: add generic Iceberg catalog adapter creation to Java / Python

extensions/iceberg/s3/build.gradle

@@ -26,8 +26,10 @@ dependencies {
     runtimeOnly libs.awssdk.sts
     runtimeOnly libs.awssdk.glue
 
-    testImplementation libs.junit4
+    compileOnly libs.autoservice
+    annotationProcessor libs.autoservice.compiler
 
+    testImplementation libs.junit4
     testImplementation project(':engine-test-utils')
 
     testImplementation libs.testcontainers

extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java

@@ -5,11 +5,9 @@
 
 import com.google.common.base.Strings;
 import org.apache.iceberg.CatalogProperties;
-import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.aws.AwsClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.aws.s3.S3FileIOProperties;
-import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.rest.RESTCatalog;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -68,12 +66,10 @@ public static IcebergCatalogAdapter createS3Rest(
             properties.put(S3FileIOProperties.ENDPOINT, endpointOverride);
         }
 
-        final FileIO fileIO = CatalogUtil.loadFileIO(S3_FILE_IO_CLASS, properties, null);
-
         final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI;
         catalog.initialize(catalogName, properties);
 
-        return new IcebergCatalogAdapter(catalog, fileIO);
+        return new IcebergCatalogAdapter(catalog, properties);
     }
 
     /**
@@ -101,11 +97,9 @@ public static IcebergCatalogAdapter createGlue(
         properties.put(CatalogProperties.URI, catalogURI);
         properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation);
 
-        final FileIO fileIO = CatalogUtil.loadFileIO(S3_FILE_IO_CLASS, properties, null);
-
         final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI;
         catalog.initialize(catalogName, properties);
 
-        return new IcebergCatalogAdapter(catalog, fileIO);
+        return new IcebergCatalogAdapter(catalog, properties);
     }
 }

extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/S3InstructionsProviderPlugin.java

@@ -0,0 +1,54 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.iceberg.util;
+
+import com.google.auto.service.AutoService;
+import io.deephaven.extensions.s3.Credentials;
+import io.deephaven.extensions.s3.S3Instructions;
+import io.deephaven.iceberg.internal.DataInstructionsProviderPlugin;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
+import org.jetbrains.annotations.NotNull;
+
+import java.net.URI;
+import java.util.Map;
+
+/**
+ * {@link io.deephaven.iceberg.internal.DataInstructionsProviderPlugin} implementation used for reading files from S3.
+ */
+@AutoService(io.deephaven.iceberg.internal.DataInstructionsProviderPlugin.class)
+@SuppressWarnings("unused")
+public final class S3InstructionsProviderPlugin implements DataInstructionsProviderPlugin {
+    @Override
+    public Object createInstructions(@NotNull final URI uri, @NotNull final Map<String, String> properties) {
+        // If the URI scheme is "s3","s3a","s3n" or if the properties contain one of these specific keys, we can
+        // create a useful S3Instructions object.
+        if (uri.getScheme().equals("s3")
+                || uri.getScheme().equals("s3a")
+                || uri.getScheme().equals("s3n")
+                || properties.containsKey(AwsClientProperties.CLIENT_REGION)
+                || properties.containsKey(S3FileIOProperties.ACCESS_KEY_ID)
+                || properties.containsKey(S3FileIOProperties.SECRET_ACCESS_KEY)
+                || properties.containsKey(S3FileIOProperties.ENDPOINT)) {
+
+            final S3Instructions.Builder builder = S3Instructions.builder();
+            if (properties.containsKey(AwsClientProperties.CLIENT_REGION)) {
+                builder.regionName(properties.get(AwsClientProperties.CLIENT_REGION));
+            }
+            if (properties.containsKey(S3FileIOProperties.ENDPOINT)) {
+                builder.endpointOverride(properties.get(S3FileIOProperties.ENDPOINT));
+            }
+            if (properties.containsKey(S3FileIOProperties.ACCESS_KEY_ID)
+                    && properties.containsKey(S3FileIOProperties.SECRET_ACCESS_KEY)) {
+                builder.credentials(
+                        Credentials.basic(properties.get(S3FileIOProperties.ACCESS_KEY_ID),
+                                properties.get(S3FileIOProperties.SECRET_ACCESS_KEY)));
+            }
+            return builder.build();
+        }
+
+        // We have no useful properties for creating an S3Instructions object.
+        return null;
+    }
+}

extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergLocalStackTest.java

@@ -8,6 +8,8 @@
 import org.junit.BeforeClass;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 
+import java.util.Map;
+
 public class IcebergLocalStackTest extends IcebergToolsTest {
 
     @BeforeClass
@@ -25,4 +27,9 @@ public Builder s3Instructions(final Builder builder) {
     public S3AsyncClient s3AsyncClient() {
         return SingletonContainers.LocalStack.s3AsyncClient();
     }
+
+    @Override
+    public Map<String, String> s3Properties() {
+        return SingletonContainers.LocalStack.s3Properties();
+    }
 }

extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergMinIOTest.java

@@ -10,6 +10,8 @@
 import org.junit.BeforeClass;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 
+import java.util.Map;
+
 public class IcebergMinIOTest extends IcebergToolsTest {
 
     @BeforeClass
@@ -29,4 +31,10 @@ public Builder s3Instructions(final Builder builder) {
     public S3AsyncClient s3AsyncClient() {
         return SingletonContainers.MinIO.s3AsyncClient();
     }
+
+    @Override
+    public Map<String, String> s3Properties() {
+        return SingletonContainers.MinIO.s3Properties();
+    }
+
 }