[
https://issues.apache.org/jira/browse/FLINK-40640?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ferenc Csaky updated FLINK-40640:
---------------------------------
Fix Version/s: cdc-3.6.1
> [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
> Priority: Major
> Labels: pull-request-available
> Fix For: cdc-3.6.1
>
>
> 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)
