[abrarsher23]

Summary

Problem. In the Flink DynamicIcebergSink, a table's schema can change in the middle of a checkpoint, for example when a column is added. If a row written before the change is updated or deleted after it, the table can end up with two visible copies of that row. This affects tables written with equality fields, in both upsert and CDC mode. Upstream issue: apache/iceberg#18262.

Fix. All writers of a table now share one in-memory record of the rows written in the current checkpoint, the inserted-row tracker. The writer opened after the schema change finds the earlier row and removes it with a position delete, the same way a single writer already does.

Scope. The change is in core and in the two Flink modules this fork publishes, flink/v2.0 and flink/v2.1, which get the same change. flink/v1.20 is unchanged. A follow-up stacked on this PR, #20 (open), handles key type changes.

Why the duplicate happens

1. Writer A writes id=7 as row 0 of file A and records it in its tracker.
2. A column is added, and the sink opens writer B for the new schema.
3. An update of id=7 goes to B. B's tracker is empty, so B writes an equality delete for id=7.
4. Files A and B are committed together and share one sequence number. An equality delete only hides rows from earlier commits, so A's row stays visible, and the table now has two rows for id=7.

With a shared tracker, B finds id=7 in file A and writes a position delete for row 0 of file A. A position delete does apply within the same commit.

A later update or delete of the key, in a later checkpoint, removes both copies. Snapshots and tags taken in between still show both.

What changes

File	Change
core/.../io/BaseTaskWriter.java	Adds InsertedRowTracker, and a BaseEqualityDeltaWriter constructor that takes a shared tracker. A writer only clears a tracker it created. The API change is additive, and revapi is clean.
flink/v2.{0,1}/.../sink/RowDataTaskWriterFactory.java, BaseDeltaTaskWriter.java, PartitionedDeltaWriter.java, UnpartitionedDeltaWriter.java	Pass an optional tracker resolver down to the delta writers. The resolver isn't serialized, so other users of the factory behave as before. BaseDeltaTaskWriter's old constructor is removed because it had no callers left.
flink/v2.{0,1}/.../sink/dynamic/DynamicWriter.java	Owns one tracker per table and partition. Clears them at every checkpoint and on close. Logs a WARN when a schema change opens a second writer for a table within a checkpoint.
flink/v2.{0,1}/.../sink/dynamic/WriteTarget.java	withoutSchemaId() defines the sharing scope: table, branch, partition spec, upsert mode and equality fields. sameScopeAs() compares two scopes without copying.
Tests	TestTaskEqualityDeltaWriter, TestDeltaTaskWriter, TestDynamicWriter, TestWriteTarget

Writers share a tracker only when their key types match: the same field IDs, types and optionality, compared position by position. Renaming a key column or changing its doc still shares.

Not fixed in this PR

All of these happen within one checkpoint.

Case	What happens
A key column is promoted (int → long, float → double)	The writers keep separate trackers and log Not sharing inserted-row tracker …. Addressed in #20 (open).
Writers after a key promotion don't share with each other	Addressed in #20 (open).
Promoted negative int keys or float keys go to different writer subtasks	Predates this PR. Not addressed here.
A key column becomes optional, or the columns of a multi-column key change order	Not fixed. The writers log Not sharing inserted-row tracker ….
A non-key identity or truncate partition column is promoted	The checkpoint fails to commit because of a separate, pre-existing bug (apache/iceberg#18355), so sharing can't help. #20 adds tests that pin this behaviour.
The equality fields, partition spec or upsert mode change	The writers get independent trackers, because "the same row" isn't well defined across a change of key.

Risk

• Low to medium. In the normal case, one writer per table per checkpoint, nothing changes: the tracker is the same map as before, only owned by DynamicWriter instead of the writer.
• Behaviour changes only when a table gets a second writer within a checkpoint. Its re-writes then become position deletes instead of equality deletes.
• Other users of RowDataTaskWriterFactory and BaseTaskWriter are unaffected: the non-dynamic Flink sink, Spark and Kafka Connect.
• No data, metadata or Flink state format changes.

Rollout and rollback

• Rollout:
  1. Review and merge this PR. Squash on merge, because a few intermediate commit messages describe states the PR has since moved past.
  2. Release PR (version.txt, AFFIRM_VERSIONS.md).
  3. Tag 1.11.0-affirm-patch.3 and publish.
  4. Downstream version-bump PR. The change-management approvals happen there.
  5. Deploy.
• Rollback: go back to 1.11.0-affirm-patch.2. Position deletes are standard Iceberg files that every reader applies, so there is nothing to migrate.

How to tell it's working

• When a schema change opens a second writer within a checkpoint, the sink logs WARN Opening another writer for table … Re-writes of a key across the schema change are retired with position deletes while the writers share an inserted-row tracker. Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20 replaces this with an INFO that is logged only when sharing actually happens.
• When a second writer can't share a tracker (the cases above), the sink logs WARN Not sharing inserted-row tracker ….
• In the table, the commit has a position delete in the same snapshot as both data files, and the key is visible once.

Testing

• Unit tests (JDK 21), all passing:

  Module	Tests	Failures	Skipped
  iceberg-core	7,580	0	474
  iceberg-data	861	0	87
  iceberg-flink-2.0	4,466	0	371
  iceberg-flink-2.1	4,511	0	370

  The skipped tests were skipped before this change too. core and data ran at 1921606d2, and the Flink suites at 62edc85be. The later commits touch only the Flink modules, and the last one, 102990cf0, only changes a log message. revapi and spotlessCheck are clean when run locally; this PR's CI doesn't run revapi.

• Coverage: JaCoCo shows every changed executable line (141 of 141) and every changed branch (62 of 62) exercised, for both Flink modules.

• The tests catch the bug: giving each writer its own tracker again, as before this PR, makes 7 of the 19 TestDynamicWriter tests fail, in both Flink modules.

• End to end (Flink 2.0): the real writer fed by a live database change stream. In one checkpoint, a row was inserted, a column added, and the same row updated.

  • On the unpatched build, the key was visible twice: two equality deletes and no position delete.
  • With this PR, it was visible once: the second writer retired the first writer's row with a position delete, and the WARN was logged.
  • The same sequence passed again as part of Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20's end-to-end test.

New tests by area

Area	Tests
Shared tracker in core	Writers retire each other's rows with position deletes. Closing a writer, once or twice, keeps the shared tracker. A row delete (CDC UPDATE_BEFORE / DELETE) works through another writer. A tracker of a different key type is rejected. Writers without a shared tracker use the existing constructor unchanged.
Key-type rules	Names, docs and defaults are ignored. Field ID, optionality, type and field count must match.
RowDataTaskWriterFactory	Two writers share the resolver's tracker. A deserialized factory has no resolver and keeps one tracker per writer. A tracker of another key type falls back to a private one.
DynamicWriter, one checkpoint with a schema change	Upsert of the same key. Alternating old and new schema, where the newest value wins. Distinct keys. Partitioned table. Key-column doc change. DELETE through the second writer. CDC update without upsert mode. Equality-field type promotion, which falls back without failing.
DynamicWriter, tracker lifetime	Trackers are dropped at prepareCommit and close, so the next checkpoint uses an equality delete as before. Trackers are per table.
WriteTarget	withoutSchemaId() equality. sameScopeAs() agrees with it for every pair of targets.

Worth knowing

• The failing CI checks are unrelated to this change. The base branch's own CI has failed the same way since 09-25, and so did Release 1.11.0-affirm-patch.2 #15, the previous PR into this branch:

  • core-tests never finishes. The S3 tests in :iceberg-aws:test can't start their MinIO container, so the job runs until GitHub stops it at about 6 hours, which shows as cancelled or failed.
  • kafka-connect-tests fails or is cancelled. Its integration tests can't start their Docker Compose setup.

  This PR doesn't touch any of that code. TestTaskEqualityDeltaWriter passes inside core-tests before the job is cancelled, and the Flink, Spark and Hive checks pass.

• apache/iceberg#18288 is an open upstream PR for the same issue with the same approach. I've commented there on two differences: it compares key names and docs, and it shares only in upsert mode.

[rayhondo]

I'm not approving yet. Mostly I want to confirm the Flink 2.1 point below. The fix itself looks correct to me.

To confirm before approving

The fix only lands in flink/v2.0. The flink/v1.20 and flink/v2.1 copies keep the old behaviour. all-the-things master is on Flink 2.0.1 everywhere today, and mysql-replication-core resolves iceberg-flink-2.0. I vaguely remember earlier talk about moving streaming to Flink 2.1, though. @abrarsher23, did you check with the streaming folks whether we should patch flink/v2.1 as well? Can you confirm we don't need 2.1 any time soon? If we do, this should be ported before the upgrade, or the duplicate-row bug comes back.

Non-blocking

1. warnIfSplitsTable builds a withoutSchemaId() copy of every open writer's target each time a new writer is created, so the cost per checkpoint grows quadratically with the number of open writers. It's cheap at our scale, but caching the scope key per writer would avoid the extra allocations.
   

   iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java

   Lines 189 to 196 in 1921606

   	private void warnIfSplitsTable(WriteTarget newTarget) {
   	WriteTarget scope = newTarget.withoutSchemaId();
   	for (WriteTarget existing : writers.keySet()) {
   	if (existing.withoutSchemaId().equals(scope)) {
   	LOG.warn(
   	"Opening another writer for table {} branch {} within one checkpoint: schema ID {} -> {}, "
   	+ "spec ID {}, subtask {}, attempt {}. Re-writes of a key across the schema change "

2. TableTrackers.acceptsKeyType returns true while byPartition is empty, so the first partition's tracker decides the key type for the table. The per-tracker check in BaseDeltaTaskWriter falls back to a private tracker, so the result stays correct. It does mean sharing can be lost without anything in the log beyond the fallback WARN.
   

   iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java

   Lines 221 to 225 in 1921606

   	private boolean acceptsKeyType(Types.StructType otherKeyType) {
   	return byPartition.isEmpty()
   	|| byPartition.values().iterator().next().acceptsKeyType(otherKeyType);
   	}

[abrarsher23]

Thanks @rayhondo.

Flink 2.1: we don't need 2.1 right now, but the release publishes iceberg-flink-2.1 as well, so I've ported the fix to flink/v2.1 in this PR (e9e6b6a). The touched files were identical in the two modules, so it's the same diff. On 2.1 the full iceberg-flink-2.1 suite passes (4,511 tests, 0 failures), JaCoCo covers every changed line and branch, and giving each writer its own tracker again fails the same 7 of the 19 TestDynamicWriter tests as on 2.0. flink/v1.20 is unchanged.

Non-blocking:

1. Done in 62edc85. warnIfSplitsTable now uses a new WriteTarget.sameScopeAs(), which compares the fields directly instead of copying each target. It only ran when a writer was created, not per row, but there's no reason to allocate. TestWriteTarget checks that it agrees with withoutSchemaId().equals() for every pair of targets.

2. This isn't a correctness problem. In each checkpoint, the first writer for a table sets the key type that the table's shared tracker uses. The empty-map case is just that first call, so there's nothing to compare against yet. If a later writer has a different key type (for example, id went from int to long mid-checkpoint), it gets its own tracker instead, and DynamicWriter logs a WARN showing both key types. So it shows up in the logs, just not as a metric. I can add a counter if you think it's worth it.

   In 102990c I also fixed that WARN's text. It blamed "type promotion or rename", but a rename still shares, because names aren't compared. What actually stops sharing is a key column changing type, becoming optional, or (with a multi-column key) the key columns changing order. That last case is now in the PR description under "Not fixed, on purpose".

(The full-suite and coverage numbers above are from 62edc85. 102990c only changes the log text; the targeted tests pass on it in both modules.)

[huaxingao]

@abrarsher23

The approach looks correct to me, and nothing blocking from my side.

I'd still like Peter (@pvary) and Max (@mxm) to look at it because they're the Flink experts, and apache#18288 is the natural place for that review to happen.

Your version is the more complete of the two: it covers CDC as well as upsert, keeps one tracker per partition, and compares key types structurally rather than through StructType.equals, the gaps you'd already raised on apache#18288 yourself. So I'd rather see that PR converge on yours than have the two drift apart. That way your fork ends up dropping a patch instead of carrying it.

[huaxingao]

LGTM. The failing checks are the MinIO image pull failure; porting apache#18247 should fix them. I'd like the upstream fix in apache#18288 to converge on this version so the fork can drop the patch later.