[FLINK-40640][iceberg] Fix duplicate rows when an UPDATE changes the partition value

[dataengineervishal]

What is the purpose of this pull request?

Fixes FLINK-40640.

When the Iceberg pipeline connector processes an UPDATE that changes the value of a partitioned column, the row ends up duplicated in the Iceberg table instead of moved: a stale copy stays in the old partition, and the correct copy lands in the new partition. The stale copy never goes away on its own, and running compaction afterward does not help either, since rewrite_data_files only applies deletes that already exist and no delete marker for the stale row was ever created.

The cause is in RowDataUtils#convertDataChangeEventToRowData. Every UPDATE was collapsed into a single after-image row tagged RowKind.INSERT, and the before-image was discarded entirely. Iceberg's equality deletes are partition-scoped (a delete file only applies to data files in the same partition with a lower sequence number), so the equality-delete that Iceberg's upsert writer synthesizes was derived solely from the after-image and routed to the new partition, never touching the row sitting in the old partition.

Brief change log

• RowDataUtils#convertDataChangeEventToRowData now returns List<RowData> instead of a single RowData. For UPDATE, it returns a DELETE row built from the before-image followed by an INSERT row built from the after-image, so the old partition gets a real delete instead of losing the before-image entirely. INSERT/REPLACE/DELETE are unchanged (single row).
• IcebergWriter#write() writes each row returned by the conversion instead of assuming exactly one row per event.

Verifying this change

This change added tests and can be verified as follows:

• Added testUpdateChangingPartitionValueDeletesFromOldPartition in IcebergWriterTest, which creates a table partitioned by region, inserts and commits id=1, region=north, then applies an UPDATE moving it to region=east, and asserts the table contains exactly id=1, region=east afterward (no duplicate). This test fails against the pre-fix code with the exact duplicate described above, and passes with the fix.
• All existing tests in IcebergWriterTest continue to pass.

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[dataengineervishal]

@ferenc-csaky Can you please take a look and approve this?

[dataengineervishal]

@flinkbot run azure

[ferenc-csaky]

LGTM, straightforward change.

One non-blocking follow-up: AFAIK for updates that do not change either the Iceberg partition or identifier/equality key, the explicit DELETE is redundant because the upsert writer already creates an equality delete for the subsequent INSERT.

Avoiding that extra delete could reduce delete-file overhead for update-heavy workloads. That optimization should compare the actual Iceberg partition transform result (not only raw partition columns) and identifier fields, so that's more like an improvement handled separately from this correctness fix.

If no objections until then I'll merge this by EOD tomorrow.

[lvyanquan]

Great fix!

One optimization suggestion:

Since the Iceberg upsert writer automatically calls deleteKey() on every INSERT row, the explicit DELETE row is redundant for non-partitioned tables — the auto-generated equality-delete already covers the old row in the same partition. The bug only affects partitioned tables where the partition value changes.

Suggest only splitting when the table is partitioned. TableSchemaWrapper already holds the Schema, so adding a cached isPartitioned flag keeps per-record overhead to a single field read:

// TableSchemaWrapper
private final boolean partitioned;

public TableSchemaWrapper(Schema schema, ZoneId zoneId) {
    ...
    this.partitioned = !schema.partitionKeys().isEmpty();
}

public boolean isPartitioned() {
    return partitioned;
}

// RowDataUtils — UPDATE case
if (isPartitioned) {
    return Arrays.asList(
            toRowData(before, fieldGetters, RowKind.DELETE),
            toRowData(after, fieldGetters, RowKind.INSERT));
} else {
    return Collections.singletonList(
            toRowData(after, fieldGetters, RowKind.INSERT));
}

Overall LGTM, the optimization is non-blocking.

[dataengineervishal]

@ivyanquan Nice suggestion! I was also thinking along similar lines. I was considering checking whether the partition column value is the same in the before and after values. If they are the same, we could send only the INSERT otherwise, we would send both the DELETE for the old row and the INSERT for the new row.

I’ll include the isPartitioned check as well, since it will avoid the unnecessary overhead for non-partitioned tables.

I’ll take this up as a separate optimization task.

[lvyanquan]

Sounds good, looking forward to your updates. I was concerned that checking whether the partition value actually changed on every record might have a performance impact, so I'd suggest only checking whether the table is partitioned. Feel free to factor performance into your consideration.

[ferenc-csaky]

@lvyanquan is there a plan for a 3.6.1 release? Just asking if you think it would worth to backport this commit to release-3.6 or 3.7 is already on the horizon?

[lvyanquan]

@lvyanquan is there a plan for a 3.6.1 release? Just asking if you think it would worth to backport this commit to release-3.6 or 3.7 is already on the horizon?

There is currently no release plan for 3.6.1. Just wait for 3.7 — it's scheduled to be released in the coming weeks.