VISHAL B created FLINK-40640:
--------------------------------
Summary: [Iceberg] UPDATE that changes a partitioned column's
value leaves a duplicate stale row in the old partition
Key: FLINK-40640
URL: https://issues.apache.org/jira/browse/FLINK-40640
Project: Flink
Issue Type: Bug
Components: Flink CDC
Affects Versions: cdc-3.6.0
Reporter: VISHAL B
When a MySQL→Iceberg CDC pipeline processes an UPDATE that changes the value of
a
partitioned column, the row ends up duplicated in the Iceberg table: one stale
copy
in the old partition, and the correct copy in the new partition.
To reproduce this, create a table partitioned by identity(region) with id as
the primary key. Insert id=1 with region='north' and let it commit. Then
update that same row to region='east'. Reading the table back afterward shows
id=1 twice, once under region='north' (stale) and once under region='east'
(correct).
The cause is in RowDataUtils which convert DataChangeEvent to RowData inside
flink-cdc-pipeline-connector-iceberg. Every UPDATE gets collapsed into a
single after image row tagged RowKind.INSERT, and the before image is simply
thrown away.
Iceberg's equality deletes only apply within the partition they were written
to, so a delete file targets rows in the same partition with a lower sequence
number. Since the before image never makes it into the write path, the
connector never produces a delete for the row's original partition. The
equality delete that Iceberg's upsert writer generates automatically (through
BaseDeltaTaskWriter, since IcebergWriter always runs with upsert enabled) is
built entirely from the after image, so it lands in the new partition
instead. Nothing ever deletes the row sitting in the old partition, so it
just stays there permanently.
Running compaction afterward does not help either. rewrite_data_files only
applies deletes that already exist, and since no delete marker was ever
created for the stale row, compaction leaves it untouched.
It's fair to say this is working as documented from Iceberg's point of view.
Equality deletes being partition scoped is intentional Iceberg behavior, not
a bug in Iceberg itself. But updating a partitioned column is a pretty common
thing to hit in real CDC pipelines, often the column being partitioned on is
something like status, region, or tenant, and those are exactly the columns
that tend to change. So the connector still needs to handle this correctly
rather than letting Iceberg's partition scoping quietly corrupt data. Sending
one extra delete for the before image on every update is about the same
amount of work the writer already does for every insert under upsert mode,
so fixing this properly should not add any meaningful overhead.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
