Initial commit of Iceberg integration.

extensions/iceberg/build.gradle

@@ -38,16 +38,18 @@ dependencies {
     implementation "org.apache.iceberg:iceberg-core:${icebergVersion}"
     implementation "org.apache.iceberg:iceberg-core:${icebergVersion}:tests"
     implementation "org.apache.iceberg:iceberg-aws:${icebergVersion}"
+    // The following lines can be uncommented when we enable support for Azure and GCP
     // implementation "org.apache.iceberg:iceberg-azure:${icebergVersion}"
     // implementation "org.apache.iceberg:iceberg-gcp:${icebergVersion}"
     implementation "org.apache.iceberg:iceberg-bundled-guava:${icebergVersion}"
 
     implementation "org.apache.hadoop:hadoop-common:${hadoopVersion}"
     implementation "org.apache.hadoop:hadoop-hdfs-client:${hadoopVersion}"
 
-    runtimeOnly "org.apache.iceberg:iceberg-aws-bundle:${icebergVersion}"
+    // The following lines can be uncommented when we enable support for Azure and GCP
+    // runtimeOnly "org.apache.iceberg:iceberg-aws-bundle:${icebergVersion}"
     // runtimeOnly "org.apache.iceberg:iceberg-azure-bundle:${icebergVersion}"
-    // runtimeOnly "org.apache.iceberg:iceberg-gcp-bundle:${icebergVersion}"
+    runtimeOnly "org.apache.iceberg:iceberg-gcp-bundle:${icebergVersion}"
 
     testImplementation depCommonsCompress
     testImplementation depCommonsIo

extensions/iceberg/src/main/java/io/deephaven/iceberg/location/IcebergTableLocationFactory.java

@@ -5,17 +5,11 @@
 
 import io.deephaven.engine.table.impl.locations.TableKey;
 import io.deephaven.engine.table.impl.locations.TableLocation;
-import io.deephaven.engine.table.impl.locations.impl.NonexistentTableLocation;
 import io.deephaven.engine.table.impl.locations.impl.TableLocationFactory;
 import io.deephaven.engine.table.impl.locations.util.TableDataRefreshService;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-import java.io.File;
-import java.net.URI;
-
-import static io.deephaven.parquet.base.ParquetFileReader.FILE_URI_SCHEME;
-
 /**
  * {@link TableLocationFactory} for {@link IcebergTableLocation}s.
  */
@@ -31,11 +25,6 @@ public IcebergTableLocationFactory(@NotNull final Object readInstructions) {
     public TableLocation makeLocation(@NotNull final TableKey tableKey,
             @NotNull final IcebergTableLocationKey locationKey,
             @Nullable final TableDataRefreshService refreshService) {
-        final URI fileURI = locationKey.getURI();
-        if (!FILE_URI_SCHEME.equals(fileURI.getScheme()) || new File(fileURI).exists()) {
-            return new IcebergTableLocation(tableKey, locationKey, readInstructions);
-        } else {
-            return new NonexistentTableLocation(tableKey, locationKey);
-        }
+        return new IcebergTableLocation(tableKey, locationKey, readInstructions);
     }
 }

extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalog.java

@@ -69,18 +69,24 @@ public class IcebergCatalog {
 
         properties.put(CatalogProperties.FILE_IO_IMPL, instructions.fileIOImpl().value);
         if (instructions.fileIOImpl() == IcebergInstructions.FILEIO_IMPL.S3) {
-            final S3Instructions.Builder builder = S3Instructions.builder()
-                    .credentials(Credentials.basic(instructions.s3AccessKeyId(), instructions.s3SecretAccessKey()))
-                    .regionName(instructions.s3Region());
-
-            properties.put(AwsClientProperties.CLIENT_REGION, instructions.s3Region());
-            properties.put(S3FileIOProperties.ACCESS_KEY_ID, instructions.s3AccessKeyId());
-            properties.put(S3FileIOProperties.SECRET_ACCESS_KEY, instructions.s3SecretAccessKey());
+            // Configure the properties map from the Iceberg instructions.
+            if (instructions.s3AccessKeyId().isPresent() && instructions.s3SecretAccessKey().isPresent()) {
+                properties.put(S3FileIOProperties.ACCESS_KEY_ID, instructions.s3AccessKeyId().get());
+                properties.put(S3FileIOProperties.SECRET_ACCESS_KEY, instructions.s3SecretAccessKey().get());
+            }
+            if (instructions.s3Region().isPresent()) {
+                properties.put(AwsClientProperties.CLIENT_REGION, instructions.s3Region().get());
+            }
             if (instructions.s3EndpointOverride().isPresent()) {
                 properties.put(S3FileIOProperties.ENDPOINT, instructions.s3EndpointOverride().get());
-                builder.endpointOverride(instructions.s3EndpointOverride().get());
             }
-            s3Instructions = builder.build();
+
+            // The user may have provided readInstructions. If they did, we'll use them for the data file access.
+            // Otherwise we need to build one from the properties.
+            s3Instructions = instructions.readInstructions().isPresent()
+                    ? (S3Instructions) instructions.readInstructions().get()
+                    : buildS3Instructions(properties);
+
             // TODO: create a FileIO interface wrapping the Deephaven S3SeekableByteChannel/Provider
             fileIO = CatalogUtil.loadFileIO(instructions.fileIOImpl().value, properties, conf);
         } else {
@@ -91,6 +97,22 @@ public class IcebergCatalog {
         catalog.initialize(catalogName, properties);
     }
 
+    private static S3Instructions buildS3Instructions(final Map<String, String> properties) {
+        final S3Instructions.Builder builder = S3Instructions.builder();
+        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)));
+        }
+        if (properties.containsKey(AwsClientProperties.CLIENT_REGION)) {
+            builder.regionName(properties.get(AwsClientProperties.CLIENT_REGION));
+        }
+        if (properties.containsKey(S3FileIOProperties.ENDPOINT)) {
+            builder.endpointOverride(properties.get(S3FileIOProperties.ENDPOINT));
+        }
+        return builder.build();
+    }
+
     @SuppressWarnings("unused")
     public List<TableIdentifier> listTables(final Namespace namespace) {
         // TODO: have this return a Deephaven Table of table identifiers

extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergInstructions.java

@@ -81,17 +81,22 @@ public FILEIO_IMPL fileIOImpl() {
     /**
      * The AWS access key, used to identify the user interacting with services.
      */
-    public abstract String s3AccessKeyId();
+    public abstract Optional<String> s3AccessKeyId();
 
     /**
      * The AWS secret access key, used to authenticate the user interacting with services.
      */
-    public abstract String s3SecretAccessKey();
+    public abstract Optional<String> s3SecretAccessKey();
 
     /**
      * The AWS region to use for this connection.
      */
-    public abstract String s3Region();
+    public abstract Optional<String> s3Region();
+
+    /**
+     * The {@link io.deephaven.extensions.s3.S3Instructions} to use for reading the Iceberg data files.
+     */
+    public abstract Optional<Object> readInstructions();
 
     public interface Builder {
         Builder catalogImpl(CATALOG_IMPL catalogImpl);
@@ -110,6 +115,9 @@ public interface Builder {
 
         Builder s3Region(String s3Region);
 
+        @SuppressWarnings("unused")
+        Builder readInstructions(Object readInstructions);
+
         IcebergInstructions build();
     }
 
@@ -126,20 +134,4 @@ final void checkWarehouseLocation() {
             throw new IllegalArgumentException("warehouseLocation must be provided");
         }
     }
-
-    @Check
-    final void checkS3Fields() {
-        if (fileIOImpl() == FILEIO_IMPL.S3) {
-            if (s3AccessKeyId() == null || s3AccessKeyId().isEmpty()) {
-                throw new IllegalArgumentException("When using S3 FileIO, s3AccessKeyId must be provided");
-            }
-            if (s3SecretAccessKey() == null || s3SecretAccessKey().isEmpty()) {
-                throw new IllegalArgumentException("When using S3 FileIO, s3SecretAccessKey must be provided");
-            }
-            if (s3Region() == null || s3Region().isEmpty()) {
-                throw new IllegalArgumentException("When using S3 FileIO, s3Region must be provided");
-            }
-        }
-    }
-
 }