feat: Added support to write iceberg tables

extensions/iceberg/s3/build.gradle

@@ -29,7 +29,6 @@ dependencies {
     compileOnly libs.autoservice
     annotationProcessor libs.autoservice.compiler
 
-    testImplementation libs.junit4
     testImplementation project(':engine-test-utils')
 
     testImplementation libs.testcontainers
@@ -45,10 +44,20 @@ dependencies {
     testRuntimeOnly libs.slf4j.simple
 }
 
-TestTools.addEngineOutOfBandTest(project)
+test {
+    useJUnitPlatform {
+        excludeTags("testcontainers")
+    }
+}
+
+tasks.register('testOutOfBand', Test) {
+    useJUnitPlatform {
+        includeTags("testcontainers")
+    }
 
-testOutOfBand.dependsOn Docker.registryTask(project, 'localstack')
-testOutOfBand.systemProperty 'testcontainers.localstack.image', Docker.localImageName('localstack')
+    dependsOn Docker.registryTask(project, 'localstack')
+    systemProperty 'testcontainers.localstack.image', Docker.localImageName('localstack')
 
-testOutOfBand.dependsOn Docker.registryTask(project, 'minio')
-testOutOfBand.systemProperty 'testcontainers.minio.image', Docker.localImageName('minio')
+    dependsOn Docker.registryTask(project, 'minio')
+    systemProperty 'testcontainers.minio.image', Docker.localImageName('minio')
+}

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

@@ -7,7 +7,9 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.aws.AwsClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
+import org.apache.iceberg.aws.s3.S3FileIO;
 import org.apache.iceberg.aws.s3.S3FileIOProperties;
+import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.rest.RESTCatalog;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -16,12 +18,10 @@
 import java.util.Map;
 
 /**
- * Tools for accessing tables in the Iceberg table format.
+ * Tools for accessing tables in the Iceberg table format from S3.
  */
 @SuppressWarnings("unused")
 public class IcebergToolsS3 extends IcebergTools {
-    private static final String S3_FILE_IO_CLASS = "org.apache.iceberg.aws.s3.S3FileIO";
-
     /**
      * Create an Iceberg catalog adapter for a REST catalog backed by S3 storage. If {@code null} is provided for a
      * value, the system defaults will be used.
@@ -47,14 +47,6 @@ public static IcebergCatalogAdapter createS3Rest(
 
         // Set up the properties map for the Iceberg catalog
         final Map<String, String> properties = new HashMap<>();
-
-        final RESTCatalog catalog = new RESTCatalog();
-
-        properties.put(CatalogProperties.CATALOG_IMPL, catalog.getClass().getName());
-        properties.put(CatalogProperties.URI, catalogURI);
-        properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation);
-
-        // Configure the properties map from the Iceberg instructions.
         if (!Strings.isNullOrEmpty(accessKeyId) && !Strings.isNullOrEmpty(secretAccessKey)) {
             properties.put(S3FileIOProperties.ACCESS_KEY_ID, accessKeyId);
             properties.put(S3FileIOProperties.SECRET_ACCESS_KEY, secretAccessKey);
@@ -66,9 +58,8 @@ public static IcebergCatalogAdapter createS3Rest(
             properties.put(S3FileIOProperties.ENDPOINT, endpointOverride);
         }
 
-        final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI;
-        catalog.initialize(catalogName, properties);
-        return IcebergCatalogAdapter.of(catalog);
+        final Catalog catalog = new RESTCatalog();
+        return createAdapterCommon(name, catalogURI, warehouseLocation, catalog, properties);
     }
 
     /**
@@ -90,15 +81,27 @@ public static IcebergCatalogAdapter createGlue(
         // Set up the properties map for the Iceberg catalog
         final Map<String, String> properties = new HashMap<>();
 
-        final GlueCatalog catalog = new GlueCatalog();
+        final Catalog catalog = new GlueCatalog();
+        return createAdapterCommon(name, catalogURI, warehouseLocation, catalog, properties);
+    }
 
+    private static IcebergCatalogAdapter createAdapterCommon(
+            @Nullable final String name,
+            @NotNull final String catalogURI,
+            @NotNull final String warehouseLocation,
+            @NotNull final Catalog catalog,
+            @NotNull final Map<String, String> properties) {
         properties.put(CatalogProperties.CATALOG_IMPL, catalog.getClass().getName());
         properties.put(CatalogProperties.URI, catalogURI);
         properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation);
 
+        // Following is needed to write new manifest files when writing new data.
+        // Not setting this will result in using ResolvingFileIO.
+        properties.put(CatalogProperties.FILE_IO_IMPL, S3FileIO.class.getName());
+
         final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI;
         catalog.initialize(catalogName, properties);
 
-        return new IcebergCatalogAdapter(catalog, properties);
+        return IcebergCatalogAdapter.of(catalog, properties);
     }
 }

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

@@ -4,8 +4,9 @@
 package io.deephaven.iceberg.util;
 
 import io.deephaven.extensions.s3.S3Instructions.Builder;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
 import io.deephaven.extensions.s3.testlib.SingletonContainers.LocalStack;
-import org.junit.BeforeClass;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 
 import java.util.Map;
@@ -15,10 +16,11 @@
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.SECRET_ACCESS_KEY;
 
-public class IcebergLocalStackTest extends IcebergToolsTest {
+@Tag("testcontainers")
+class IcebergLocalStackTest extends IcebergToolsTest {
 
-    @BeforeClass
-    public static void initContainer() {
+    @BeforeAll
+    static void initContainer() {
         // ensure container is started so container startup time isn't associated with a specific test
         LocalStack.init();
     }

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

@@ -7,7 +7,8 @@
 import io.deephaven.extensions.s3.testlib.SingletonContainers.MinIO;
 import io.deephaven.stats.util.OSUtil;
 import org.junit.jupiter.api.Assumptions;
-import org.junit.BeforeClass;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 
 import java.util.Map;
@@ -17,10 +18,11 @@
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.SECRET_ACCESS_KEY;
 
-public class IcebergMinIOTest extends IcebergToolsTest {
+@Tag("testcontainers")
+class IcebergMinIOTest extends IcebergToolsTest {
 
-    @BeforeClass
-    public static void initContainer() {
+    @BeforeAll
+    static void initContainer() {
         // TODO(deephaven-core#5116): MinIO testcontainers does not work on OS X
         Assumptions.assumeFalse(OSUtil.runningMacOS(), "OSUtil.runningMacOS()");
         // ensure container is started so container startup time isn't associated with a specific test

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

@@ -0,0 +1,176 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.iceberg.util;
+
+import io.deephaven.parquet.table.ParquetInstructions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+class IcebergParquetWriteInstructionsTest {
+
+    @Test
+    void defaults() {
+        final IcebergParquetWriteInstructions instructions = IcebergParquetWriteInstructions.builder().build();
+        assertThat(instructions.tableDefinition().isEmpty()).isTrue();
+        assertThat(instructions.dataInstructions().isEmpty()).isTrue();
+        assertThat(instructions.dhToIcebergColumnRenames().isEmpty()).isTrue();
+        assertThat(instructions.createTableIfNotExist()).isFalse();
+        assertThat(instructions.verifySchema()).isEmpty();
+        assertThat(instructions.compressionCodecName()).isEqualTo("SNAPPY");
+        assertThat(instructions.maximumDictionaryKeys()).isEqualTo(1048576);
+        assertThat(instructions.maximumDictionarySize()).isEqualTo(1048576);
+    }
+
+    @Test
+    void testSetCreateTableIfNotExist() {
+        assertThat(IcebergParquetWriteInstructions.builder()
+                .createTableIfNotExist(true)
+                .build()
+                .createTableIfNotExist())
+                .isTrue();
+    }
+
+    @Test
+    void testSetVerifySchema() {
+        assertThat(IcebergParquetWriteInstructions.builder()
+                .verifySchema(true)
+                .build()
+                .verifySchema())
+                .hasValue(true);
+    }
+
+    @Test
+    void testSetCompressionCodecName() {
+        assertThat(IcebergParquetWriteInstructions.builder()
+                .compressionCodecName("GZIP")
+                .build()
+                .compressionCodecName())
+                .isEqualTo("GZIP");
+    }
+
+    @Test
+    void testSetMaximumDictionaryKeys() {
+        assertThat(IcebergParquetWriteInstructions.builder()
+                .maximumDictionaryKeys(100)
+                .build()
+                .maximumDictionaryKeys())
+                .isEqualTo(100);
+    }
+
+    @Test
+    void testSetMaximumDictionarySize() {
+        assertThat(IcebergParquetWriteInstructions.builder()
+                .maximumDictionarySize(100)
+                .build()
+                .maximumDictionarySize())
+                .isEqualTo(100);
+    }
+
+    @Test
+    void testSetTargetPageSize() {
+        assertThat(IcebergParquetWriteInstructions.builder()
+                .targetPageSize(1 << 20)
+                .build()
+                .targetPageSize())
+                .isEqualTo(1 << 20);
+    }
+
+    @Test
+    void testMinMaximumDictionaryKeys() {
+
+        try {
+            IcebergParquetWriteInstructions.builder()
+                    .maximumDictionaryKeys(-1)
+                    .build();
+            failBecauseExceptionWasNotThrown(IllegalArgumentException.class);
+        } catch (IllegalArgumentException e) {
+            assertThat(e).hasMessageContaining("maximumDictionaryKeys");
+        }
+    }
+
+    @Test
+    void testMinMaximumDictionarySize() {
+        try {
+            IcebergParquetWriteInstructions.builder()
+                    .maximumDictionarySize(-1)
+                    .build();
+            failBecauseExceptionWasNotThrown(IllegalArgumentException.class);
+        } catch (IllegalArgumentException e) {
+            assertThat(e).hasMessageContaining("maximumDictionarySize");
+        }
+    }
+
+    @Test
+    void testMinTargetPageSize() {
+        try {
+            IcebergParquetWriteInstructions.builder()
+                    .targetPageSize(1024)
+                    .build();
+            failBecauseExceptionWasNotThrown(IllegalArgumentException.class);
+        } catch (IllegalArgumentException e) {
+            assertThat(e).hasMessageContaining("targetPageSize");
+        }
+    }
+
+    @Test
+    void testSetToIcebergColumnRename() {
+        final IcebergParquetWriteInstructions instructions = IcebergParquetWriteInstructions.builder()
+                .putDhToIcebergColumnRenames("dh1", "ice1")
+                .putDhToIcebergColumnRenames("dh2", "ice2")
+                .build();
+        assertThat(instructions.dhToIcebergColumnRenames().size()).isEqualTo(2);
+        assertThat(instructions.dhToIcebergColumnRenames().get("dh1")).isEqualTo("ice1");
+        assertThat(instructions.dhToIcebergColumnRenames().get("dh2")).isEqualTo("ice2");
+
+        final IcebergParquetWriteInstructions instructions2 = IcebergParquetWriteInstructions.builder()
+                .putAllDhToIcebergColumnRenames(Map.of(
+                        "dh1", "ice1",
+                        "dh2", "ice2",
+                        "dh3", "ice3"))
+                .build();
+        assertThat(instructions2.dhToIcebergColumnRenames().size()).isEqualTo(3);
+        assertThat(instructions2.dhToIcebergColumnRenames().get("dh1")).isEqualTo("ice1");
+        assertThat(instructions2.dhToIcebergColumnRenames().get("dh2")).isEqualTo("ice2");
+        assertThat(instructions2.dhToIcebergColumnRenames().get("dh3")).isEqualTo("ice3");
+    }
+
+    @Test
+    void testToIcebergColumnRenameUniqueness() {
+        try {
+            IcebergParquetWriteInstructions.builder()
+                    .putDhToIcebergColumnRenames("dh1", "ice1")
+                    .putDhToIcebergColumnRenames("dh2", "ice1")
+                    .build();
+            failBecauseExceptionWasNotThrown(IllegalArgumentException.class);
+        } catch (final IllegalArgumentException e) {
+            assertThat(e).hasMessageContaining("Duplicate values in column renames");
+        }
+    }
+
+    @Test
+    void toParquetInstructionTest() {
+        final IcebergParquetWriteInstructions writeInstructions = IcebergParquetWriteInstructions.builder()
+                .compressionCodecName("GZIP")
+                .maximumDictionaryKeys(100)
+                .maximumDictionarySize(200)
+                .targetPageSize(1 << 20)
+                .build();
+        final Map<Integer, String> fieldIdToName = Map.of(2, "field2", 3, "field3");
+        final ParquetInstructions parquetInstructions = writeInstructions.toParquetInstructions(
+                null, fieldIdToName);
+
+        assertThat(parquetInstructions.getCompressionCodecName()).isEqualTo("GZIP");
+        assertThat(parquetInstructions.getMaximumDictionaryKeys()).isEqualTo(100);
+        assertThat(parquetInstructions.getMaximumDictionarySize()).isEqualTo(200);
+        assertThat(parquetInstructions.getTargetPageSize()).isEqualTo(1 << 20);
+        assertThat(parquetInstructions.getFieldId("field1")).isEmpty();
+        assertThat(parquetInstructions.getFieldId("field2")).hasValue(2);
+        assertThat(parquetInstructions.getFieldId("field3")).hasValue(3);
+        assertThat(parquetInstructions.onWriteCompleted()).isEmpty();
+    }
+}