[abrarsher23]

Change Management

This PR is the change-management record for two changes: this PR (released as 1.11.0-affirm-patch.4), and the downstream PR that bumps our CDC sink's Iceberg version to that release. Approvals for both are collected here.

• Jira: LAKE-7102 (follow-up to LAKE-7032, which shipped Core, Flink: Share the inserted-row tracker across DynamicIcebergSink writers of one table #17)
• Upstream: apache/iceberg#18262 (the bug), #18350 (known gaps), #18355 (partition promotion)

What this changes

#17 lets the writers of one table share an inserted-row tracker within a checkpoint. When a schema change opens a new writer mid-checkpoint, the new writer can then retire the old writer's row with a position delete, so the row isn't left visible twice. But #17 only shares when the key types match exactly. If a key column is widened (int → long) in the same checkpoint, the row is still left visible twice.

This PR closes that gap:

• Writers share a tracker across a key promotion. These promotions are covered: int↔long, float↔double and same-scale decimals. A promoted writer reads the first writer's tracker through a view that converts each key exactly (7L → 7). A key that has no exact equivalent, such as a long above the int range, goes into a small side map that only other promoted writers check. No int writer can have written that key, so nothing is missed.
• Same key type, same path as before. A writer whose key type matches gets the tracker itself.
• Logs say what happened. An INFO is logged when a writer shares another writer's tracker. A WARN is logged when it can't share, and another WARN for the partition case below.

The changes are in core/.../io/BaseTaskWriter.java (canShareWith, the converting view) and flink/v2.0 and v2.1 (DynamicWriter, plus small changes in BaseDeltaTaskWriter and RowDataTaskWriterFactory). The two Flink versions get identical changes, and flink/v1.20 is untouched.

Not fixed here. Each of these happens within one checkpoint:

• A promoted identity/truncate partition column. The commit fails because of a separate upstream bug (Core: Committing data files fails after promoting the source column of an identity partition field apache/iceberg#18355), so no duplicate is left. The behaviour is pinned by tests.
• A key becoming optional.
• The equality fields or the key column order changing.
• A promotion in a key that also has a nested field.
• A key's rows reaching different writer subtasks.

The writers log a WARN for these. Any duplicate left behind is removed by the key's next upsert or delete.

Risk

Low. Checkout impact: none. This is a library change.

• Wrong delete after a promotion. The risk is that the view matches the wrong row. It's covered by 12 new tests × 9 format combinations: NaN and -0.0, keys outside the int range, null keys, two promoted fields, and a reused mutable row. canShareWith is checked against TypeUtil.isPromotionAllowed for all 289 type pairs.
• No change without a promotion. If no key is promoted mid-checkpoint, the write path is the same as in .3.
• No format changes. No data, metadata or Flink state formats change, so jobs restore from savepoint across the bump.
• API. canShareWith replaces Core, Flink: Share the inserted-row tracker across DynamicIcebergSink writers of one table #17's acceptsKeyType, which had no callers. revapi reports 0 differences.

Release and rollback

1. Merge (squash).
2. Release PR (version.txt, AFFIRM_VERSIONS.md), tag 1.11.0-affirm-patch.4 and publish.
3. Downstream version-bump PR: stage deploy, then prod.

Rollback: pin back to 1.11.0-affirm-patch.3 and redeploy. The deletes this version writes are standard Iceberg files that .3 reads, so nothing needs migrating.

Observability

• Jobs return to RUNNING after the deploy, and checkpoints keep completing.
• New snapshots report iceberg-version 1.11.0-affirm-patch.4.
• Logs:
  • INFO … share an inserted-row tracker …: sharing worked.
  • WARN Not sharing inserted-row tracker …: a case listed above. Investigate if it appears after a promotion.
  • WARN Partition … in other types …: the commit will fail (Core: Committing data files fails after promoting the source column of an identity partition field apache/iceberg#18355).

Testing

Unit tests (JDK 21). The full core, data, flink-2.0 and flink-2.1 suites pass, with 0 failures. Those runs were at 5b4df4ffa. The head, 0f0e6b9c0, adds one test and no production code. At the head, TestTaskEqualityDeltaWriter passes 216 of 216.

Test class	New
TestTaskEqualityDeltaWriter	12 × 9 formats: promotions both ways, overflow keys, decimals, nulls, deletes, reused rows
TestDeltaTaskWriter (each Flink module)	2 × 9: sharing across a promotion; a resolver that returns null
TestDynamicWriter (each Flink module)	12: the reviewer's sequence, alternating schemas, keys above the int range, a key becoming optional, partition promotion failing to commit, a later upsert removing a leftover duplicate

Coverage. JaCoCo shows 166 of 166 changed lines and 104 of 105 branches exercised. The one uncovered branch only skips repeating a WARN.

Mutation checks. Breaking the change makes tests fail in each case:

Mutation	Failing tests
No sharing	15 of 31 TestDynamicWriter
Promotions refused	108 of 207 TestTaskEqualityDeltaWriter
View never used	72 of 207 TestTaskEqualityDeltaWriter
View stores its reused lookup key	18 of 216 TestTaskEqualityDeltaWriter

Stage testing checklist. A dev build of 5b4df4ffa ran in our staging environment against a live CDC stream, with an INT-keyed test table. Each session landed in one checkpoint:

• The writer ran the dev build, verified from the jar's commit ID.
• Insert, add a column, update: one position delete, and the key is visible once.
• Add a column, then update existing keys: equality deletes only, which is correct.
• Insert int keys (one negative), widen id to BIGINT, update, delete, insert keys above the int range, add a column, update again. All in one commit: every writer shared, and every key's count matched the source.
• Updates across another added column after the widening: shared, and the counts were correct.
• The final table was identical to the source, row for row.

Only int → long on format v2 was exercised in stage. The other promotions are covered by unit tests only. No benchmark was run.

CI. core-tests and kafka-connect-tests fail the same way on the base branch: the MinIO and Docker setups can't start. Neither failure is in code this PR touches. The Flink, Spark and Hive checks pass.

🤖 Generated with Claude Code

[rayhondo]

Reviewed at 5b4df4f. The shared-tracker design looks coherent, and the 207 new TestTaskEqualityDeltaWriter cases pass in the core-tests (17) CI log. Two items below, one question and one follow-up request. Separately: core-tests and kafka-connect-tests are red in CI (S3/MinIO tests and testcontainers launch errors, none in the files this PR touches). The description says the core suite passed locally, so it would help to confirm whether the same failures occur on the base branch.

[abrarsher23]

On CI: the same failures are on the base branch. On the 1.11.0-affirm-patch push at 2767711, core-tests (17) has the same S3/MinIO failures (TestS3FileIO 36, TestS3FileIOWithLegacyMinIO 13, TestMinioUtil 1, TestFlakyS3InputStream 1, against 36/12/1/1 here) and is cancelled at the 6-hour job limit, as here. kafka-connect-tests (17) fails with the same 3 initializationErrors (TestIntegration, TestIntegrationDynamicTable, TestIntegrationMultiTable). None of these are in modules this PR touches.

[rayhondo]

Both earlier comments are addressed: the key-copy concern does not apply (StructLikeUtil.copy is shallow on both paths), and the partition-promotion case is pinned by the FailsToCommit tests. Approving at 0f0e6b9.