spoorthibasu opened a new pull request, #4360:
URL: https://github.com/apache/flink-cdc/pull/4360
**TL;DR:** Fix duplicate rows caused by same-snapshot equality-deletes by
committing per-batch snapshots with increasing sequence numbers.
## Root Cause
When a schema-change event arrives mid-checkpoint, the writer flushes the
affected table before applying the new schema.
This produces two batches within one checkpoint:
- batch 0: pre-schema-change data
- batch 1: post-schema-change data (including equality-delete files
targeting batch 0 rows)
Previously, all batches for a table were merged into a single Iceberg
`RowDelta` and committed as one snapshot.
Iceberg equality-delete files only suppress data with strictly lower
sequence numbers. When data files and equality-deletes are committed in the
same snapshot, they share the same sequence number, so deletes are ineffective.
As a result, both versions of a row remain visible, causing duplicates.
A secondary issue was that `flush(boolean)` rotated all table writers
globally, unnecessarily splitting unrelated tables into multiple batches.
## Fix
### Writer-side (scope reduction)
- `flush(boolean)` no longer rotates task writers globally and becomes a
no-op for non-schema-change paths
- Schema-change events call `flushTableWriter(tableId)`, flushing only the
affected table
- A per-table `batchIndex` increments on each flush and is propagated via
`WriteResultWrapper`
### Committer-side (primary correctness fix)
- Each batch is committed as a separate Iceberg snapshot, giving later
batches strictly higher sequence numbers
- Equality-delete files in batch N correctly supersede data files from batch
M (M < N)
- Two snapshot summary properties are introduced:
- `flink.batch-index`
- `flink.checkpoint-id`
- On retry, the committer scans snapshot history, finds the highest
committed batch index for the checkpoint, and resumes from the next uncommitted
batch
- This ensures idempotent recovery after partial commits
- `MAX_COMMITTED_CHECKPOINT_ID` is written only on the final non-empty
batch, preserving compatibility with Flink checkpoint semantics
## Why `getLastCommittedBatchIndex()` is safe
- Batches are committed in ascending order
- Scanning snapshot history from newest to oldest always encounters the
highest committed batch first
- The scan stops when a snapshot with
`MAX_COMMITTED_CHECKPOINT_ID < checkpointId` is encountered
which marks the boundary of the previously completed checkpoint
- No intermediate batch snapshots for the current checkpoint can appear
before this boundary
## Tests Added
- **testNoDuplicateWhenFlushSplitsSamePkUpdatesWithinCheckpoint**
Verifies `flush(false)` is a no-op and same-PK updates produce a single
correct row
- **testNoDuplicateWhenSchemaChangeFlushSplitsSamePkUpdates**
Verifies schema-change split produces correct dedup via batch ordering
- **testRetryAfterPartialBatchCommit**
Verifies idempotent recovery when batch 0 is already committed and retry
occurs
- **testNoDuplicateWithMultipleSchemaChangesInOneCheckpoint**
Verifies correctness across multiple sequential batches
- **testSchemaChangeFlushDoesNotAffectOtherTable**
Verifies schema-change flush is scoped to the affected table only
## Notes
- The common (non-schema-change) path remains unchanged: one checkpoint to
one batch to one snapshot
- Additional snapshots occur only when a table is split into multiple
batches within a checkpoint, currently triggered by schema changes
- Existing constructor paths default `batchIndex` to `0`, preserving
compatibility with current usage
--
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]
