[I] Flink: DynamicIcebergSink still duplicates a row when a key column's optionality, type or order, or the equality-field set, changes within one checkpoint [iceberg]

["abrarsher23 (via GitHub)" <gi...@apache.org>]

abrarsher23 opened a new issue, #18350:
URL: https://github.com/apache/iceberg/issues/18350

   **Apache Iceberg version:** `main` (`flink/v2.3`) with the fix proposed for #18262 in #18288. On released versions, including 1.11.0, these cases already duplicate as part of #18262.
   
   **Query engine:** Flink (`DynamicIcebergSink`, upsert and CDC modes)
   
   **Please describe the bug 🐞**
   
   ### Summary
   
   #18262 describes the base bug. A schema change in the middle of a checkpoint splits a table across two writers. When the second writer updates a key that the first writer already wrote in that checkpoint, it produces an equality delete with the same sequence number as the first writer's row, so the delete never applies and both versions stay visible.
   
   The fix proposed in #18288 lets all writers of a table share one inserted-row tracker for the length of a checkpoint. But writers can share only when their key types are compatible. In each case below the second writer still gets its own tracker, so the duplicate remains. Every case needs the same key to be written both before and after the change within one checkpoint.
   
   ### Cases
   
   1. **An equality field becomes optional.**
      - **When it happens:** when equality fields are set explicitly and aren't identifier fields (identifier fields must be required). `EvolveSchemaVisitor` makes a column optional when the input field is optional.
      - **Why the obvious fix is unsafe:** you can't simply ignore optionality when comparing key types. For a required `int` or `long` field, the comparator throws on null, and `StructLikeWrapper.equals` treats the exception as "not equal". So a tracker built for a required key never matches null keys, and null-key upserts through the new writer would duplicate.
   
   2. **An equality field's type is promoted** (`int` → `long`, `float` → `double`, or decimal precision).
      - **Why writers split:** `TableMetadataCache` resolves records still on the old input schema to the old table schema. So a narrow writer and a wide writer can both be open, and records can alternate between them.
      - **Why the tracker can't be shared as is:** `Integer` and `Long` keys never match in a `StructLikeMap`. Negative values hash differently, and otherwise the comparison throws a `ClassCastException` that is swallowed as "not equal".
   
   3. **The equality-field set changes**, through per-record `DynamicRecord.setEqualityFields`, or identifier fields changed outside the sink.
      - **Why the tracker can't follow it:** the tracker stores only key → position, so after any change one lookup direction is ambiguous. For example, after `(id)` → `(id, tenant)`, an update for `(7, B)` projected back to `id = 7` could retire tenant A's row.
      - **Routing makes it worse:** with key-hashed distribution, the old and new key of a row usually land on different subtasks anyway.
   
   4. **With a multi-column key, the key columns change order.** `EvolveSchemaVisitor` can move columns to follow the input order, and the key type follows table column order. So a position-based key-type comparison doesn't match, and the writers don't share.
   
   ### Expected behaviour
   
   Either the writers share the tracker, so the later write retires the earlier row with a position delete; or, where sharing isn't possible, the sink makes the gap visible (a WARN and a metric) and the limitation is documented.
   
   ### Possible directions (not yet evaluated upstream)
   
   - **Case 1:** build the tracker's map from an all-optional copy of the key type. Non-null keys compare exactly as before, and null matches null, as it does in an equality delete. Small and contained.
   - **Case 2:**
     - Same-scale decimal promotion keeps identical `BigDecimal` values, so it only needs the key-type check relaxed.
     - `int` → `long` and `float` → `double` need key conversion:
       - keep the first writer's key type as canonical
       - give writers of a promoted type a converting view
       - keep wide values that don't fit the narrow type in a separate map
       - normalise identity-partition keys the same way
     - The normal path stays unchanged.
   - **Case 3:** probably a documented limitation, plus a WARN and a metric when a second key set opens for the same table within a checkpoint.
   - **Case 4:** either compare key fields by field ID and reorder stored keys, or build the key projection in field-ID order so a column move doesn't change the key type. Not yet checked: what the second option means for equality-delete file schemas.
   
   Earlier discussion: https://github.com/apache/iceberg/issues/18262#issuecomment-5956648553
   
   ### Willingness to contribute
   
   - [ ] I can contribute a fix for this bug independently
   - [x] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
   - [ ] I cannot contribute a fix for this bug at this time
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [I] Flink: DynamicIcebergSink still duplicates a row when a key column's optionality, type or order, or the equality-field set, changes within one checkpoint [iceberg]

["abrarsher23 (via GitHub)" <gi...@apache.org>]

abrarsher23 commented on issue #18350:
URL: https://github.com/apache/iceberg/issues/18350#issuecomment-5984145273

   Update: our fork now has a fix for Case 2 and for the first finding in my follow-up comment above: https://github.com/Affirm/iceberg/pull/20. It's stacked on https://github.com/Affirm/iceberg/pull/17, the fork's fix for #18262.
   
   - **Case 2 (key type promotion).** A writer whose key type is a promotion of the tracker's key type uses the tracker through a converting view. That covers `int`/`long`, `float`/`double` and same-scale decimal, in either direction. Keys that don't fit the narrower type, such as a `long` above the `int` range, go into a small second map keyed by the wide type. When the key types are equal, the writer gets the tracker itself, so the usual path does no conversion.
   - **Writers after a promotion.** Trackers are grouped by key type, so later writers with the same key type share with each other, not only with the first writer.
   - **Not covered there:** Cases 1, 3 and 4, and the partition case, which is blocked by #18355.
   
   **Testing.**
   - Unit tests in `core`, `data` and both Flink modules, covering every changed line and branch.
   - An end-to-end run of the real sink fed by a live CDC stream, with an `int` key promoted to `long` mid-checkpoint. Every key ended up visible once.
   
   Reviews are very welcome, especially on whether this direction would work upstream. If it would, I'm happy to port it as a PR here on top of whichever #18262 fix lands.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [I] Flink: DynamicIcebergSink still duplicates a row when a key column's optionality, type or order, or the equality-field set, changes within one checkpoint [iceberg]

["abrarsher23 (via GitHub)" <gi...@apache.org>]

abrarsher23 commented on issue #18350:
URL: https://github.com/apache/iceberg/issues/18350#issuecomment-5964508104

   Three more findings after implementing the Case 2 direction above and testing it. All three concern key and partition promotion.
   
   **1. Writers after a promotion don't share with each other.** If the first writer's key type stays the reference for the whole checkpoint, this sequence still duplicates: insert `id=9` (`int`), promote `id` to `long`, update `id=9`, add a column, update `id=9`. The last two writers both have `long` keys, but each is compared with the original `int` key, so they get separate trackers. Fix approach: group trackers by key type. A writer joins the first group whose key type it can share, otherwise it opens a group, which later writers of the same type join.
   
   **2. A promoted key's old and new records can go to different subtasks.** With equality-field distribution (and unpartitioned hash distribution), the key hash comes from `Objects::hashCode` per field. `Integer.hashCode` and `Long.hashCode` differ for negative values, and `Float.hashCode` and `Double.hashCode` differ for nearly every value. So the records of one key written before and after the promotion can reach different writers, and no tracker can be shared. In a probe against `HashKeyGenerator` on `flink/v2.0` and `v2.1` (one `int` equality field, write parallelism 4), all 50 keys in -50..-1 split, and none of 0..50 did. I have not checked `main`. Fix approach: hash promoted primitive keys in their widest type (`int` as `long`, `float` as `double`). Non-negative `int` keys keep their hash, so their routing doesn't change.
   
   **3. The non-key partition column case is blocked by a separate bug.** If an identity or truncate partition column is promoted, a checkpoint with `Integer` and `Long` partition values of one spec fails to commit, even with no re-write, so a shared tracker can't help. That is a commit failure in core, filed as #18355. This also means the "normalise identity-partition keys the same way" part of the Case 2 direction doesn't help until that is fixed.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org