From 8f75501084f3ecf233b7a185850b8e235df6d6d1 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 18 Jun 2026 11:58:15 -0400 Subject: [PATCH 1/3] improvement --- .../IO_Iceberg_Integration_Tests.json | 2 +- .../apache/beam/sdk/io/iceberg/AddFiles.java | 28 +++++++++++-------- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index b73af5e61a43..7ab7bcd9a9c6 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 1 + "modification": 2 } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index dc8106321db7..03d5ba996e4b 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -31,6 +31,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; import java.util.List; @@ -73,6 +74,7 @@ import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; @@ -97,7 +99,6 @@ import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.avro.Avro; -import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NoSuchTableException; @@ -532,17 +533,22 @@ private Table getOrCreateTable(String filePath, FileFormat format) throws IOExce org.apache.iceberg.Schema schema = getSchema(filePath, format); PartitionSpec spec = PartitionUtils.toPartitionSpec(partitionFields, schema); SortOrder sortOrder = SortOrderUtils.toSortOrder(sortFields, schema); - - Catalog.TableBuilder builder = - catalogConfig - .catalog() - .buildTable(tableId, schema) - .withPartitionSpec(spec) - .withSortOrder(sortOrder); - if (tableProps != null) { - builder.withProperties(tableProps); + Map properties = MoreObjects.firstNonNull(tableProps, new HashMap<>()); + if (properties.get(TableProperties.DEFAULT_NAME_MAPPING) == null) { + // Forces Name based resolution instead of position based resolution + NameMapping mapping = MappingUtil.create(schema); + String mappingJson = NameMappingParser.toJson(mapping); + properties.put(TableProperties.DEFAULT_NAME_MAPPING, mappingJson); } - return builder.create(); + + return catalogConfig + .catalog() + .buildTable(tableId, schema) + .withPartitionSpec(spec) + .withSortOrder(sortOrder) + .withProperties(properties) + .create(); + } catch (AlreadyExistsException e2) { // if table already exists, just load it return catalogConfig.catalog().loadTable(TableIdentifier.parse(identifier)); } From 68775ca97f5e2ae9107bfc75a382191036e64982 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 18 Jun 2026 12:02:59 -0400 Subject: [PATCH 2/3] fix nullness --- .../main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index 03d5ba996e4b..40421bc9e789 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -74,7 +74,6 @@ import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; @@ -533,7 +532,10 @@ private Table getOrCreateTable(String filePath, FileFormat format) throws IOExce org.apache.iceberg.Schema schema = getSchema(filePath, format); PartitionSpec spec = PartitionUtils.toPartitionSpec(partitionFields, schema); SortOrder sortOrder = SortOrderUtils.toSortOrder(sortFields, schema); - Map properties = MoreObjects.firstNonNull(tableProps, new HashMap<>()); + Map properties = new HashMap<>(); + if (tableProps != null) { + properties.putAll(tableProps); + } if (properties.get(TableProperties.DEFAULT_NAME_MAPPING) == null) { // Forces Name based resolution instead of position based resolution NameMapping mapping = MappingUtil.create(schema); From d4d1a399ec87fb2590d5a06575e81891503a8cd2 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 18 Jun 2026 12:20:57 -0400 Subject: [PATCH 3/3] clean --- .../main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index 40421bc9e789..7e0eeecd9e5a 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -532,10 +532,8 @@ private Table getOrCreateTable(String filePath, FileFormat format) throws IOExce org.apache.iceberg.Schema schema = getSchema(filePath, format); PartitionSpec spec = PartitionUtils.toPartitionSpec(partitionFields, schema); SortOrder sortOrder = SortOrderUtils.toSortOrder(sortFields, schema); - Map properties = new HashMap<>(); - if (tableProps != null) { - properties.putAll(tableProps); - } + Map properties = + tableProps != null ? new HashMap<>(tableProps) : new HashMap<>(); if (properties.get(TableProperties.DEFAULT_NAME_MAPPING) == null) { // Forces Name based resolution instead of position based resolution NameMapping mapping = MappingUtil.create(schema);