Repository navigation
[Feature][Connector-V2][Iceberg] Support v3 CDC and replay-safe commits - #12881
goutamadwant wants to merge 8 commits into
Conversation
… commits Use equality deletes for committed rows and Puffin deletion vectors for invalidated rows in new files. Make ordinary and restore commits recognize immutable writer payloads without treating checkpoint numbers as publication proof. Add v2/v3 recovery fixtures for Flink and Zeta, nanosecond equality-delete reader coverage, and matching English and Chinese documentation. Related to apache#12763. Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
|
The red build is due to the existing JDBC |
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for the substantial contribution. I reviewed the complete current diff and traced the writer, commit, recovery, and read paths on the latest head.
What Problem Does This PR Solve?
- User pain point: Iceberg format-version 3 tables reject the connector's former position-delete path for CDC/upsert, and a recovered committable can be replayed through the ordinary commit path.
- Fix approach: select a v3 deletion-vector writer, retain equality deletes for already committed rows, and fingerprint immutable writer payloads in snapshot metadata so recovery can recognize a published payload.
- One-sentence summary: this enables v3 CDC/upsert and makes checkpoint replay identify the payload rather than trusting a reusable checkpoint number.
1. Code Change Review
1.1 Core Logic Analysis
The critical path starts at IcebergWriterFactory.createTaskWriter: delta writes on format v3 select DeletionVectorTaskWriter (IcebergWriterFactory.java:151-177). Its per-partition writer records the position of each newly written key; a replacement or delete in that same writer emits a Puffin deletion vector, while a key not produced by that writer goes to an equality-delete writer (DeletionVectorTaskWriter.java:289-322). The checkpoint then reaches IcebergAggregatedCommitter.commit, which groups committables by checkpoint and invokes IcebergFilesCommitter (IcebergAggregatedCommitter.java:55-76).
Before, v3 delta writes were rejected before any data was produced. After, the committer creates a stable fingerprint from immutable data/delete-file identities, stores it in the committed snapshot, and reuses the same pending-payload check from both normal and restore commits (IcebergFilesCommitter.java:85-160, 281-353). The reader-side change additionally normalizes nanosecond timestamp equality-delete keys before filtering records (IcebergFileScanTaskReader.java:102-146).
Key findings:
- The normal streaming path reaches the new v3 writer whenever identifier fields or upsert mode make the write a delta write.
- The implementation deliberately uses deletion vectors only for rows produced by the current task writer; prior rows remain equality-delete targets.
- Recovery is replay-safe by payload fingerprint, so another job's reused checkpoint number is not treated as proof of publication.
- Legacy snapshots without a fingerprint fail closed on a partial match rather than silently dropping or duplicating a payload.
Runtime path:
Source/transform emits RowKind
-> IcebergWriterFactory.createTaskWriter()
-> format v3 delta write -> DeletionVectorTaskWriter.write()
-> same-writer replacement/delete: Puffin deletion vector
-> prior-row delete: equality delete
-> checkpoint result -> IcebergAggregatedCommitter.commit()
-> IcebergFilesCommitter.pendingResults()
-> fingerprint/history verification
-> atomic AppendFiles or RowDelta commit
-> recovery replay -> ordinary commit() -> same verification
-> IcebergFileScanTaskReader -> equality-delete normalization/filtering
1.2 Compatibility Impact
Fully compatible. Existing v1/v2 write selection and configuration names remain intact. Format v3 changes from a documented early rejection to supported CDC/upsert behavior, while format v4 and later remain explicitly rejected. No API, protocol, serialization format, or default-value change was found. The English and Chinese connector documentation describe the new operational and recovery constraints consistently.
1.3 Performance / Side-Effect Analysis
Normal fingerprint-matched replays and new positive-ID commits avoid a live-manifest scan. Legacy unmatched payloads can read history and, with a broken ancestry chain and a non-positive checkpoint ID, scan live manifests; the documentation now calls out that compatibility cost. The writer keeps only per-key positions for the current task writer, releases writer resources during close/abort, and propagates close/manifest errors rather than reporting a successful commit.
1.4 Error Handling and Logging
No formal source issue found. The writer retains the first close failure and suppresses subsequent close failures, abort cleans up files created by the attempt, and publication-state read failures are propagated. No sensitive values are logged.
2. Code Quality Assessment
2.1 Coding Standards
The new production class has the ASF header and documents its v3 boundary and cleanup behavior. The change is localized to the Iceberg writer, committer, reader, tests, E2E fixtures, and paired documentation.
2.2 Test Coverage and Test Stability
Coverage includes writer-result fingerprinting, reused checkpoint IDs, legacy-history behavior, shared Puffin deletion-vector identities, reader timestamp-nanos equality deletes, and Flink/Zeta recovery scenarios. The recovery E2Es use Awaitility with bounded condition checks rather than fixed sleeps; temporary paths are UUID-scoped and the assertions verify committed rows, snapshots, and deletion-vector metadata.
Stability rating: Stable. IcebergFlinkRecoveryIT.java:215-350 and IcebergZetaRecoveryIT.java:121-177 wait on observed job/checkpoint state and committed rows; no fixed wait, hard-coded host port, or shared mutable test state was found in the changed tests.
2.3 Documentation Updates
Required documentation was updated in both docs/en/connectors/sink/Iceberg.md and docs/zh/connectors/sink/Iceberg.md. The examples and constraints match the implementation, including the partition-key and checkpoint-recovery limitations.
3. Architectural Soundness
3.1 Elegance of the Solution
Long-term solution. The writer selection is constrained by Iceberg table format, and replay protection is stored beside the snapshot publication it proves.
3.2 Maintainability
The writer/committer boundary is explicit: writers create immutable files, while the committer owns publication verification. The legacy path is isolated and documented.
3.3 Extensibility
Versioned snapshot fingerprints permit future fingerprint evolution without reintroducing checkpoint-number identity assumptions.
3.4 Historical-Version Compatibility
Historical snapshots remain readable. Their recovery evidence is evaluated conservatively, and partially published legacy payloads fail rather than being guessed.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| — | No formal source issue found | — | — |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
-
Blockers — must be fixed
- The current Build gate is red in the four
unit-testmatrix jobs (Java 11/17 on Ubuntu and Windows). The Iceberg integration jobs passed. The contributor identifies the failure as the pre-existingJdbcOptionRuleCompletenessTestissue addressed by #12875; that PR is still open and is not indev. Please wait for that fix to enterdev, sync this branch, and rerun the failed unit-test matrix. A sync before #12875 merges will not resolve this particular gate.
- The current Build gate is red in the four
-
Recommended fixes — non-blocking
- None.
Overall assessment: the source change is a well-scoped long-term implementation with strong unit and recovery-E2E coverage. I found no source-side blocker in the current head. Once the external unit-test gate is green, the source is ready from my technical-review perspective. I have read-only review controls here, so this COMMENT does not change GitHub's visible review decision; a write-capable maintainer must perform any final approval action.
|
@goutamadwant thanks for this. I'm the author of #12937 ( In a connector-level test with an unpartitioned, auto-created table, primary key
Reading the table returns both
The v2 case already exists on There is also a recovery integration point. #12937 tags split checkpoints with When combining the changes, I suggest retaining ordered groups and recording the fingerprints of each group's results on that group's snapshot, while preserving publication checks on both ordinary commit and restore. We should test recovery after only the first group is published, replay through ordinary commit, and checkpoint-ID reuse. I haven't validated that combined behavior yet. I'm happy to rebase #12937 onto this PR if that is the preferred landing order. @DanielLeens, does that approach make sense? |
|
Sorry for missing this in my previous review. This is a carry-over omission in my earlier source-clear conclusion, not an issue introduced by a new commit: the head is still I fetched that head and traced the relevant path: That supports the reported stale-row outcome: the new writer no longer knows the pre-schema-change row's file/position, and an equality delete does not remove a data file with the same data sequence number. The Iceberg scan-planning contract requires an older data sequence for an equality delete to apply. Deletion vectors would handle same-commit positions, but the earlier writer's positions are not carried across this reset. The v2 case is pre-existing; enabling the analogous v3 path makes it relevant to this PR as well. I withdraw the earlier CI-only/source-clear conclusion. This is a P0 data-correctness blocker for the schema-change + upsert path when those results reach one commit. I did not run your reproduction locally, and I am not claiming every Zeta schema-change sequence reaches the window. The connector contract is affected, and the Flink sink schema-event handler also calls Yes, preserving ordered groups and placing each group's result fingerprints on that group's snapshot is the right direction. Both ordinary commit and restore must share the publication check; a reused checkpoint number must not identify a different payload as already published. Please retain the exact final-row assertion for v2/v3 across a schema change, and cover a crash after only the first group is published, replay through ordinary commit, and checkpoint-ID reuse. A combined tree has not been validated by this reply. For landing order, please coordinate the two PRs so the ordering fix is present before the newly supported v3 schema-change/upsert combination is mergeable. Merging this PR first while leaving that correctness gap for a later PR would not be safe. A prerequisite/common correctness change, or explicitly keeping the unsupported combination gated until it lands, is preferable. This reply answers the existing counterexample and corrects my prior conclusion; it is not a claim that a fresh full review of all 21 files or a combined v3 E2E has been completed. |
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for the update, @goutamadwant. The head moved from 427c7721 to f391f0d2 only by merging the latest dev (0 commits behind now). I compared the PR's own change set (21 files, 3842 insertions, 373 deletions) against the merge base before and after the merge and it is identical, so nothing in the code has changed in response to the open correctness finding below.
What Problem Does This PR Solve?
Iceberg format-version 3 tables reject the former position-delete path for CDC/upsert, and a recovered committable can be replayed through the ordinary commit path. The PR adds a v3 deletion-vector writer, keeps equality deletes for already committed rows, and fingerprints writer payloads in snapshot metadata so recovery can recognize a published payload.
1. Code Change Review
1.1 Core Logic Analysis
Unchanged from my last review. Delta writes on format v3 select DeletionVectorTaskWriter (IcebergWriterFactory.java); committables flow through IcebergAggregatedCommitter.commit into IcebergFilesCommitter, which verifies the payload fingerprint before an atomic AppendFiles or RowDelta commit.
1.2 Compatibility Impact
Fully compatible for existing v1/v2 behavior and configuration. Format v3 changes from early rejection to supported CDC/upsert. No API or serialization change.
1.3 Performance / Side-Effect Analysis
Unchanged: fingerprint-matched replays avoid a live-manifest scan; the legacy path is documented.
1.4 Error Handling and Logging
Close and publication-state failures are propagated. One correctness issue is open (carried over from the thread, and acknowledged in my comment of 2026-10-22):
Issue 1
- Location:
IcebergRecordWriter.applySchemaUpdate(flush thenresetWriter),DeletionVectorTaskWriterpartition writer state,IcebergFilesCommittersingleRowDeltacommit - Problem: after a schema change the new task writer has no memory of earlier rows' file positions, so
UPDATE_AFTERfor a key written before the schema change becomes an equality delete. Both writer results are committed in one snapshot, so the equality delete has the same data sequence number as the earlier data file and does not apply. @zhengxiang378928908-code reproduced this at the connector level on v2 and v3 (INSERT, schema change, UPDATE_AFTER,complete(), one aggregated commit): the table returns both1=aand1=a2. - Potential risk: silent stale rows (data correctness) for the schema-change plus upsert combination on the newly supported v3 path. The v2 variant already exists on
devand is addressed by #12937. - Best improvement: commit the results of each writer generation as ordered groups (distinct snapshots) so equality deletes apply, record each group's result fingerprints on that group's snapshot, and share the publication check between ordinary commit and restore; coordinate landing order with #12937, or gate the v3 schema-change/upsert combination until the ordering fix is present. Add exact final-row assertions for v2 and v3 across a schema change, plus a crash after only the first group is published, replay through ordinary commit, and checkpoint-ID reuse.
- Severity: High
- Raised by another reviewer: Yes (@zhengxiang378928908-code); I am +1 on the finding.
2. Code Quality Assessment
2.1 Coding Standards
Unchanged: ASF headers and Javadoc are in place.
2.2 Test Coverage and Test Stability
Existing unit and recovery E2E coverage is solid, but none of it covers a schema change followed by an update in the same commit, which is why this slipped through. Stability: Stable for the existing tests; the missing case must be added.
CI for f391f0d2: the fork Build run is still in progress (its changes and Code style jobs passed). The earlier red unit-test legs were the dev regression JdbcOptionRuleCompletenessTest (#12874), and its fix #12875 is still open. Syncing before #12875 lands will not clear that gate.
2.3 Documentation Updates
EN and ZH Iceberg sink docs are updated consistently; they should also state the schema-change/upsert limitation or behavior once the ordering is resolved.
3. Architectural Soundness
The writer/committer split and snapshot-stored fingerprints remain the right design. The open point is that publication ordering across writer generations must be part of that design, and it overlaps with #12937 (seatunnel.commit-group metadata), so the two PRs need one agreed grouping and fingerprint scheme.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Schema change then upsert in one commit leaves stale rows (equality delete not applied) | IcebergRecordWriter.applySchemaUpdate, DeletionVectorTaskWriter, IcebergFilesCommitter |
High |
5. Merge Recommendation
Conclusion: Not recommended for merge
- Blockers - must be fixed
- Issue 1: ordered commit groups (or an explicit gate for v3 schema-change plus upsert) with the final-row and recovery tests listed above, coordinated with #12937.
- CI: wait for the in-progress fork
Buildof this head; thedevJDBC break (#12875) must be resolved before the unit-test matrix can go green.
- Recommended non-blocking fixes
- Document the schema-change/upsert behavior in the sink docs.
Overall assessment: the v3 writer and replay-safe commit design is sound, but I withdrew my earlier source-clear conclusion because of Issue 1. The latest merge changed nothing relevant, so the verdict stays the same until the fix lands.
…replay Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
|
@goutamadwant I reran my connector-level reproduction (INSERT, |
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thank you @goutamadwant for the focused fix in beeda6fac, and @zhengxiang378928908-code for re-running your reproduction on the new head and posting the result. I re-reviewed the full PR at head 311b653abc12cbf2db7188ff5fbe21b8787ceb11, not only the increment. Since my last review, the head gained one real commit (beeda6fac, "Preserve writer generation ordering during checkpoint replay") plus a merge of dev.
The High-severity blocker from my previous review is resolved. The stale-row case (INSERT, schema change, UPDATE_AFTER, one aggregated commit) no longer reaches a single snapshot. IcebergAggregatedCommitter now publishes writer generations as ordered commit groups, so an equality delete from a later generation always gets a newer data sequence number than the rows it must remove. The fix covers both format v2 and v3, and it comes with exactly the tests I asked for: an exact final-row assertion for v2 and v3, a crash after only the first group is published (replayed through both ordinary commit and restore), and checkpoint-ID reuse. My CHANGES_REQUESTED was based on that issue only, so from the source side it can now be cleared.
What Problem Does This PR Solve?
- User pain point: format-version 3 Iceberg tables rejected CDC/upsert writes, a recovered committable could be published twice through the ordinary commit path, and (on
devalready, for v2) a schema change inside a checkpoint could leave stale rows. - Fix approach: a v3
DeletionVectorTaskWriter(Puffin deletion vectors for rows the current writer produced, equality deletes for older rows); versioned payload fingerprints stored in the published snapshot and checked by both ordinary commit and restore; and now ordered commit groups across writer generations. - One-sentence summary: v3 CDC/upsert becomes supported, with replay protection based on the payload and with correct delete ordering across schema changes.
1. Code Change Review
1.1 Core Logic Analysis
New in beeda6fac (IcebergAggregatedCommitter.java:93-142):
// before (dev and the previous head): every result of a checkpoint in one snapshot
filesCommitter.doCommit(flattenResults(commitInfos), checkpointId);
// after: round r = r-th WriteResult of each writer (one per writer generation)
for (round ...) {
results = r-th result of every writer
if (round has an equality delete with recordCount > 0 && !pending.isEmpty()) {
groups.add(pending); pending = new ArrayList<>(); // start a new snapshot
}
pending.addAll(results);
}
// groups.size() == 1 -> doCommit(results, id) (unchanged summary)
// otherwise -> doCommit(group, id, g, groupCount) (+ seatunnel.commit-group[-count])Runtime path:
IcebergSinkWriter.applySchemaChange / auto schema evolution in write()
-> IcebergRecordWriter.applySchemaUpdate: flush() saves generation g, resetWriter() starts g+1
prepareCommit(checkpointId) -> IcebergRecordWriter.complete() -> [g0, g1, ...] in one IcebergCommitInfo
-> (checkpoint state; serialized commit info)
IcebergAggregatedCommitter.commit / restoreCommit (same path)
-> per checkpoint (LinkedHashMap, :58-67) -> orderedGroups()
-> for each group: IcebergFilesCommitter.pendingResults() (fingerprints of published results are skipped)
-> AppendFiles or RowDelta with seatunnel.commit-ids for that group's results
Correctness analysis of the grouping, under the documented key-affinity constraint (all changes of a key go to one writer, in source order):
- Any equality delete in round r is published in a snapshot strictly after all data of rounds
< r. So it now applies to the row that the earlier generation of the same writer wrote. This is the case that was broken. - Later rounds without equality deletes are merged into the same snapshot as round r. Their data shares round r's sequence number, so round r's equality deletes do not remove it, which is correct (that data is newer).
- Position deletes / deletion vectors only reference files of their own generation, so they are valid in any snapshot that also contains those files.
- Writers with fewer generations (for example, no rows before the schema change) put their only result into round 0. Their keys are disjoint from other writers' keys by the affinity constraint, so this is safe.
testWriterGenerationsRemainOrderedAcrossUnevenWorkerspins this behavior. - Replay: fingerprints are computed per result, and each group is atomic. After a crash between groups, replaying the same commit info skips group 0 (all its results are found in history) and publishes the remaining groups.
recoversAfterPublishingOnlyFirstWriterGenerationshows this for v2 and v3, through both ordinary commit and restore. Group numbers are written but never read for publication decisions, so a reused checkpoint ID cannot make a different payload look published. - Multiple checkpoints in one aggregated commit now get one snapshot each, in list order. On
devthey shared a single snapshot, which had the same same-sequence-number weakness. This is a side improvement.
The rest of the PR is unchanged since my first review: v3 writer selection in IcebergWriterFactory (format > 3 still rejected), the removal of the old v3 rejection guards in IcebergWriterFactory / SchemaUtils (intended, because the PR enables the feature), the fingerprint-based pendingResults, and the nanosecond equality-delete normalization in IcebergFileScanTaskReader.
1.2 Compatibility Impact
Partially incompatible, and now documented in docs/en|zh/introduction/concepts/incompatible-changes.md. For v2 and v3 CDC/upsert jobs, a checkpoint that contains a schema change can now produce several ordered snapshots instead of one. External readers can see a prefix of that checkpoint between groups. Single-generation and append-only checkpoints keep exactly one snapshot, and the summary keys are unchanged for them (the group properties are added only when groupCount > 1). No option, default, public SPI or checkpoint serialization format changes (IcebergCommitInfo and WriteResult are untouched). Old snapshots stay readable, and legacy, non-fingerprinted history is still evaluated fail-closed.
1.3 Performance / Side-Effect Analysis
orderedGroups is linear in the number of results and delete files of one checkpoint, with no I/O. The extra snapshots appear only for checkpoints with schema changes, so the commit count stays proportional to schema-change frequency, not to data volume. Each extra group runs one more pendingResults history walk. That walk reads snapshot summaries in memory, and the manifest reads are bounded as described in my first review. No new threads, buffers or retries.
1.4 Error Handling and Logging
doCommit(..., group, groupCount) validates its arguments. A failure between groups propagates, and replay completes the remaining groups. The partial-publication guard (alreadyPublished) still fails closed. No sensitive data is logged.
- Medium: the new grouping contract has no method-level documentation. Location:
IcebergAggregatedCommitter.java:104(orderedGroups),IcebergFilesCommitter.java:86(new publicdoCommit(results, checkpointId, group, groupCount)),:63-64(the new summary constants). Problem: correctness depends on three non-obvious assumptions. The r-th result of a writer is its r-th writer generation, keys are writer-affine, and the group properties are diagnostic only. Today these are spread over one inline comment and the docs. Risk: a future change toIcebergRecordWriter.complete()(for example theflush_row_sizerotation in #12937, which also emits several results per writer) could break the "index = generation" assumption without anyone noticing. Best improvement: a short Javadoc onorderedGroupsand the newdoCommitoverload that states these assumptions, and a one-line comment on the two constants saying they are never used as proof of publication. Raised by another reviewer: no. - Low: the order of multiple checkpoints in one aggregated commit is taken from list order. Location:
IcebergAggregatedCommitter.java:58-67. Problem: checkpoints are committed in insertion order. If an engine ever handed over commit infos with checkpoint N+1 before N, N+1's equality deletes would be published first. I did not find a path in Zeta or the Flink translation that does this, andtestNativeRebatchingRetainsCheckpointBoundariescovers the ascending case. This is a hardening suggestion, not a finding of a live bug. Best improvement: commit positive checkpoint IDs in ascending order (for example, aTreeMapwhen every ID is positive), plus a reversed-input test. Raised by another reviewer: no. - Low: the PR description no longer matches the code. Location: PR description. Problem: it still says
incompatible-changes.mdis "not applicable", and it does not mention that v2 jobs can now get several snapshots per checkpoint. Best improvement: update the checklist and add one sentence about ordered commit groups for v2 and v3. Also mention the overlap with #12937 (sameseatunnel.commit-groupkeys and the same committer files), so maintainers can choose the landing order. Raised by another reviewer: no.
2. Code Quality Assessment
2.1 Coding Standards
The new production code is small and readable, and the inline why-comment in orderedGroups is accurate. Item 1 covers the missing method docs. A minor style point: fully qualified names are used inside method bodies (org.mockito.InOrder at IcebergAggregatedCommitterTest.java:82,196, org.apache.iceberg.DeleteFile at IcebergSinkIT.java:196) where a normal import would do.
2.2 Test Coverage and Test Stability
preservesUpsertOrderingAcrossSchemaChange(IcebergFormatVersionTest.java:1097, v2 and v3): one real writer, INSERT,applySchemaChange, UPDATE_AFTER, then a real commit against a local catalog. It asserts exactly one live row with the new values, two snapshots and the group properties. Then it replays through commit and restore without creating new snapshots, and checks that a different payload under the same checkpoint ID is still published. If the grouping is reverted to a single snapshot, this test reads two live rows and fails.recoversAfterPublishingOnlyFirstWriterGeneration(:1145, v2 and v3 x ordinary/restore) makes the secondnewRowDelta()fail. It asserts that only the first group is visible, and that replay publishes exactly the missing group once.testWriterGenerationsRemainOrderedAcrossUnevenWorkersandtestAppendOnlyWriterGenerationsKeepSingleCommitpin the grouping and the single-snapshot fallback (exactdoCommitarguments, in order).- Removed tests: the old checkpoint-ID-based
isAlreadyCommitted*and restore-skip tests were replaced by stronger, real-table tests of the payload-based semantics (15 new cases inIcebergFilesCommitterTest). That is consistent with the intended change, since a checkpoint number is no longer proof of publication.IcebergSinkParallelCommitITonly changed comments and messages, and its>= 2snapshot assertion is unchanged. The v3 rejection E2E became a positive v3 CDC E2E with exact rows and deletion-vector metadata. I did not find any weakened assertion. - Stability rating: Stable. The new unit tests use a temporary local catalog and deterministic Mockito spies, with no sleeps or ports. The existing Flink/Zeta recovery E2Es wait on observed state.
2.3 Documentation Updates
The EN and ZH sink pages and the incompatible-changes.md entries are updated consistently and match the code. They cover ordered groups for v2 and v3, the visible prefix between groups, the single-snapshot cases, and the note that group numbers are not publication proof. The earlier "checkpoints commit everything together" sentence was correctly removed.
3. Architectural Soundness
- 3.1 Elegance: Long-term solution. Ordering is fixed where publication happens (the aggregated committer), and replay safety remains payload-based. Nothing is added to the engine or the API.
- 3.2 Maintainability: good, once item 1 documents the "result index = writer generation" contract.
- 3.3 Extensibility: #12937 (
flush_row_size) solves the same ordering problem for memory-driven rotation, with the same summary keys. The two PRs now overlap directly, and the second one to land will need to adopt the other's grouping. I recommend that maintainers decide the landing order explicitly. This PR is now self-contained and safe to land first. - 3.4 Historical-Version Compatibility: checkpoint state classes are unchanged, so in-flight checkpoints from the previous version restore and replay through the new path. Old snapshots without fingerprints keep the fail-closed legacy check.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Grouping contract (index = generation, key affinity, group keys diagnostic) not documented on the new methods | IcebergAggregatedCommitter.java:104, IcebergFilesCommitter.java:63-64,86 |
Medium |
| 2 | Multiple checkpoints in one aggregated commit are ordered by list order, not checkpoint ID (hardening) | IcebergAggregatedCommitter.java:58-67 |
Low |
| 3 | PR description outdated (incompatible-changes, v2 multi-snapshot, #12937 overlap) | PR description | Low |
| - | Previous Issue 1: schema change then upsert in one commit left stale rows | IcebergAggregatedCommitter, IcebergFilesCommitter |
Fixed in beeda6fac |
5. Merge Recommendation
CI: the fork Build for this head (run 40077518263) and the apache-side Build check are both still queued, so there is no result for 311b653ab yet. The branch is level with dev (ahead 4, behind 0), with no conflicts. Please let the run finish. If a lane fails, tell me which one, and I will check whether it relates to this diff.
Conclusion: Ready to merge
- Blockers (must fix): none from the source, tests or docs. No blocking issues found. The remaining gate is a green
Buildon this head. - Recommended fixes (non-blocking): item 1 (Javadoc on the grouping contract), item 3 (PR description), and optionally item 2.
Overall assessment: the v3 writer, payload-based replay protection and the new ordered publication fit together correctly, and the fix closes the data-correctness gap for v2 and v3 together. Residual risks: the documented visible prefix between groups, the key-affinity requirement for partitioned upserts, and recovery after the needed snapshot history has expired (documented). Nothing was built or executed locally. This review is based on the full diff, the call chain at this head and the reproduction reported on this head. I will move my review state once CI is green.
Better alternative: none for this PR. Longer term, a single shared "commit group" helper used by both this PR and #12937 would avoid two slightly different grouping rules for the same summary keys.
|
@davidzollo @DanielLeens Could you approve this workflow to run too? changes requseted are addressed and verified locally. Pls review as well. Thanks! |
|
Thanks @goutamadwant for the quick follow-up and for the fix in Two things are needed on your side before that can happen:
I cannot approve or trigger workflow runs from my side as a reviewer, so a maintainer with write access would need to do that if the run stays pending. The non-blocking suggestions from my review (Javadoc on the grouping contract, refreshing the PR description) can be picked up in the same push. Thanks again for the patience and the thorough tests! |
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks again, @goutamadwant, for the quick follow-up on my last round! The head moved from 311b653 (the version I last reviewed in full) to 308f17b. I compared that range: it contains five upstream dev commits that arrived through the merge (CSV preamble, Kafka tombstone, nested row byte accounting, nullable Avro ROW layout, Zeta metrics reset), the merge commit itself, and one PR-owned commit, [Docs][Iceberg] Explain checkpoint group ordering and replay identity, which only adds Javadoc/comments in IcebergAggregatedCommitter.java and IcebergFilesCommitter.java. I re-verified the Iceberg code path and the full PR diff against the new head, and none of the merged dev changes touch the Iceberg connector. The PR also no longer conflicts with dev (MERGEABLE, 0 behind of its own merge base for the Iceberg files). This is a code-reading review only; nothing was executed locally.
What Problem Does This PR Solve?
- User pain point: Iceberg format-version 3 tables rejected CDC/upsert writes; a recovered committable could be re-published through the ordinary commit path; and a schema change inside one checkpoint could leave stale rows because an equality delete shared the data sequence number of the rows it had to remove (v2 and v3).
- Fix approach: a v3
DeletionVectorTaskWriter(Puffin deletion vectors for rows produced by the current writer, equality deletes for older rows); versioned payload fingerprints stored in the published snapshot and checked by both ordinary commit and restore; and ordered commit groups per writer generation. - One sentence: v3 CDC/upsert becomes supported, with replay protection that identifies the payload rather than trusting a reusable checkpoint number, and with correct delete ordering across schema changes.
1. Code Change Review
1.1 Core Logic Analysis
Key logic (unchanged since my last full review, now with explanatory docs):
// IcebergAggregatedCommitter: round r = r-th WriteResult of each writer (one per writer generation)
for each round:
if (round has equality delete with recordCount > 0 && !pending.isEmpty()) { groups.add(pending); pending = new ArrayList<>(); }
pending.addAll(round results);
// 1 group -> doCommit(results, id) (unchanged single snapshot)
// n groups -> doCommit(group, id, g, n) (+ seatunnel.commit-group[-count], diagnostic only)Runtime path:
IcebergSinkWriter.applySchemaChange -> IcebergRecordWriter.applySchemaUpdate: flush() saves generation g, resetWriter() starts g+1
prepareCommit -> complete() -> [g0, g1, ...] in one IcebergCommitInfo (checkpoint state)
IcebergAggregatedCommitter.commit / restoreCommit (same path) -> orderedGroups()
-> per group IcebergFilesCommitter.pendingResults() (fingerprints of already published results are skipped)
-> AppendFiles or RowDelta, one snapshot per group
Correctness holds under the documented key-affinity requirement: an equality delete in round r is published after all data of earlier rounds so it removes them; later rounds without equality deletes share the sequence number and are correctly not removed; deletion vectors only reference same-generation files; replay after a crash between groups skips the already published group by payload fingerprint and publishes the rest; group ordinals are never used as proof of publication. The new Javadoc now states exactly these assumptions (result-list index = writer generation, key affinity, group keys are diagnostics), which resolves my previous Issue 1. The normal CDC/upsert streaming path reaches this logic whenever a delta write is used.
1.2 Compatibility Impact
Partially incompatible (documented). For v2 and v3 CDC/upsert jobs, a checkpoint that spans writer generations can now publish several ordered snapshots instead of one, so external readers can observe a committed prefix between groups. Single-generation and append-only checkpoints keep a single snapshot with unchanged summary keys. No option, default, public SPI or checkpoint serialization format changes (IcebergCommitInfo/WriteResult untouched); old snapshots stay readable and legacy, non-fingerprinted history is evaluated fail-closed. docs/en|zh/introduction/concepts/incompatible-changes.md and the PR checklist now describe this correctly, which resolves my previous Issue 3 (the description also mentions the #12937 overlap).
1.3 Performance / Side-Effect Analysis
Grouping is linear in the results of one checkpoint with no I/O. Extra snapshots only appear for checkpoints containing a generation change (schema change), so the commit count tracks schema-change frequency, not data volume. Each additional group performs one more in-memory history walk in pendingResults; manifest reads stay bounded as described in my earlier reviews. No new threads, retries or unbounded buffers.
1.4 Error Handling and Logging
A failure between groups propagates and replay finishes the remaining groups; doCommit validates group arguments; the partial-publication guard fails closed; nothing sensitive is logged.
- Issue 1 (Low, hardening, carried over from my last review, still open): multiple checkpoints inside one aggregated commit are committed in list/insertion order (
IcebergAggregatedCommitter.java:58-67), not by ascending checkpoint ID. I found no Zeta or Flink translation path that hands them over out of order andtestNativeRebatchingRetainsCheckpointBoundariescovers the ascending case, so this is not a live bug. Best fix: sort by checkpoint ID when all IDs are positive and add a reversed-input test. Raised by another reviewer: no. - Issue 2 (Low, style): fully qualified names are used inside method bodies (
org.mockito.InOrderatIcebergAggregatedCommitterTest.java:82,196,org.apache.iceberg.DeleteFileatIcebergSinkIT.java:196) where a normal import would do. Raised by another reviewer: no.
2. Code Quality Assessment
2.1 Coding Standards
ASF headers and Javadoc are in place; the new method-level docs and the one-line comment on the group-property constants are accurate. See Issue 2 for the remaining style nit.
2.2 Test Coverage and Test Stability
Rating: Stable.
preservesUpsertOrderingAcrossSchemaChange(IcebergFormatVersionTest.java, v2 and v3): real writer, INSERT, schema change, UPDATE_AFTER, real commit against a local catalog; asserts one live row with new values, two snapshots, group properties, replay through commit and restore without new snapshots, and that a different payload under a reused checkpoint ID is still published. Reverting the grouping makes it read two live rows and fail. The independent reproduction by @zhengxiang378928908-code on311b653(v2 and v3 now return only1=a2) matches this.recoversAfterPublishingOnlyFirstWriterGeneration(v2/v3, ordinary/restore) fails the secondnewRowDelta()and checks that replay publishes exactly the missing group once.testWriterGenerationsRemainOrderedAcrossUnevenWorkersandtestAppendOnlyWriterGenerationsKeepSingleCommitpin grouping and the single-snapshot fallback.- Removed checkpoint-ID-based
isAlreadyCommitted*tests were replaced by stronger real-table tests of the payload-based semantics; I did not find a weakened assertion. The v3 rejection E2E became a positive v3 CDC E2E with exact rows and deletion-vector metadata. The new tests use temporary local catalogs and deterministic spies (no sleeps or fixed ports); the Flink/Zeta recovery E2Es wait on observed state.
2.3 Documentation Updates
English and Chinese Iceberg sink docs and both incompatible-changes.md files are updated consistently and match the code (ordered groups for v2/v3, visible prefix, single-snapshot cases, group numbers are not publication proof).
3. Architectural Soundness
3.1 Elegance
Long-term solution. Ordering is fixed where publication happens (the aggregated committer), replay safety stays payload-based, and nothing is added to the engine or public API.
3.2 Maintainability
Good: the "result index = writer generation" contract is now documented on the grouping method.
3.3 Extensibility
#12937 (flush_row_size) addresses ordering for memory-driven rotation with the same summary keys and committer files, so the overlap is real. The PR description says whichever lands second must reconcile with the first; this PR is self-contained and safe to land first.
3.4 Historical-Version Compatibility
Checkpoint state classes are unchanged, so in-flight checkpoints from a previous version restore and replay through the new path; pre-fingerprint snapshots keep the fail-closed legacy check.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Multiple checkpoints in one aggregated commit ordered by list order, not checkpoint ID (hardening, carried over) | IcebergAggregatedCommitter.java:58-67 |
Low |
| 2 | Fully qualified names inside method bodies | IcebergAggregatedCommitterTest.java:82,196, IcebergSinkIT.java:196 |
Low |
| - | Previously: stale rows after schema change plus upsert (High); undocumented grouping contract; outdated PR description | - | Fixed |
5. Merge Recommendation
Conclusion: Ready to merge
Blockers (must fix): none from source, tests or docs. My earlier CHANGES_REQUESTED was based only on the stale-row issue, which is fixed and covered, so from the source side it can be cleared.
Recommended fixes (non-blocking): Issue 1 (optional hardening) and Issue 2 (imports).
CI status for head 308f17b: the apache-side Build check is still in progress (run), so there is no result yet. The previous unit-test failure on this PR was the unrelated dev JDBC test break, and it is fixed in dev now. No further sync is needed. I will move my review state to approve once Build is green on this head; a maintainer with write access must still give the formal approval (REVIEW_REQUIRED).
Overall assessment: the v3 writer, payload-based replay protection and ordered publication fit together correctly and close a data-correctness gap for both v2 and v3. Residual risks, all documented: the visible prefix between groups, the key-affinity requirement for partitioned upserts, and recovery after the required snapshot history has expired. Better alternative: none for this PR; longer term a single shared commit-group helper for this PR and #12937 would avoid two slightly different grouping rules for the same keys.
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks again for the contribution, @goutamadwant. The CI fact in my previous review has changed: the Build check (run) is now SUCCESS on the unchanged head 308f17b81cbf.
In my last review CI was the only remaining gate and I had no outstanding source-level blocker, so I am changing my earlier comment-only review to an approval. The final merge still needs a write-capable committer. Thanks for the careful work!
|
Thanks for the contribution. The merge-gate fact cited in my previous conclusion has changed: GitHub now reports a real merge conflict. This follow-up corrects that gate on the unchanged reviewed head; it is not another full source review or a sync request based only on being behind. Gate snapshot (2026-10-28T15:53:58+00:00), head CI is green, but that does not remove the separate conflict gate below. My previous source conclusion and any independent source requests remain as recorded; no new effective approval or dismissal is performed by this clarification. Separately, GitHub reports merge conflicts with |
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Sorry for missing the remaining core-state comment coverage in the last review. Issue 1 below is a carryover omission, not a new regression introduced by this head. I re-reviewed the complete 24-file diff at 029ad39 and the actual writer, committer, reader and engine-adapter callers. The earlier stale-row blocker remains resolved. The previous conflict/CI guidance is superseded: this head is mergeable and Build is green, including the recovery executions detailed below.
What Problem Does This PR Solve?
User pain point: Iceberg v3 replaces position-delete files with deletion vectors, while replay and schema-driven writer rotation must preserve CDC updates without duplicating old rows.
Fix approach: Select a v3 writer that emits Puffin vectors for its own new data files and equality deletes for prior rows; publish writer generations in order and identify committed immutable payloads in snapshot summaries.
One-sentence summary and example: INSERT(1,a), a schema change, then UPDATE_AFTER(1,b) is published in ordered snapshots so only the new row remains visible after completion and replay.
1. Code Change Review
1.1 Core Logic Analysis
IcebergSinkWriter.write reaches IcebergRecordWriter and format-based writer selection. Schema updates flush the old task writer and reset it, producing chronological WriteResult entries. DeletionVectorTaskWriter remembers the file and row position of current-writer keys; replacements remove those positions with DVs, while absent keys become equality deletes. IcebergAggregatedCommitter groups by checkpoint, then result-list generation, starting a new snapshot group before later equality deletes. Both ordinary and restore commits use pendingResults, which walks the selected branch and recognizes versioned per-result fingerprints. Legacy candidates require actual file identities, never just a checkpoint number.
Before:
// v3 delta writer rejected; generation results flattened into one snapshot
// checkpoint-number-only recovery could confuse different jobsAfter:
if (deltaWrite && formatVersion == 3) return deletionVectorWriter;
// publish ordered generations, recording immutable result fingerprints
// restoreCommit delegates to the same replay-safe commit pathKey Findings
- Position capture precedes RollingDataWriter.write, which can roll after a row; the 2,100-row fixture checks this boundary.
- Deep copies stabilize retained binary/nanosecond keys; reader-side normalization applies only when equality keys contain TIMESTAMP_NANO.
- Fingerprints include the referenced data file and Puffin blob offset/length, distinguishing vectors sharing one physical file.
- Duplicate results are removed, mixed published/new results retain only new units, and a partially matched legacy unit fails closed.
- Group ordinal/count are diagnostics, not proof of publication. Append-only and single-generation checkpoints retain one snapshot.
- No aggregate abort deletes an already-published prefix; writer abort only removes attempt-owned new files.
Complete runtime path
RowKind -> IcebergSinkWriter -> IcebergRecordWriter -> v2 writer / v3 DeletionVectorTaskWriter
-> schema update: flush old writer, reset -> chronological WriteResult list
-> prepareCommit(checkpointId) -> engine combine -> IcebergAggregatedCommitter
-> checkpoint grouping -> ordered writer generations -> IcebergFilesCommitter
-> branch/history fingerprint check -> pending files -> AppendFiles or RowDelta
-> crash/replay through ordinary or restore commit -> skip published units
-> source scan -> GenericDeleteFilter + normalized nanos equality keys -> residual/projection
The source resolves the independently reproduced schema-change stale-row case for both v2 and v3. Later equality deletes get a newer sequence number than preceding generations. Actual Flink 1.20 calls ordinary commit after combining recovered committables, so sharing the check with restoreCommit is necessary. Different jobs reusing checkpoint 1 are distinguished by files. Correctness remains conditional on documented key affinity, per-key order, retained evidence and fencing of overlapping attempts. No general atomic multi-snapshot checkpoint or HA-failover guarantee is inferred.
1.2 Compatibility Impact
Partially incompatible, explicitly documented. v2 remains the default and retains its writer selection, but both v2/v3 CDC checkpoints spanning generations may now expose a committed prefix across several snapshots. EN/ZH incompatible-changes documents explain this. Public options, plugin registration, distribution entries and checkpoint state classes remain unchanged. v3 moves from rejection to supported bounded CDC/upsert behavior; formats above v3 remain rejected for delta writes.
1.3 Performance / Side-Effect Analysis
The writer retains per-key positions per active partition until completion, plus deletion vectors and owned file paths; memory scales with checkpoint cardinality. Each group walks retained snapshot ancestry and computes fingerprints; legacy candidates may read manifests, with a live-file fallback only for unknown IDs and broken ancestry. This avoids a live-table scan for normal fingerprinted positive-ID commits, but history length still matters. No throughput or bounded-memory claim is made.
1.4 Error Handling and Logging
Write/close failures are sticky, subsequent complete fails, secondary cleanup errors are suppressed, and partial Puffin output is tracked for abort. Manifest-read failures and partial legacy publication propagate instead of guessing. Failure between groups leaves a visible prefix that replay completes. Fingerprint checking is not a cross-process lock; engine fencing is a documented prerequisite.
Issue 1: Complete comments for the new replay and writer-state invariants
- Location:
seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/writer/DeletionVectorTaskWriter.java:75-79,281,289-321; sink/commit/IcebergFilesCommitter.java:110-181 - Problem description: The new rows map, createdFiles/dvOutput/failure lifecycle and PartitionWriter.insert/delete lack explanatory comments. pendingResults has useful branch comments, but no complete method contract for retained-history scope, legacy partial matches and concurrency/fencing. The class/grouping comments do not explain every nontrivial state field and core method required by this review protocol. This was present and omitted in the previous review.
- Potential risk: A later cleanup, rolling or replay change can violate ownership, sticky-failure or publication-proof assumptions without recognizing the invariant.
- Best improvement: Add concise comments explaining current-writer-only key positions, why position is captured before the rolling write, DV versus equality-delete selection, attempt-owned file cleanup and sticky failure. Document pendingResults input/output, retained branch-history boundary, partial-match failure and external fencing requirement. No functional rewrite is requested.
- Severity: Medium; blocking.
- Raised by another reviewer: No
Issue 2: Keep the checkpoint-ordering assumption explicit
- Location:
IcebergAggregatedCommitter.java:58-67 - Problem description: Multiple checkpoint groups preserve incoming insertion order rather than sorting positive checkpoint IDs. I did not establish an actual caller delivering reversed IDs, so this remains optional hardening rather than a demonstrated correctness defect.
- Potential risk: A future adapter that changes ordering could publish a later checkpoint first.
- Best improvement: Document the adapter ordering precondition or add explicit positive-ID ordering with a real-table reversed-input regression if that input is intended to be supported.
- Severity: Low; non-blocking.
- Raised by another reviewer: Carried forward from the earlier Daniel review; no duplicate independent finding.
Issue 3: Use normal imports in the expanded fixtures
- Location:
IcebergAggregatedCommitterTest.java:82,196; IcebergSinkIT.java:196 - Problem description: Several method bodies use fully qualified Mockito and Iceberg types despite no name conflict.
- Potential risk: Minor readability cost in already substantial recovery fixtures.
- Best improvement: Use ordinary imports when touching these lines.
- Severity: Low; non-blocking.
- Raised by another reviewer: No
2. Code Quality Assessment
2.1 Coding Standards
The grouping Javadoc and diagnostic-summary comments are accurate, ASF headers are present and no dependency is added. Issue 1 narrows the remaining required comment work; it does not reopen the resolved grouping-contract discussion. Immutable file identities and existing state classes avoid a new serialized checkpoint protocol.
2.2 Test Coverage and Test Stability
Test stability: Stable. IcebergFormatVersionTest checks exact rows and duplicate keys, all three data formats, rolling, abort/partial-Puffin failures, schema/partition changes and ordinary/restore replay. recoversAfterPublishingOnlyFirstWriterGeneration asserts the exposed prefix and exact final rows/snapshot count. Real-table committer tests replace obsolete number-only expectations; delegate mocks supplement those tests rather than being sole correctness evidence. IcebergFlinkRecoveryIT:215-350 uses bounded state polling, verifies a different worker registration and original checkpoint restoration, and completes new checkpoints. IcebergZetaRecoveryIT:121-177 checks actual restored offsets and new rows after SIGKILL. Paths are unique and waits are bounded; the intentional polling sleeps are not blind success criteria. Spark is excluded from CDC row-kind testing with a stated capability reason. No local execution was performed.
Local analysis: fetched the head, created independent branch seatunnel-review-12881, read the complete 24-file diff and traced affected callers with git show/git grep. No local compilation, unit tests, integration tests, E2E, containers, or job execution was performed, as requested. These are source-analysis conclusions; execution evidence is the current-head CI below.
2.3 Documentation Updates
Both sink guides describe v3 writer boundaries, retained recovery evidence, partition/key-affinity requirements and visible ordered prefixes. Both incompatible-changes files correctly include v2 as affected. The PR body labels earlier local recovery evidence as historical; the current hosted evidence below is stronger for the checked matrix and should be kept distinct from HA/cloud/Spark coverage.
3. Architectural Soundness
3.1 Elegance of the Solution
Long-term solution. Publication ordering belongs in the Iceberg aggregated committer, while immutable payload verification belongs in the file committer. The connector-specific timestamp normalization avoids changing unrelated readers. The minimal remaining change is comment completion. Coordinate the overlapping grouping/summary-key implementation with #12937 when either lands, without importing an unmerged prerequisite.
3.2 Maintainability
Keep generation ordering, payload identity and cleanup ownership separate. A sequence-number or group-number shortcut would reopen the old failure. The new core state deserves the comments requested in Issue 1; no framework extraction is necessary for this PR.
3.3 Extensibility
Versioned fingerprint summaries leave an explicit evolution boundary. Existing-DV rewriting, row lineage, Variant/default values and broader v3 features remain separate slices; this PR does not establish them.
3.4 Historical-Version Compatibility
No checkpoint-state class changes were found. Legacy publication checks are preserved conservatively, but unchanged serialization alone is not proof that every old binary checkpoint restores. Evidence-expired recovery and overlapping unfenced attempts remain outside the documented guarantee.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Complete comments for the new replay and writer-state invariants | seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/writer/DeletionVectorTaskWriter.java:75-79,281,289-321; sink/commit/IcebergFilesCommitter.java:110-181 |
Medium |
| 2 | Keep the checkpoint-ordering assumption explicit | IcebergAggregatedCommitter.java:58-67 |
Low |
| 3 | Use normal imports in the expanded fixtures | IcebergAggregatedCommitterTest.java:82,196; IcebergSinkIT.java:196 |
Low |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
- Blockers — must be fixed
- Issue 1: Complete comments for the new replay and writer-state invariants Complete the small comment-only requirement in Issue 1 and retain green checks on the resulting head; effective maintainer approval remains required.
- Recommended fixes — non-blocking
- Issue 2: Keep the checkpoint-ordering assumption explicit; Issue 3: Use normal imports in the expanded fixtures
Overall assessment: The former data-correctness issue is resolved and current hosted recovery tests pass. The remaining source request is documentation of core invariants, explicitly acknowledged as my earlier omission.
Existing-review reconciliation: +1 to @zhengxiang378928908-code on the original stale-row counterexample and combined grouping/fingerprint approach. orderedGroups plus the v2/v3 exact-row and partial-publication replay tests resolve it; no duplicate High issue is raised. The earlier grouping-contract and incompatible-change documentation requests are addressed. @nzw921rx's historical conditional LGTM is not a current-head maintainer approval. The earlier conflict at 308f17b is no longer the current gate, and historical Daniel approval does not cover this new head. Author replies have been incorporated into this single full review.
Head 029ad3935b02345559e175b427e6a486f8c54ecf. Live base dev at 3ac2d1cec6aa7b3741c4dacdd23020f778e23c02; comparison ahead, ahead 7 / behind 0; merge state BLOCKED. Behind/diverged alone is recorded metadata, not a request to synchronize upstream. Checks: Signal pull request update = SUCCESS; Build = SUCCESS
Current Build is SUCCESS, backed by fork run 41017562419. All four unit-test matrix lanes and both Iceberg integration lanes passed. The JDK 17 Iceberg job log explicitly reports IcebergFlinkRecoveryIT 8/0/0/0, IcebergZetaRecoveryIT 2/0/0/0, IcebergSinkParallelCommitIT 11/0/0/0 and IcebergSinkIT 35/0/0/0 (tests/failures/errors/skips), followed by BUILD SUCCESS. Skipped unrelated lanes are not counted as executed coverage. No new upstream synchronization request is justified by branch age alone.
Daniel has READ repository permission. This COMMENT records the technical conclusion; it is not an effective maintainer approval or a dismissal of any historical requested-changes entry. A write-capable maintainer owns the final repository approval gate.
|
Thanks for keeping this contribution moving. I need to update the merge-gate information on unchanged head Please sync the latest The existing comment requirement for the replay and writer-state invariants remains open; the previously non-blocking recommendations stay non-blocking. The preceding complete source review remains the source assessment; this is a gate clarification, not another full review or a new list of source findings. Current existing-head checks: Build=SUCCESS. Compare |
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for the quick follow-up, @goutamadwant. I re-read the complete current diff at e9a942f8f35 (24 files, +4097/-386 against dev) together with the single commit since my last full pass (029ad393 -> e9a942f8: a comment-only change plus a merge of current dev). The conflict I reported is gone and the comment request from my last review is addressed. One carryover below is mine: I missed it in earlier rounds.
Sorry about that: Issue 1 is a carryover omission from my previous reviews of the replay-fingerprint design, not something introduced by your latest commit. I should have measured the snapshot-summary size earlier.
What Problem Does This PR Solve?
- User pain: Iceberg format v3 replaced position-delete files with deletion vectors, so CDC/upsert into a v3 table was rejected. Separately, a schema change inside a checkpoint reset the task writer, and the new writer's equality delete shared a data sequence number with the earlier writer's data file, leaving the stale row visible (reproduced independently on v2 and v3 by @zhengxiang378928908-code).
- Fix approach: A v3 delta writer that records the position of each key in its own new files (Puffin deletion vectors) and writes equality deletes for older rows; ordered publication groups per writer generation so later equality deletes land in a newer snapshot; and replay-safe commits that recognize already-published results by immutable payload fingerprints in the snapshot summary, shared by ordinary commit and restore.
- One sentence: INSERT(1,a), schema change, UPDATE_AFTER(1,b) now publishes two ordered snapshots and reads back only
1=b, including after a crash between the two.
1. Code Change Review
1.1 Core Logic Analysis
Delta since my last review (029ad393 -> e9a942f8), Iceberg paths only:
// DeletionVectorTaskWriter: comments on createdFiles, dvOutput, failure, rows,
// insert() (position captured before the rolling write) and delete() (DV vs equality delete)
// IcebergFilesCommitter.pendingResults: Javadoc on retained-history scope, legacy
// partial-match failure and the external-fencing requirementThe rest of the changed files in the merge are dev content that is not part of this PR's diff: git diff apache/dev against this head lists only the 24 PR files, and the head is 0 commits behind dev.
Runtime path (re-checked against the code on this head):
RowKind -> IcebergSinkWriter -> IcebergRecordWriter -> IcebergWriterFactory
v3 + delta -> DeletionVectorTaskWriter (formatVersion > 3 delta -> rejected)
v2 / append -> existing writers
schema change: flush old writer, reset -> chronological List<WriteResult> per writer
prepareCommit(checkpointId) -> combine -> IcebergAggregatedCommitter
group by checkpoint (insertion order) -> orderedGroups(): result-list index = generation
new group when a generation has a nonempty equality delete and earlier results are pending
-> IcebergFilesCommitter.doCommit(group) -> pendingResults(): walk branch ancestry,
match "v1:" fingerprints, legacy candidates only by file identity -> AppendFiles / RowDelta
with seatunnel.commit-ids (+ diagnostic group ordinal/count)
restoreCommit == commit (Flink replays recovered committables through ordinary commit)
source: GenericDeleteFilter + IcebergInternalRecordWrapper keys (timestamp_ns normalized)
Key findings:
- The default v2 path keeps its writer selection; v3 CDC/upsert now reaches the new writer. For the stale-row case the group split gives the equality delete a newer sequence number than the earlier data file, so
1=ais removed. The v3 delta writer behaves like Iceberg's own equality-delta writer: a key inserted twice keeps the later row, a delete of an unknown key becomes an equality delete, andUPDATE_BEFOREis ignored in upsert mode. - The removed v3 guards (
checkDeltaWriteSupportedinIcebergWriterFactoryandcheckIdentifierFieldsSupportedinSchemaUtils) are exactly the ones this feature intends to replace; nothing unrelated is deleted. Formats above v3 are still rejected for delta writes. - Replay safety: fingerprints include content type, spec ID, path, format, record count, size, and for deletion vectors the referenced data file and Puffin blob offset/length. A checkpoint number alone only selects legacy candidates; a partially matched legacy payload fails rather than replaying a subset.
- The reader change (
IcebergFileScanTaskReader) only replaces Iceberg'sInternalRecordWrapperwith the connector'sIcebergInternalRecordWrapper(same conversions plustimestamp_ns) and routes equality-delete loading through a copy-retaining loader only when the key schema contains atimestamp_nscolumn; other key schemas use Iceberg's own loader.
1.2 Compatibility Impact
Partially incompatible, documented. v2 stays the default; public options, plugin registration and checkpoint state classes are unchanged. For v2 and v3 CDC/upsert jobs, a checkpoint that spans writer generations can now produce several snapshots, so external readers may observe a committed prefix between groups. Both incompatible-changes.md files (under ## dev) describe this. The new entry is a single paragraph, whereas neighboring entries use "Affected jobs / Behavior change / Migration" bullets (Issue 3).
1.3 Performance and Side Effects
Writer memory: one entry per distinct key per partition per checkpoint, the same shape as Iceberg's own equality-delta writer. Commit: each group walks in-memory snapshot ancestry; manifests are read only for legacy candidates. The new cost I had not sized is the snapshot summary (Issue 1).
1.4 Error Handling, Logging and Issues
Write and close failures are sticky and complete() fails after them; abort deletes only files created by this attempt (including a partially written Puffin stream); manifest-read errors propagate. A crash between groups leaves a visible prefix that replay completes (tested). Fingerprint checking is not a lock against two concurrent committers of the same payload; the docs state that external fencing is required.
Medium
Issue 1 (carryover, mine): seatunnel.commit-ids grows with sink parallelism and with every retained snapshot
- Location:
IcebergFilesCommitter.javacommit(...)(theidsstring:"v1:"plus one 64-hex SHA-256 per pending writer result, comma separated) andfileId/commitId. - Problem: Every writer result that has files adds a 64-character digest to the summary of the snapshot it is published in. Snapshot summaries are stored in the table metadata JSON, which Iceberg rewrites on each commit and every reader loads. Rough size: parallelism 32 gives about 2 KB of fingerprints per snapshot, comparable to or larger than the standard summary entries. At one checkpoint per minute and seven days of retained snapshots (about 10,000) that is on the order of 20 MB of extra metadata. At parallelism 100 and above it is correspondingly larger. This is my estimate from the code, not a measurement.
- Risk: Slower commits, table loads and catalog traffic for high-parallelism, high-checkpoint-frequency CDC jobs that retain many snapshots, which are exactly the jobs this feature targets.
- Best improvement: Keep per-result membership (it is what lets mixed published/new results be separated) but store a shorter digest (for example the first 128 bits, base64 or 32 hex characters), and add a sentence to the docs recommending routine snapshot expiration consistent with the recovery-evidence window. A single digest for the whole group would not work because replay needs per-result matching.
- Severity: Medium, non-blocking.
- Raised by another reviewer: No.
Low
Issue 2: Document the checkpoint-ordering precondition in IcebergAggregatedCommitter.commit (carried forward, unchanged)
- Location:
IcebergAggregatedCommitter.java:58-67. - Problem: Multiple checkpoints in one commit are published in incoming (insertion) order rather than sorted by positive checkpoint ID. I did not find a caller that delivers them reversed.
- Risk: A future adapter that changes ordering could publish a later checkpoint first.
- Best improvement: One sentence stating the precondition, or an explicit sort with a real-table regression if reversed input is meant to be supported.
- Severity: Low, non-blocking.
- Raised by another reviewer: No.
Issue 3: Style and doc-format nits
- Location:
IcebergFilesCommitter.java:356(java.util.stream.Collectors.joiningwritten fully qualified in production code whiletoListis already statically imported); fully qualified Mockito/Iceberg types inIcebergAggregatedCommitterTestandIcebergSinkIT(carried forward); the newincompatible-changes.mdentry (EN and ZH) is one paragraph rather than the Affected jobs / Behavior change / Migration bullets used by neighboring entries. - Risk: Readability only.
- Best improvement: Normal imports and the bullet structure when touching these lines.
- Severity: Low, non-blocking.
- Raised by another reviewer: No.
2. Code Quality
2.1 Coding Standards
The comment request from my last review is now met: the new writer state (createdFiles, dvOutput, failure, rows), insert/delete and pendingResults all explain the invariant and the why. ASF headers and the Javadoc on the grouping contract are present. No new dependency.
2.2 Tests and Stability
IcebergFormatVersionTest (exact final rows for v2/v3, all three data formats, rolling at the file-size boundary, abort and partial-Puffin failure, schema and partition changes, ordinary and restore replay), real-table IcebergFilesCommitterTest and IcebergAggregatedCommitterTest (including publishing only the first generation then replaying), IcebergFlinkRecoveryIT and IcebergZetaRecoveryIT (kill and restore with exact rows and snapshot counts, bounded Awaitility waits). No test was removed or loosened in the delta since my last review; earlier number-only expectations were replaced by file-identity expectations in prior rounds.
Flaky-stability rating: Stable. Waits are condition-based with explicit limits, paths are unique, and the recovery ITs assert state, not timing. Spark is excluded from CDC row-kind testing with a stated reason.
2.3 Documentation
English and Chinese sink guides, changelog and incompatible-changes.md are updated and consistent with the code (v3 writer behavior, partition-column and key-affinity requirements, retained recovery evidence, the visible-prefix behavior, and the manifest-scan cost for legacy payloads without a positive checkpoint ID).
3. Architectural Soundness
3.1 Fix versus Workaround
A root-cause fix: ordering belongs in the aggregated committer (it sees all writers of a checkpoint), payload identity belongs in the file committer, and the reader-side timestamp normalization stays connector-specific. Coordinate the overlapping grouping and summary keys with #12937 when either lands, as the PR description already says.
3.2 Maintainability
Generation ordering, payload identity and cleanup ownership stay separate; the new comments protect those boundaries.
3.3 Extensibility
Fingerprints are versioned (v1:), leaving room for a later format. Existing-DV rewriting, row lineage and other v3 features are explicitly separate slices.
3.4 Historical Compatibility
No checkpoint state class changes. Legacy snapshots without fingerprints are handled conservatively (file-identity match, fail on partial match). Recovery after the relevant snapshot history has expired is documented as outside the guarantee.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Fingerprint list in snapshot summary scales with parallelism x retained snapshots | IcebergFilesCommitter.commit |
Medium, non-blocking |
| 2 | Checkpoint-ordering precondition undocumented | IcebergAggregatedCommitter.java:58-67 |
Low |
| 3 | FQN in production code and tests; changelog entry format | IcebergFilesCommitter.java:356, tests, docs |
Low |
5. Merge Recommendation
Conclusion: Ready to merge
CI: on this exact head, the fork run 41352844311 was still in progress when I checked: license, dead-links, code-style, sanity and Helm checks and updated-modules-integration-test-part-1/2 are green; unit-test (4 lanes) and iceberg-connector-it (JDK 11 and 17) were still running. The previous head 029ad393 had a green Build, including the Iceberg recovery ITs, and the delta since then is comments plus a merge of dev, so I see no source-side reason for it to change, but the result for this head is not yet in. The branch is 0 commits behind dev and mergeable. Required approvals are still outstanding (review decision is REVIEW_REQUIRED).
- Blockers: none from source review. Remaining gates are the exact-head CI result and a write-capable maintainer approval.
- Recommended non-blocking: Issue 1 (shorter digest plus a docs note on snapshot expiration), then Issues 2 and 3.
Thank you for the thorough work and for coordinating with #12937.
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
CI status update for head e9a942f8f3: my previous review on this head left the current-head Build result as the only open gate. The required Build check has since completed successfully on this exact head (finished 2026-11-02T03:48:28Z), so the CI fact cited in that review has changed.
CI was previously the only remaining blocker, and I found no source, test, compatibility or documentation blocker in my last full review; CI is now passing, so I am changing my conclusion to approve. Issues 1 to 3 from that review remain optional, non-blocking follow-ups.
Note that I have read-only permission on this repository, so this approval records my technical conclusion; a maintainer with write access still owns the effective approval and the merge. Thanks again for the careful work!
Purpose of this pull request
Related to #12763. This is the second Iceberg v3 implementation slice after #12777; it does not close the remaining v3 roadmap.
Does this PR introduce any user-facing change?
Yes. Format-version 3 CDC/upsert configurations become supported. Format version 2 remains the default and keeps its existing writer selection.
For both v2 and v3 CDC/upsert jobs, a checkpoint spanning writer generations can publish several ordered snapshots when needed for equality-delete ordering. External readers may observe a committed prefix between groups. Single-generation and append-only checkpoints retain their single-snapshot behavior. Both incompatible-changes documents explain this compatibility impact.
The result-list index is each writer's chronological generation, not a global order between independent writers. Existing key-affinity, ordering and partition-key requirements remain necessary. Group ordinals/counts are diagnostic metadata, not evidence that a payload was published.
No new dependency, configuration option, public API or checkpoint serialization format is introduced. Recovery is scoped to the configured table branch and requires retained snapshot history, manifests and pending files, plus engine fencing against overlapping attempts publishing the same payload. Partial legacy matches and manifest-read failures fail closed. Recovery after the necessary evidence expires is not covered.
Overlap with #12937: both changes touch generation grouping and the same commit-group summary keys. This PR is self-contained; whichever lands second must reconcile with the landed implementation rather than introduce a second grouping rule. No unmerged #12937 code is included here.
Variant/default values, row-lineage exposure, incremental v3 reads and multi-argument transforms remain separate roadmap slices.
How was this patch tested?
The current connector suite passed on JDK 11 and JDK 17: 187 reported tests per runtime, 186 executed and passed, and one existing skipped metastore test. Scoped Spotless, verification and packaging passed on JDK 11. The suite covers ordered publication and replay for v2/v3, repeated updates/deletes, schema and partition/key changes, file rolling, checkpoint-ID reuse, regrouped committables, branch isolation, legacy publication checks and deletion-vector identity.
Earlier recovery evidence, not a claim that the entire matrix was rerun on the current head:
Full-reactor execution, Spark recovery, native Windows, cloud catalogs and a fresh complete Docker recovery matrix are not claimed. Hosted CI and maintainer approval remain separate gates.
Check list
incompatible-changes.mdupdated in both languages for ordered multi-snapshot publication and its visible-prefix boundary.