[
https://issues.apache.org/jira/browse/FLINK-38450?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18070196#comment-18070196
]
Spoorthi Basu commented on FLINK-38450:
---------------------------------------
Hi, I have investigated this issue and was able to reproduce the duplication
scenario.
The root cause appears to be related to Iceberg’s sequence number semantics for
equality deletes. When multiple flushes occur within a single Flink checkpoint,
the resulting data and equality-delete files are committed into the same
Iceberg snapshot and therefore share the same sequence number. Since equality
deletes only apply to data files with strictly lower sequence numbers, deletes
from later batches cannot suppress stale rows written by earlier batches in the
same snapshot, which can lead to duplicate records.
To address this, I propose tracking per-table write batches within a checkpoint
and committing them sequentially instead of merging them into a single
RowDelta. This would ensure each batch is assigned a higher sequence number,
allowing equality deletes to correctly suppress earlier data.
In addition, I propose:
* avoiding global writer rotation on non-checkpoint flush(false) calls
* limiting writer completion to table-scoped cases such as schema changes
* ensuring correct retry behavior by handling partial batch commits
idempotently
I have implemented this approach locally and validated it with tests covering
the main scenario and related edge cases.
If this approach looks good to the community, I would appreciate being assigned
to this ticket so I can open a PR.
> flink-cdc sink to iceberg table record duplication
> --------------------------------------------------
>
> Key: FLINK-38450
> URL: https://issues.apache.org/jira/browse/FLINK-38450
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: 3.0.0
> Reporter: ChaoFang
> Priority: Major
> Labels: Flink-CDC, iceberg
> Fix For: cdc-3.5.0
>
>
> When multiple modifications to the same PrimaryId occur within a single
> checkpoint and are flushed separately, records are written to different file
> batches (data/eqDelete/posDelete files) sharing the same seq_number. During
> reads, the eqDeleteFile fails to correctly filter stale data, resulting in
> duplicate records.
> *Repro Steps* (in 1 checkpoint):
> # Got Rcoed(id=1,name="a") insert event.
> # Got flush event (Triggered from some other change).
> # Rcoed(id=1,name="2") flush to w1-0001.parquet (with it deleteFiles).
> # Got Rcoed(id=1,name="b") update event.
> # New writer.
> # Rcoed(id=1,name="b") flush to w2-0001.parquet (with it deleteFiles).
> # Got checkpoint commit event.
> # commit w1-0001.parquet, w2-0001.parquet (with their deleteFiles) metadata
> to snapshot.
> Since it is a new writer, the distribution of data records cannot be correct
> (for example, the distribution of data records to different files by Hash).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
