You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Asthenia0412 (via GitHub)" <gi...@apache.org> on 2026/09/27 20:14:31 UTC
[PR] Flink: Fix same-checkpoint upserts after schema evolution [iceberg]
Asthenia0412 opened a new pull request, #18288:
URL: https://github.com/apache/iceberg/pull/18288
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 #18262 by sharing equality-key position tracking across schema-specific Flink writers and adding regression coverage.
--
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: [PR] Flink: Fix same-checkpoint upserts after schema evolution [iceberg]
Posted by "abrarsher23 (via GitHub)" <gi...@apache.org>.
abrarsher23 commented on PR #18288:
URL: https://github.com/apache/iceberg/pull/18288#issuecomment-5918387868
Following up on my earlier comment. We needed this fix before an upstream release, so we implemented it in our fork: https://github.com/Affirm/iceberg/pull/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.
--
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: [PR] Flink: Fix same-checkpoint upserts after schema evolution [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #18288:
URL: https://github.com/apache/iceberg/pull/18288#discussion_r4120992132
##########
core/src/main/java/org/apache/iceberg/io/PositionDeleteTracker.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.io;
+
+import java.util.Map;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeMap;
+
+/** Tracks data-file positions for equality keys written by related task writers. */
+public final class PositionDeleteTracker {
Review Comment:
Something like this?
```
public class InsertedRowTracker {
private final Map<StructLike, PathOffset> rows;
/** Creates a tracker for row keys of the given type. ... */
public static InsertedRowTracker create(Types.StructType keyType) { ... }
private InsertedRowTracker(Types.StructType keyType) { ... }
/** Records where a row was written. ... */
public void track(StructLike key, CharSequence path, long rowOffset) { ... }
/** Stops tracking a row and reports where it was written. ... */
public boolean untrack(StructLike key, PositionDelete<?> position) { ... }
}```
--
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: [PR] Flink: Fix same-checkpoint upserts after schema evolution [iceberg]
Posted by "abrarsher23 (via GitHub)" <gi...@apache.org>.
abrarsher23 commented on PR #18288:
URL: https://github.com/apache/iceberg/pull/18288#issuecomment-5891814189
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.
--
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: [PR] Flink: Fix same-checkpoint upserts after schema evolution [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #18288:
URL: https://github.com/apache/iceberg/pull/18288#issuecomment-5941766766
@Asthenia0412 Thanks for taking this on. The approach looks correct to me.
I've also reviewed @abrarsher23's implementation at Affirm/iceberg#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/iceberg#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.
--
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