[Asthenia0412]

Fixes #18262. When schema evolution creates multiple task writers in one checkpoint, an UPDATE_AFTER in a later writer can emit an equality delete that does not remove the earlier same-commit row. Share the inserted-row position tracker only when table, branch, spec, and equality-key type match; clear it after the checkpoint. The regression test disables checkpointing, writes an INSERT then UPDATE_AFTER across nullable-column addition, and asserts only the updated row remains. Verified with :iceberg-core:test, :iceberg-flink:iceberg-flink-2.3:test, spotlessCheck, and :iceberg-core:revapi.

---

AI Disclosure

• Model: GPT-6
• Platform/Tool: Codex
• Human Oversight: unreviewed
• Prompt Summary: Fix Apache Iceberg issue Flink: DynamicIcebergSink writes an unbindable equality delete when a schema evolution splits one table across two writers inside one checkpoint #18262 by sharing equality-key position tracking across schema-specific Flink writers and adding regression coverage.

[abrarsher23]

Thanks for picking this up. I opened #18262 and have been testing a similar change, so here are two cases worth covering:

1. Rename or doc change on a key column in the same checkpoint. PositionDeleteTrackerKey compares Types.StructType, and NestedField.equals also compares names, docs and defaults. The dynamic sink can update a column's doc on its own, so after such a change the second writer gets its own tracker and the duplicate comes back. Comparing field ID, type and optionality by position, which is what StructLikeMap compares, keeps the writers sharing.
2. Non-upsert mode. The tracker is only created when upsertMode() is true. With equality fields and upsert off, UPDATE_BEFORE / DELETE still go through delete(row) and the same inserted-row lookup, so a CDC changelog across a schema change hits the same problem.

[abrarsher23]

Following up on my earlier comment. We needed this fix before an upstream release, so we implemented it in our fork: Affirm#17 (on 1.11.0, flink/v2.0). It's the same basic idea as this PR, so here's what we learned along the way in case it helps. Happy to help port any of it.

What we verified. Beyond unit tests, we ran the real sink end to end against a live CDC stream. In one transaction a row is inserted, a column is added, and the same row is updated, all inside one checkpoint. Without the change the key was visible twice (two equality deletes, no position delete). With it, the second writer retired the first writer's row with a position delete and the key was visible once.

Suggestions:

1. Key-type comparison (from my earlier comment, still applies). PositionDeleteTrackerKey uses Types.StructType equality, which includes field names, docs and defaults. A doc-only change on a key column, which the dynamic sink applies on its own, gives the second writer a new tracker and the duplicate comes back. Comparing field ID, type and optionality by position, which is what StructLikeMap compares, avoids that. In our PR that's InsertedRowTracker.acceptsKeyType.
2. CDC without upsert (also from before). Trackers are only created when upsertMode() is true, but UPDATE_BEFORE / DELETE go through delete(row) and the same inserted-row lookup, so a changelog written without upsert hits the same bug. Passing the tracker regardless of upsert mode fixes it, and single-writer behaviour doesn't change.
3. One tracker per partition. BaseDeltaTaskWriter hands the same tracker to every RowDataDeltaWriter, so in PartitionedDeltaWriter all partitions of a dynamic-sink table now share one map, even without a schema change. writePosDelete writes the delete under the current writer's partitionKey, so if a key were ever found through another partition's entry, the position delete would land in a different partition from the data file it names. With the partition fields in the upsert key that shouldn't happen today, but keeping one tracker per partition (the existing per-partition map, shared across writers) removes the dependency on it.
4. Extra loadTable. When a record carries no equality fields, the tracker path calls catalog.loadTable again for each new writer, although the factory was just built from the same table. Reusing the equality fields the factory already resolved saves a catalog round trip per writer.
5. Checkstyle. build-checks fails on PositionDeleteTracker.PathOffset: path and rowOffset must be private with accessors (VisibilityModifier).
6. API shape. +1 to @pvary's create / track / untrack suggestion. We ended up with something close: InsertedRowTracker.create(keyType), a key-type check, and put/remove kept package-private. A writer only clears a tracker it created itself; the sink owns the shared ones and drops them at prepareCommit and close.
7. Tests. The current test covers the unpartitioned upsert + add-column case. Other cases we found worth pinning: a partitioned table; a DELETE through the second writer; alternating old/new schema versions within one checkpoint (newest value wins); a key-column doc change; a CDC update without upsert; trackers dropped at the checkpoint, so the next checkpoint falls back to an equality delete; and trackers scoped per table. A quick mutation check was useful too: giving each writer its own tracker again makes 7 of the 19 TestDynamicWriter tests in our PR fail.

[huaxingao]

@Asthenia0412 Thanks for taking this on. The approach looks correct to me.

I've also reviewed @abrarsher23's implementation at Affirm#17, which takes the same approach with more cases covered, and I agree with his points above. Of the seven, I'd treat (1) key-type comparison, (2) CDC without upsert and (3) one tracker per partition as required before merge, since each changes behaviour; (5) is already failing CI. And +1 to @pvary's create/track/untrack shape on (6): Affirm#17 ended up close to it, with put/remove kept package-private and the sink owning the shared trackers. I'd like to see the two converge rather than drift apart.

[zhengxiang378928908-code]

Thanks @Asthenia0412. I've been working on the same problem in another sink, so I read this with the cases from @abrarsher23 and @huaxingao in mind. I agree with the direction and with the four items marked as required. One more case, which tracker sharing cannot cover by design.

Type promotion of an equality column still keeps both rows

Trackers are keyed by (table, branch, specId, equalityKeyType). If an equality column is promoted inside a checkpoint, for example int -> long, the dynamic sink evolves the table inline (EvolveSchemaVisitor.updateColumn -> UpdateSchema.updateColumn, which has no restriction on identifier fields). The next writer then has a different equalityKeyType and starts with an empty tracker. Sharing would not help here anyway, since the old keys are Integer and the new ones Long. So the update becomes an equality delete in the same RowDelta as the old row, and both rows stay.

I checked this on this branch (a1e01f37f) with a test next to upsertAcrossSchemaEvolutionInCheckpoint, using the same setup: upsert, equality field id, immediateUpdate = true, parallelism 1, one commit.

• INSERT (id=1, amount=100) with id int
• then UPDATE_AFTER (id=1L, amount=200) with id long

Test

@Test
void upsertAcrossKeyTypePromotionInCheckpoint() throws Exception {
  String tableName = "key_type_promotion_upsert";
  TableIdentifier tableId = TableIdentifier.of(DATABASE, tableName);
  Schema schema =
      new Schema(
          Types.NestedField.required(1, "id", Types.IntegerType.get()),
          Types.NestedField.required(2, "amount", Types.IntegerType.get()));
  Schema promotedSchema =
      new Schema(
          Types.NestedField.required(1, "id", Types.LongType.get()),
          Types.NestedField.required(2, "amount", Types.IntegerType.get()));
  PartitionSpec spec = PartitionSpec.unpartitioned();
  CATALOG_EXTENSION
      .catalog()
      .createTable(tableId, schema, spec, ImmutableMap.of(TableProperties.FORMAT_VERSION, "2"));

  DynamicIcebergDataImpl firstWrite =
      new DynamicIcebergDataImpl(
          schema, tableName, SnapshotRef.MAIN_BRANCH, spec, true, Sets.newHashSet("id"), false);
  firstWrite.rowProvided = Row.of(1, 100);
  DynamicIcebergDataImpl update =
      new DynamicIcebergDataImpl(
          promotedSchema, tableName, SnapshotRef.MAIN_BRANCH, spec, true, Sets.newHashSet("id"), false);
  GenericRowData updateRow = new GenericRowData(2);
  updateRow.setField(0, 1L);
  updateRow.setField(1, 200);
  updateRow.setRowKind(RowKind.UPDATE_AFTER);
  update.rowDataProvided = updateRow;

  env =
      StreamExecutionEnvironment.getExecutionEnvironment(
          MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG);
  executeDynamicSink(Lists.newArrayList(firstWrite, update), env, true, 1, null);

  Table table = CATALOG_EXTENSION.catalog().loadTable(tableId);
  assertThat(table.schema().findType("id")).isEqualTo(Types.LongType.get());
  List<Record> actualRows;
  try (CloseableIterable<Record> rows = IcebergGenerics.read(table).build()) {
    actualRows = Lists.newArrayList(rows);
  }
  assertThat(actualRows).hasSize(1);
}

Result: upsertAcrossSchemaEvolutionInCheckpoint passes, and this one fails with [Record(1, 100), Record(1, 200)]. The snapshot summary shows added-data-files=2, added-equality-deletes=2, total-position-deletes=0, the same signature as #18262.

I don't think this should block the PR. It would help to make the gap explicit, though: for example a WARN when a writer for a table/branch that already has a tracker in this checkpoint gets a new one, plus this test pinning the current behaviour.

A way to close it later

Keep the order: commit what was written before the key change as an earlier snapshot, so the later equality deletes get a higher sequence number and apply. The regular sink already relies on this across checkpoints: commitDeltaTxn commits each pending checkpoint as its own RowDelta. Doing it inside a checkpoint would need two changes:

• the writer reports one result per tracker scope, in order;
• DynamicWriteResultAggregator, which today merges everything per table and spec, keeps those results apart.

I did this for the Iceberg sink in Apache SeaTunnel (apache/seatunnel#12937, I'm the author). It costs extra snapshots, so here it would only be worth doing when a scope change actually happens. That belongs in a follow-up issue rather than this PR; I'm happy to open one.

(I also looked at a partition spec change mid-checkpoint. That is not specific to this PR: partitioned equality deletes only match data files of the same spec and partition in DeleteFileIndex, so they never applied to rows in the old spec, whatever the sequence number.)