dataengineervishal opened a new pull request, #4535: URL: https://github.com/apache/flink-cdc/pull/4535
## What is the purpose of this pull request? Fixes [FLINK-40640](https://issues.apache.org/jira/browse/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 --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) <!-- Generated-by: Claude Sonnet 5 (Anthropic), via Claude Code --> Generated-by: Claude Sonnet 5 (Anthropic), via Claude Code -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
