This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 017de09d2 [FLINK-40640][iceberg] Fix duplicate rows when an UPDATE
changes the partition value (#4536)
017de09d2 is described below
commit 017de09d204450e4e0346035391688185666aee3
Author: Vishal <[email protected]>
AuthorDate: Fri Sep 18 10:39:02 2026 +0530
[FLINK-40640][iceberg] Fix duplicate rows when an UPDATE changes the
partition value (#4536)
---
.../iceberg/sink/utils/RowDataUtils.java | 39 +++++++----
.../connectors/iceberg/sink/v2/IcebergWriter.java | 6 +-
.../iceberg/sink/v2/IcebergWriterTest.java | 79 ++++++++++++++++++++++
3 files changed, 107 insertions(+), 17 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
index 32ffa4a0a..51b6a8595 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
@@ -23,34 +23,43 @@ import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
/** Utils for convertion of {@link RowData} and {@link DataChangeEvent}. */
public class RowDataUtils {
- /** Convert {@link DataChangeEvent} to {@link RowData}. */
- public static RowData convertDataChangeEventToRowData(
+ /**
+ * Convert a {@link DataChangeEvent} to one or two {@link RowData}.
+ *
+ * <p>An {@code UPDATE} is split into a {@code DELETE} built from the
before-image followed by
+ * an {@code INSERT} built from the after-image, rather than a single
after-image row. Iceberg
+ * derives both the equality-delete key and the delete's target partition
from the row passed to
+ * the writer; collapsing an UPDATE to just the after-image (as before)
means a partition key
+ * change is never deleted from its old partition, leaving a stale
duplicate behind.
+ */
+ public static List<RowData> convertDataChangeEventToRowData(
DataChangeEvent dataChangeEvent, List<RecordData.FieldGetter>
fieldGetters) {
- RecordData recordData;
- RowKind kind;
switch (dataChangeEvent.op()) {
case INSERT:
- case UPDATE:
case REPLACE:
- {
- recordData = dataChangeEvent.after();
- kind = RowKind.INSERT;
- break;
- }
+ return Collections.singletonList(
+ toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
case DELETE:
- {
- recordData = dataChangeEvent.before();
- kind = RowKind.DELETE;
- break;
- }
+ return Collections.singletonList(
+ toRowData(dataChangeEvent.before(), fieldGetters,
RowKind.DELETE));
+ case UPDATE:
+ return Arrays.asList(
+ toRowData(dataChangeEvent.before(), fieldGetters,
RowKind.DELETE),
+ toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
default:
throw new IllegalArgumentException("don't support type of " +
dataChangeEvent.op());
}
+ }
+
+ private static RowData toRowData(
+ RecordData recordData, List<RecordData.FieldGetter> fieldGetters,
RowKind kind) {
GenericRowData genericRowData = new
GenericRowData(recordData.getArity());
genericRowData.setRowKind(kind);
for (int i = 0; i < recordData.getArity(); i++) {
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java
index 9af29d88c..6945ed57f 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java
@@ -166,10 +166,12 @@ public class IcebergWriter
writerMap.computeIfAbsent(
tableId, tableId1 ->
writerFactoryMap.get(tableId1).create());
TableSchemaWrapper tableSchemaWrapper = schemaMap.get(tableId);
- RowData rowData =
+ List<RowData> rowDataList =
RowDataUtils.convertDataChangeEventToRowData(
dataChangeEvent,
tableSchemaWrapper.getFieldGetters());
- writer.write(rowData);
+ for (RowData rowData : rowDataList) {
+ writer.write(rowData);
+ }
} else {
SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
TableId tableId = schemaChangeEvent.tableId();
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterTest.java
index 71e08de4c..4eda834ed 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterTest.java
@@ -1568,6 +1568,85 @@ public class IcebergWriterTest {
.containsExactlyInAnyOrder("1, c, null, null", "2, d, null,
null");
}
+ /**
+ * Verifies the core partition-key-update bug: when an UPDATE changes the
value of a partitioned
+ * column, the before-image must be deleted from its original partition,
not just the
+ * after-image inserted into the new one. Iceberg's equality deletes are
partition-scoped —
+ * deriving the delete purely from the after-image (the pre-fix behavior)
routes it to the new
+ * partition and never touches the old one, leaving the stale row behind
and producing a
+ * duplicate primary key on read.
+ */
+ @Test
+ public void testUpdateChangingPartitionValueDeletesFromOldPartition()
throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("type", "hadoop");
+ catalogOptions.put(
+ "warehouse",
+ new File(temporaryFolder.toFile(),
UUID.randomUUID().toString()).toString());
+ catalogOptions.put("cache-enabled", "false");
+ Catalog catalog =
+ CatalogUtil.buildIcebergCatalog(
+ "cdc-iceberg-catalog", catalogOptions, new
Configuration());
+ IcebergMetadataApplier icebergMetadataApplier = new
IcebergMetadataApplier(catalogOptions);
+
+ String jobId = UUID.randomUUID().toString();
+ String operatorId = UUID.randomUUID().toString();
+ IcebergWriter icebergWriter =
+ new IcebergWriter(
+ catalogOptions,
+ 1,
+ 1,
+ ZoneId.systemDefault(),
+ 0,
+ jobId,
+ operatorId,
+ new HashMap<>());
+
+ TableId tableId = TableId.parse("test.iceberg_table");
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("region",
DataTypes.VARCHAR(100).notNull())
+ .primaryKey("id")
+ .partitionKey("region")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId,
schema);
+ icebergMetadataApplier.applySchemaChange(createTableEvent);
+ icebergWriter.write(createTableEvent, null);
+
+ BinaryRecordDataGenerator generator =
+ new
BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ RecordData north =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("north")});
+ RecordData east =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("east")});
+
+ // INSERT(id=1, region="north") is committed as its own snapshot,
landing in a real,
+ // already-closed data file under the region=north partition.
+ icebergWriter.write(DataChangeEvent.insertEvent(tableId, north), null);
+ Collection<WriteResultWrapper> insertResults =
icebergWriter.prepareCommit();
+ IcebergCommitter icebergCommitter = new
IcebergCommitter(catalogOptions, new HashMap<>());
+ icebergCommitter.commit(
+ insertResults.stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList()));
+
+ // UPDATE(id=1, region: "north" -> "east") moves the row to a
different partition.
+ icebergWriter.write(DataChangeEvent.updateEvent(tableId, north, east),
null);
+ Collection<WriteResultWrapper> updateResults =
icebergWriter.prepareCommit();
+ icebergCommitter.commit(
+ updateResults.stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList()));
+ icebergCommitter.close();
+
+ // Only "1, east" must survive. Pre-fix, "1, north" also remained: the
equality-delete
+ // was derived solely from the after-image and routed to region=east,
never touching the
+ // region=north partition holding the original row.
+ List<String> result = fetchTableContent(catalog, tableId, null);
+ Assertions.assertThat(result).containsExactly("1, east");
+ }
+
private static long countSnapshots(Table table) {
long count = 0;
for (Snapshot ignored : table.snapshots()) {
