dataengineervishal opened a new pull request, #4607:
URL: https://github.com/apache/flink-cdc/pull/4607
## What is the purpose of this pull request?
Follow-up to FLINK-40640 (#4536), built on top of FLINK-41057 (#4588).
Since FLINK-40640, every `UPDATE` on a table with equality fields is written
as `DELETE(before)` + `INSERT(after)`. The upsert `INSERT` already makes
Iceberg write an equality delete for the after-image key in the after-image
partition. When neither the identifier key nor a partition source column
changes, that is exactly the row the explicit `DELETE` targets, so every such
update writes a duplicate equality-delete record.
This PR skips the before-image `DELETE` only when the update leaves **both**
of these unchanged:
- **Partition source columns.** Equality deletes are partition-scoped, so a
partition move still needs the `DELETE` in the old partition. This is the
FLINK-40640 case.
- **Identifier (equality) fields of the target table.** The upsert delete
covers only the *new* key. Debezium splits a change to the *source* PK into
DELETE + INSERT, but the sink key can differ from it: a transform with
`primary-keys` ("Reassign primary key"), or a pre-existing table whose
identifier fields differ from the CDC PK (FLINK-41057). In those cases a key
change arrives as a plain `UPDATE`, and skipping the `DELETE` would leave the
old row behind. This follows the review suggestion on #4536.
## Brief change log
- `RowDataUtils`:
- New `convertDataChangeEventToRowData(event, fieldGetters,
supportsEqualityDeletes, deleteKeyGetters)` overload. It skips the before-image
`DELETE` when none of the `deleteKeyGetters` columns changed. `null` keeps the
current behavior, and the existing 3-arg method delegates with `null`.
- New `createDeleteKeyGetters(schema, columns)`. It builds CDC-native
getters, so the comparison works on `BinaryStringData` etc. without the Iceberg
conversions. It returns `null` if a column can't be resolved in the CDC schema
(e.g. a nested identifier field), and the `DELETE` is then always kept.
- `IcebergWriter`: the `TableWriterFactory` cache from #4588 now also holds
the getters for the target table's identifier fields plus partition source
columns. They are computed once per factory and rebuilt with it on schema
change.
- Append-only tables are unchanged.
## Verifying this change
- `RowDataUtilsTest`: added tests for:
- same / changed partition, a multi-column partition, and null ↔ value
- unpartitioned tables
- key change within a partition, and an unpartitioned key change
- binary keys compared by content
- unresolved column → always `DELETE`
- append-only tables ignoring the getters
- INSERT/REPLACE/DELETE are unchanged
- `IcebergWriterTest`:
- several updates of the same key within one checkpoint and across
checkpoints, followed by a delete (partitioned and unpartitioned)
- an update that changes the key keeps only the new row (partitioned and
unpartitioned)
- `IcebergWriterDeleteCapabilityTest`: the target table is keyed on a
different column than the CDC PK, and an update changes the target key but not
the CDC key. Only the new row remains.
- I checked that the 4 key-change writer tests **fail** when only partition
columns are compared (the old row survives as a duplicate).
- All 84 tests in the Iceberg connector module pass, including
`IcebergSinkITCase`.
`IcebergMetadataApplierTest#testApplySchemaChangeHiveCatalog` was excluded
because it fails the same way on master in my local Docker environment.
### Benchmarks (JMH, JDK 21, local; harness not included in this PR)
Conversion only (`RowDataUtils`, ns/op, 5 forks):
| Scenario | 3 columns | 20 columns |
|---|---|---|
| UPDATE, same partition | 89.5 → 62.5 | 805 → 403 (alloc 5136 → 2664 B/op) |
| UPDATE, unpartitioned table | 91.5 → 41.9 | 827 → 391 |
| UPDATE, partition changed | 90.6 → 97.5 | 819 → 800 |
| UPDATE, key changed | ≈ baseline | ≈ baseline |
| INSERT / DELETE | unchanged | unchanged |
Iceberg write path (100k updates through the upsert `TaskWriter` up to
`complete()`, identity-partitioned table, format v2):
| % updates changing partition | equality-delete records | equality-delete
bytes | write time |
|---|---|---|---|
| 0% | 200,000 → 100,000 | 462 KB → 161 KB | within noise |
| 10% | 200,000 → 110,000 | 493 KB → 176 KB | within noise |
| 100% | unchanged | unchanged | unchanged |
Fewer equality deletes also means less merge work for readers and compaction.
## 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?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change the checkbox below to `[X]` followed by the name of the tool, and
uncomment the
"Generated-by" line. See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this
PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed.
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name and Version]
-->
--
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]
