feat: Added support to write iceberg tables

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

@@ -45,12 +45,8 @@ public static IcebergCatalogAdapter createS3Rest(
  @Nullable final String secretAccessKey,
  @Nullable final String endpointOverride) {
 
- // Set up the properties map for the Iceberg catalog
+ // Set up the properties map for the Iceberg catalog, and configure it from Iceberg instructions.
  final Map<String, String> properties = new HashMap<>();
-
- final Catalog catalog = new RESTCatalog();
-
- // 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);
@@ -61,6 +57,8 @@ public static IcebergCatalogAdapter createS3Rest(
  if (!Strings.isNullOrEmpty(endpointOverride)) {
  properties.put(S3FileIOProperties.ENDPOINT, endpointOverride);
  }
+
+ final Catalog catalog = new RESTCatalog();
  return createAdapterCommon(name, catalogURI, warehouseLocation, catalog, properties);
  }
 

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

@@ -4,18 +4,23 @@
 package io.deephaven.iceberg.util;
 
 import io.deephaven.extensions.s3.S3Instructions.Builder;
-import io.deephaven.extensions.s3.testlib.SingletonContainers;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Tag;
+import io.deephaven.extensions.s3.testlib.SingletonContainers.LocalStack;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 
 import java.util.Map;
 
+import static org.apache.iceberg.aws.AwsClientProperties.CLIENT_REGION;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ACCESS_KEY_ID;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.SECRET_ACCESS_KEY;
+
 @Tag("testcontainers")
-public class IcebergLocalStackTest extends IcebergToolsTest {
+class IcebergLocalStackTest extends IcebergToolsTest {
 
  @BeforeAll
- public static void initContainer() {
+ 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

@@ -13,8 +13,13 @@
 
 import java.util.Map;
 
+import static org.apache.iceberg.aws.AwsClientProperties.CLIENT_REGION;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ACCESS_KEY_ID;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
+import static org.apache.iceberg.aws.s3.S3FileIOProperties.SECRET_ACCESS_KEY;
+
 @Tag("testcontainers")
-public class IcebergMinIOTest extends IcebergToolsTest {
+class IcebergMinIOTest extends IcebergToolsTest {
 
  @BeforeAll
  static void initContainer() {

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

@@ -7,7 +7,9 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Map;
+import java.util.Optional;
 
+import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown;
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 
 class IcebergParquetWriteInstructionsTest {
@@ -17,9 +19,9 @@ 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.dhToIcebergColumnRenames().isEmpty()).isTrue();
  assertThat(instructions.createTableIfNotExist()).isFalse();
- assertThat(instructions.verifySchema()).isFalse();
+ assertThat(instructions.verifySchema()).isEmpty();
  assertThat(instructions.compressionCodecName()).isEqualTo("SNAPPY");
  assertThat(instructions.maximumDictionaryKeys()).isEqualTo(1048576);
  assertThat(instructions.maximumDictionarySize()).isEqualTo(1048576);
@@ -40,11 +42,11 @@ void testSetVerifySchema() {
  .verifySchema(true)
  .build()
  .verifySchema())
- .isTrue();
+ .isEqualTo(Optional.of(true));
  }
 
  @Test
- public void testSetCompressionCodecName() {
+ void testSetCompressionCodecName() {
  assertThat(IcebergParquetWriteInstructions.builder()
  .compressionCodecName("GZIP")
  .build()
@@ -62,7 +64,7 @@ void testSetMaximumDictionaryKeys() {
  }
 
  @Test
- public void testSetMaximumDictionarySize() {
+ void testSetMaximumDictionarySize() {
  assertThat(IcebergParquetWriteInstructions.builder()
  .maximumDictionarySize(100)
  .build()
@@ -71,31 +73,34 @@ public void testSetMaximumDictionarySize() {
  }
 
  @Test
- public void testSetTargetPageSize() {
+ void testSetTargetPageSize() {
  assertThat(IcebergParquetWriteInstructions.builder()
- .targetPageSize(1024 * 1024)
+ .targetPageSize(1 << 20)
  .build()
  .targetPageSize())
- .isEqualTo(1024 * 1024);
+ .isEqualTo(1 << 20);
  }
 
  @Test
  void testMinMaximumDictionaryKeys() {
+
  try {
  IcebergParquetWriteInstructions.builder()
  .maximumDictionaryKeys(-1)
  .build();
+ failBecauseExceptionWasNotThrown(IllegalArgumentException.class);
  } catch (IllegalArgumentException e) {
  assertThat(e).hasMessageContaining("maximumDictionaryKeys");
  }
  }
 
  @Test
- public void testMinMaximumDictionarySize() {
+ void testMinMaximumDictionarySize() {
  try {
  IcebergParquetWriteInstructions.builder()
  .maximumDictionarySize(-1)
  .build();
+ failBecauseExceptionWasNotThrown(IllegalArgumentException.class);
  } catch (IllegalArgumentException e) {
  assertThat(e).hasMessageContaining("maximumDictionarySize");
  }
@@ -107,26 +112,63 @@ void testMinTargetPageSize() {
  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 icebergInstructions = IcebergParquetWriteInstructions.builder()
+ final IcebergParquetWriteInstructions writeInstructions = IcebergParquetWriteInstructions.builder()
  .compressionCodecName("GZIP")
  .maximumDictionaryKeys(100)
  .maximumDictionarySize(200)
- .targetPageSize(1024 * 1024)
+ .targetPageSize(1 << 20)
  .build();
  final Map<Integer, String> fieldIdToName = Map.of(2, "field2", 3, "field3");
- final ParquetInstructions parquetInstructions = icebergInstructions.toParquetInstructions(
+ 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(1024 * 1024);
+ assertThat(parquetInstructions.getTargetPageSize()).isEqualTo(1 << 20);
  assertThat(parquetInstructions.getFieldId("field1")).isEmpty();
  assertThat(parquetInstructions.getFieldId("field2")).hasValue(2);
  assertThat(parquetInstructions.getFieldId("field3")).hasValue(3);