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.
@@ -48,11 +48,7 @@ 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);
+ final Catalog catalog = new RESTCatalog();
 
  // Configure the properties map from the Iceberg instructions.
  if (!Strings.isNullOrEmpty(accessKeyId) && !Strings.isNullOrEmpty(secretAccessKey)) {
@@ -65,11 +61,7 @@ public static IcebergCatalogAdapter createS3Rest(
  if (!Strings.isNullOrEmpty(endpointOverride)) {
  properties.put(S3FileIOProperties.ENDPOINT, endpointOverride);
  }
-
- final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI;
- catalog.initialize(catalogName, properties);
-
- return new IcebergCatalogAdapter(catalog, properties);
+ return createAdapterCommon(name, catalogURI, warehouseLocation, catalog, properties);
  }
 
  /**
@@ -91,12 +83,24 @@ 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);
 

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

@@ -5,14 +5,16 @@
 
 import io.deephaven.extensions.s3.S3Instructions.Builder;
 import io.deephaven.extensions.s3.testlib.SingletonContainers;
-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;
 
+@Tag("testcontainers")
 public class IcebergLocalStackTest extends IcebergToolsTest {
 
- @BeforeClass
+ @BeforeAll
  public static void initContainer() {
  // ensure container is started so container startup time isn't associated with a specific test
  SingletonContainers.LocalStack.init();

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

@@ -7,15 +7,17 @@
 import io.deephaven.extensions.s3.testlib.SingletonContainers;
 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;
 
+@Tag("testcontainers")
 public 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,135 @@
+//
+// 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.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.columnRenames().isEmpty()).isTrue();
+ assertThat(instructions.createTableIfNotExist()).isFalse();
+ assertThat(instructions.verifySchema()).isFalse();
+ 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())
+ .isTrue();
+ }
+
+ @Test
+ public void testSetCompressionCodecName() {
+ assertThat(IcebergParquetWriteInstructions.builder()
+ .compressionCodecName("GZIP")
+ .build()
+ .compressionCodecName())
+ .isEqualTo("GZIP");
+ }
+
+ @Test
+ void testSetMaximumDictionaryKeys() {
+ assertThat(IcebergParquetWriteInstructions.builder()
+ .maximumDictionaryKeys(100)
+ .build()
+ .maximumDictionaryKeys())
+ .isEqualTo(100);
+ }
+
+ @Test
+ public void testSetMaximumDictionarySize() {
+ assertThat(IcebergParquetWriteInstructions.builder()
+ .maximumDictionarySize(100)
+ .build()
+ .maximumDictionarySize())
+ .isEqualTo(100);
+ }
+
+ @Test
+ public void testSetTargetPageSize() {
+ assertThat(IcebergParquetWriteInstructions.builder()
+ .targetPageSize(1024 * 1024)
+ .build()
+ .targetPageSize())
+ .isEqualTo(1024 * 1024);
+ }
+
+ @Test
+ void testMinMaximumDictionaryKeys() {
+ try {
+ IcebergParquetWriteInstructions.builder()
+ .maximumDictionaryKeys(-1)
+ .build();
+ } catch (IllegalArgumentException e) {
+ assertThat(e).hasMessageContaining("maximumDictionaryKeys");
+ }
+ }
+
+ @Test
+ public void testMinMaximumDictionarySize() {
+ try {
+ IcebergParquetWriteInstructions.builder()
+ .maximumDictionarySize(-1)
+ .build();
+ } catch (IllegalArgumentException e) {
+ assertThat(e).hasMessageContaining("maximumDictionarySize");
+ }
+ }
+
+ @Test
+ void testMinTargetPageSize() {
+ try {
+ IcebergParquetWriteInstructions.builder()
+ .targetPageSize(1024)
+ .build();
+ } catch (IllegalArgumentException e) {
+ assertThat(e).hasMessageContaining("targetPageSize");
+ }
+ }
+
+ @Test
+ void toParquetInstructionTest() {
+ final IcebergParquetWriteInstructions icebergInstructions = IcebergParquetWriteInstructions.builder()
+ .compressionCodecName("GZIP")
+ .maximumDictionaryKeys(100)
+ .maximumDictionarySize(200)
+ .targetPageSize(1024 * 1024)
+ .build();
+ final Map<Integer, String> fieldIdToName = Map.of(2, "field2", 3, "field3");
+ final ParquetInstructions parquetInstructions = icebergInstructions.toParquetInstructions(
+ null, fieldIdToName);
+ assertThat(parquetInstructions.getCompressionCodecName()).isEqualTo("GZIP");
+ assertThat(parquetInstructions.getMaximumDictionaryKeys()).isEqualTo(100);
+ assertThat(parquetInstructions.getMaximumDictionarySize()).isEqualTo(200);
+ assertThat(parquetInstructions.getTargetPageSize()).isEqualTo(1024 * 1024);
+ assertThat(parquetInstructions.getFieldId("field1")).isEmpty();
+ assertThat(parquetInstructions.getFieldId("field2")).hasValue(2);
+ assertThat(parquetInstructions.getFieldId("field3")).hasValue(3);
+ assertThat(parquetInstructions.onWriteCompleted()).isEmpty();
+ }
+}