[abrarsher23]

Stacked on #20, which stacks on #17. This branch starts at #20's head (5b4df4ffa), so the diff shown here is only this PR's two commits. The base will be retargeted when #20 and #17 merge.

Summary

Two defensive changes to tracker sharing in the dynamic sink. Neither matters for a pipeline whose keys never change optionality or column order, and whose keys have no negative int or float values.

1. Writers of the same key type share with each other.
   • With Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20, a writer that can't share the table's first writer's trackers keeps a private tracker. That happens, for example, after a key column becomes optional.
   • Writers after that change that have the same new key type then can't retire each other's rows either.
   • Now trackers are grouped by key type, so those writers share with each other.
2. A promoted key's rows go to the same writer subtask.
   • After a key is widened (int to bigint, float to double), negative int keys and float keys hash differently, so the old and new rows of one key can reach different parallel writers.
   • Trackers can't be shared across subtasks.
   • Now the sink hashes a key the same way before and after the widening.

Scope. Both changes are in flink/v2.0 and flink/v2.1, with byte-identical copies. flink/v1.20 is untouched.

The changes

1. Writers of the same key type share with each other

Example, in one checkpoint:

1. Upsert id=1 while id is required. This is writer A.
2. id is made optional, and id=1 is upserted again by writer B.
3. A column is added, and id=1 is upserted again by writer C.

A tracker for a required key can't safely match null keys, so B and C can't share A's trackers. With #20 alone, B and C each keep a private tracker, and three rows for id=1 stay. With this PR, C shares B's tracker and retires B's row, so two rows stay: A's and C's. The A-to-B duplicate itself isn't fixed (see "Not fixed").

How. DynamicWriter keeps a list of tracker groups per table. A writer joins the first group whose first tracker canShareWith its key type. If none fits, it opens a new group, which later writers of the same key type join. The Not sharing … WARN is logged when a writer has to open a new group.

2. A promoted key's rows go to the same writer subtask

Example at write parallelism 4, in distribution mode NONE with equality fields: upsert id=-5 with id int goes to subtask 2. id is widened to bigint. upsert id=-5 then goes to subtask 3.

The key hash used Java's own hash of each value:

• A negative Integer and the same value as a Long hash differently.
• A Float and the same value as a Double hash differently for nearly every value.

A code probe before this change, at parallelism 4, moved all 50 keys from −50 to −1 to another subtask after the promotion. None of the keys from 0 to 50 moved.

How. A new package-private WidenedEqualityFieldKeySelector widens int values to long and float values to double before hashing. HashKeyGenerator.equalityFieldKeySelector(...) uses it for:

• NONE mode with equality fields
• HASH mode on an unpartitioned table
• RANGE mode

Keys with no negative int and no float values hash exactly as before. Widening goes by column type, so date values aren't touched. HASH mode on a partitioned table routes by the partition path, which needs no change. EqualityFieldKeySelector itself is untouched, so FlinkSink and IcebergSink route as before.

What changes

File (flink/v2.0 path; flink/v2.1 is identical)	Change
.../sink/dynamic/DynamicWriter.java	TrackerGroup replaces #20's single TableTrackers per table. trackerGroup(...) joins or opens a group. The Not sharing WARN is logged when a group is opened.
.../sink/dynamic/WidenedEqualityFieldKeySelector.java	New. Widens and hashes the equality fields.
.../sink/dynamic/HashKeyGenerator.java	equalityFieldKeySelector(...) builds a WidenedEqualityFieldKeySelector.
Tests	TestDynamicWriter, TestHashKeyGenerator, TestWidenedEqualityFieldKeySelector (new)

Not fixed in this PR

Case	Why
The duplicate between the writers before and after a key becomes optional	A required-key tracker can't safely match null keys. One possible direction is in apache#18350 (Case 1).
Routing the sink doesn't control	In forward mode the sink doesn't shuffle, so where a key's rows go depends on the upstream job. A change of write parallelism mid-checkpoint can also move a key.

Risk

• Change 1: behaviour changes only for writers that can't share the table's first writer's trackers. Everything else is identical to Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20.
• Change 2:
  • Rows of negative int keys and of float keys move to a different writer subtask, once, at upgrade.
  • It's safe because DynamicWriter keeps no keyed state across checkpoints: trackers are dropped at every checkpoint, and a restart begins with empty trackers.
  • Other keys keep their subtask.
• Per row: WidenedEqualityFieldKeySelector boxes an int key as a Long (or a float as a Double) before hashing. No benchmark was run.
• API: none. Both new classes are package-private.
• State: no data, metadata or Flink state format changes.

Rollout and rollback

• Rollout: ships after Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20, in a later fork release, through the same release PR, tag, publish and downstream version-bump steps.
• Rollback: go back to the Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20 release. Nothing to migrate.

How to tell it's working

• Change 1: Not sharing inserted-row tracker … is logged once per new group, and the INFO … share an inserted-row tracker … appears for later writers of that key type.
• Change 2: no log. Its effect is that a promoted key's rows reach the same subtask.

Testing

• This head is the code that was fully tested before the split. This PR's head has exactly the tree of the earlier 6b9afeca5, which was Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20 with the grouping plus this routing commit (checked with git rev-parse HEAD^{tree}). These full suites ran at that tree:

  Module	Tests	Failures	Errors	Skipped
  iceberg-core	7,580	0	0	474
  iceberg-data	960	0	0	87
  iceberg-flink-2.0	4,501	0	0	371
  iceberg-flink-2.1	4,546	0	0	370

• Coverage (JaCoCo) at that tree: every line and branch changed since Core, Flink: Share the inserted-row tracker across DynamicIcebergSink writers of one table #17's head is exercised, 219 of 219 lines and 128 of 128 branches. That includes all of this PR's lines.

• Tests, the same in both Flink modules:

  • testUpsertAcrossKeyOptionalityChangeSharesOnlyBetweenWritersOfTheSameKeyType: two groups, and the optional-key writers retire each other's row.
  • TestHashKeyGenerator.testEqualityFieldRoutingIsStableAcrossKeyPromotion: NONE, HASH and RANGE at parallelism 2–4, with keys Integer.MIN_VALUE, −5, −1, 0 and 42.
  • TestWidenedEqualityFieldKeySelector, 5 tests:
    • int, float and multi-field keys hash like their promotions;
    • 1,001 keys without negative int or float values hash as before;
    • the selector is serializable.

• Mutations:

  • Every writer getting its own tracker fails 16 of 31 TestDynamicWriter.
  • Going back to EqualityFieldKeySelector fails 1 of 15 TestHashKeyGenerator.

• Not tested end to end. In the end-to-end test of Core, Flink: Share inserted-row trackers across equality key promotions in the dynamic sink #20 every row went to one writer subtask, and no key changed optionality, so neither change was exercised there.