This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 524fe9ef1b Flink: Remap equality field ids on table update (#18507)
524fe9ef1b is described below
commit 524fe9ef1b4aa72be19623f667738b21a02b23b2
Author: Sergei Nikolaev <[email protected]>
AuthorDate: Fri Oct 16 17:08:20 2026 +0200
Flink: Remap equality field ids on table update (#18507)
---
.../sink/dynamic/DynamicTableUpdateOperator.java | 26 +++-
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 133 ++++++++++++++-------
.../dynamic/TestDynamicTableUpdateOperator.java | 38 ++++++
3 files changed, 156 insertions(+), 41 deletions(-)
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java
index 93c268ff86..b6795bf79b 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java
@@ -18,15 +18,20 @@
*/
package org.apache.iceberg.flink.sink.dynamic;
+import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.data.RowData;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.CatalogLoader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
/**
* An optional operator to perform table updates for tables (e.g. schema
update) in a non-concurrent
@@ -87,8 +92,10 @@ class DynamicTableUpdateOperator
data.spec(),
tableCreator);
TableMetadataCache.ResolvedSchemaInfo compareInfo = newData.f0;
+ Schema tableSchema = compareInfo.resolvedTableSchema();
- data.setSchema(compareInfo.resolvedTableSchema());
+ data.setEqualityFieldIds(remapFieldIds(data.equalityFields(),
data.schema(), tableSchema));
+ data.setSchema(tableSchema);
data.setSpec(newData.f1);
RowData newRowData = (RowData)
newData.f0.recordConverter().convert(data.rowData());
@@ -96,4 +103,21 @@ class DynamicTableUpdateOperator
return data;
}
+
+ private Set<Integer> remapFieldIds(Set<Integer> fieldIds, Schema fromSchema,
Schema toSchema) {
+ if (fieldIds == null || fieldIds.isEmpty()) {
+ return fieldIds;
+ }
+
+ Set<Integer> remappedIds =
Sets.newHashSetWithExpectedSize(fieldIds.size());
+ for (Integer fieldId : fieldIds) {
+ String name = fromSchema.findColumnName(fieldId);
+ Types.NestedField field =
+ caseSensitive ? toSchema.findField(name) :
toSchema.caseInsensitiveFindField(name);
+ Preconditions.checkNotNull(field, "Equality field %s does not exist in
schema", name);
+ remappedIds.add(field.fieldId());
+ }
+
+ return remappedIds;
+ }
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 34a74cdf1b..9e2f00bca0 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -77,6 +77,7 @@ import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableUtil;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.data.IcebergGenerics;
@@ -1034,51 +1035,64 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
runTest(rows);
}
- @Test
- void testUpsert() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testUpsert(boolean immediateUpdate) throws Exception {
+ // A restart replays the records against the already created table, which
hides failures
+ Configuration configuration = new Configuration();
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "none");
+ env.configure(configuration);
+
+ // Field IDs differ from the ones the catalog assigns on table creation
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(0, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(1, "data", Types.StringType.get()));
+ Types.NestedField equalityField = schema.findField("id");
List<DynamicIcebergDataImpl> rows =
Lists.newArrayList(
// Insert one rows
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
false),
// Remaining rows are duplicates
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
true),
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
true),
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
true));
- executeDynamicSink(rows, env, true, 1, null);
+ executeDynamicSink(rows, env, immediateUpdate, 1, null);
- try (CloseableIterable<Record> iterable =
- IcebergGenerics.read(
-
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of("default", "t1")))
- .build()) {
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of("default", "t1"));
+ // The record's equality field ID doesn't exist in the created table, so
with immediateUpdate
+ // disabled the write fails unless equality field IDs are remapped to the
table schema.
+
assertThat(table.schema().findColumnName(equalityField.fieldId())).isNull();
+ try (CloseableIterable<Record> iterable =
IcebergGenerics.read(table).build()) {
List<Record> records = Lists.newArrayList();
for (Record record : iterable) {
records.add(record);
@@ -1093,61 +1107,70 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
}
}
- @Test
- void testUpsertV3() throws Exception {
- ImmutableMap<String, String> properties =
ImmutableMap.of(TableProperties.FORMAT_VERSION, "3");
- CATALOG_EXTENSION
- .catalog()
- .createTable(
- TableIdentifier.of(DATABASE, "t1"),
- SimpleDataUtil.SCHEMA,
- PartitionSpec.unpartitioned(),
- null,
- properties);
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testUpsertV3(boolean immediateUpdate) throws Exception {
+ // A restart replays the records against the already created table, which
hides failures
+ Configuration configuration = new Configuration();
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "none");
+ env.configure(configuration);
+
+ Map<String, String> tableProps =
ImmutableMap.of(TableProperties.FORMAT_VERSION, "3");
+ TableCreator tableCreator =
+ (catalog, identifier, tableSchema, spec) ->
+ catalog.createTable(identifier, tableSchema, spec, null,
tableProps);
+ // Field IDs differ from the ones the catalog assigns on table creation
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(0, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(1, "data", Types.StringType.get()));
+ Types.NestedField equalityField = schema.findField("id");
List<DynamicIcebergDataImpl> rows =
Lists.newArrayList(
// Insert one rows
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
false),
// Remaining rows are duplicates
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
true),
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
true),
new DynamicIcebergDataImpl(
- SimpleDataUtil.SCHEMA,
+ schema,
"t1",
SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned(),
true,
- Sets.newHashSet("id"),
+ Sets.newHashSet(equalityField.name()),
true));
- executeDynamicSink(rows, env, true, 1, null);
+ executeDynamicSink(rows, env, immediateUpdate, 1, null, false,
Maps.newHashMap(), tableCreator);
- try (CloseableIterable<Record> iterable =
- IcebergGenerics.read(
-
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of("default", "t1")))
- .build()) {
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of("default", "t1"));
+ assertThat(TableUtil.formatVersion(table)).isEqualTo(3);
+ // The record's equality field ID doesn't exist in the created table, so
with immediateUpdate
+ // disabled the write fails unless equality field IDs are remapped to the
table schema.
+
assertThat(table.schema().findColumnName(equalityField.fieldId())).isNull();
+ try (CloseableIterable<Record> iterable =
IcebergGenerics.read(table).build()) {
List<Record> records = Lists.newArrayList();
for (Record record : iterable) {
records.add(record);
@@ -1932,7 +1955,14 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
Map<String, String> writeProperties)
throws Exception {
executeDynamicSink(
- dynamicData, env, immediateUpdate, parallelism, null, false,
writeProperties);
+ dynamicData,
+ env,
+ immediateUpdate,
+ parallelism,
+ null,
+ false,
+ writeProperties,
+ TableCreator.DEFAULT);
verifyResults(dynamicData, writeProperties);
}
@@ -1968,6 +1998,27 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
boolean overwrite,
Map<String, String> writeProperties)
throws Exception {
+ executeDynamicSink(
+ dynamicData,
+ env,
+ immediateUpdate,
+ parallelism,
+ commitHook,
+ overwrite,
+ writeProperties,
+ TableCreator.DEFAULT);
+ }
+
+ private void executeDynamicSink(
+ List<DynamicIcebergDataImpl> dynamicData,
+ StreamExecutionEnvironment env,
+ boolean immediateUpdate,
+ int parallelism,
+ @Nullable CommitHook commitHook,
+ boolean overwrite,
+ Map<String, String> writeProperties,
+ TableCreator tableCreator)
+ throws Exception {
DataStream<DynamicIcebergDataImpl> dataStream =
env.fromData(dynamicData, TypeInformation.of(new TypeHint<>() {}));
env.setParallelism(parallelism);
@@ -1979,6 +2030,7 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
.catalogLoader(CATALOG_EXTENSION.catalogLoader())
.writeParallelism(parallelism)
.immediateTableUpdate(immediateUpdate)
+ .tableCreator(tableCreator)
.setSnapshotProperty("commit.retry.num-retries", "0")
.overwrite(overwrite)
.setAll(writeProperties)
@@ -1989,6 +2041,7 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
.catalogLoader(CATALOG_EXTENSION.catalogLoader())
.writeParallelism(parallelism)
.immediateTableUpdate(immediateUpdate)
+ .tableCreator(tableCreator)
.overwrite(overwrite)
.setAll(writeProperties)
.append();
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java
index f6b2b368c2..80f0dc2bcb 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java
@@ -25,6 +25,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.catalog.Catalog;
@@ -120,6 +121,43 @@ class TestDynamicTableUpdateOperator {
assertThat(catalog.loadTable(table).schema().schemaId()).isEqualTo(output.schema().schemaId());
}
+ @Test
+ void testEqualityFieldIdsRemappedToTableSchema() throws Exception {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier table = TableIdentifier.of(TABLE);
+
+ DynamicTableUpdateOperator operator =
+ new DynamicTableUpdateOperator(
+ CATALOG_EXTENSION.catalogLoader(),
+ TableCreator.DEFAULT,
+ flinkDynamicSinkConfiguration(CASE_SENSITIVE, PRESERVE_COLUMNS));
+ operator.open(null);
+
+ catalog.createTable(table, SCHEMA1);
+ Schema inputSchema =
+ new Schema(
+ Types.NestedField.required(2, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(1, "data", Types.StringType.get()));
+ Types.NestedField equalityField = inputSchema.findField("id");
+ DynamicRecordInternal input =
+ new DynamicRecordInternal(
+ TABLE,
+ "branch",
+ inputSchema,
+ GenericRowData.of(1, StringData.fromString("test")),
+ PartitionSpec.unpartitioned(),
+ 42,
+ true,
+ Collections.singleton(equalityField.fieldId()));
+ DynamicRecordInternal output = operator.map(input);
+
+ assertThat(output.schema().findColumnName(equalityField.fieldId()))
+ .isNotEqualTo(equalityField.name());
+ assertThat(output.equalityFields())
+ .map(output.schema()::findColumnName)
+ .containsExactly(equalityField.name());
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testCaseInSensitivity(boolean caseSensitive) throws Exception {
