mxm commented on code in PR #17630:
URL: https://github.com/apache/iceberg/pull/17630#discussion_r3801130387
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -257,6 +281,18 @@ public void processElement(StreamRecord<Trigger> element)
throws Exception {
return;
}
+ // Resolving an eq delete consumes the index entries it matches, so a
cycle that failed after
+ // its delete phase left the index without them. The cursor only
advances once the committer's
+ // marker is on the target branch, so picking the same staging snapshot
again means that cycle
+ // did not commit, and the target has not moved either, so nothing else
rebuilds the index.
+ // This also covers a restore from a checkpoint taken mid-cycle, because
the planned snapshot
+ // is part of the checkpointed state.
+ if (!rebuilt
+ && mainSnapshot != null
+ && Objects.equals(pendingStagingSnapshotId,
nextToProcess.snapshotId())) {
+ rebuildIndex(mainSnapshot, true);
+ }
Review Comment:
This feels a bit complicated. Can we move this logic into
`ensureIndexCurrent` to avoid the indirection of via the `rebuilt` flag?
##########
docs/docs/flink-maintenance.md:
##########
@@ -113,8 +113,8 @@ Notes:
**State and restart**
-- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch. Its size scales with the number of
live rows in the table for the configured equality columns; size Flink state
backend, checkpoint storage, and TaskManager memory accordingly. **RocksDB is
the preferred state backend** because the PK index can grow beyond the
available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
-- Full (re)indexing from the target branch head is only triggered in a few
cases: cold start with no checkpoint, external commits that advance the target
past the currently-indexed snapshot (see below), or when the planner detects
that the target branch has diverged from the indexed snapshot (rollback /
replace-main / expired marker).
+- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch, unless the checkpoint was taken
while a conversion cycle was still in flight (see below). Its size scales with
the number of live rows in the table for the configured equality columns; size
Flink state backend, checkpoint storage, and TaskManager memory accordingly.
**RocksDB is the preferred state backend** because the PK index can grow beyond
the available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
Review Comment:
"unless the checkpoint was taken while a conversion cycle was still in
flight (see below)"
This isn't really possible, unless unaligned checkpointing is used. The
reason is that we emit all commands at once from the planner. Any checkpoint
barriers will come afterwards.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -297,22 +335,37 @@ private void ensureIndexCurrent(Snapshot mainSnapshot) {
bootstrap ? "Bootstrapping" : "Reindexing",
currentMainSnapshotId,
eqFieldIds);
- if (reindex) {
- // Evict keyed entries the reindex will not re-add (e.g. data file
removed by CoW).
- output.collect(
- CLEAR_BROADCAST_STREAM,
- new StreamRecord<>(
- IndexCommand.clearBeforeReindex(
- currentMainSnapshotId, mainSnapshot.sequenceNumber())));
- reindexCounter.inc();
- }
-
- indexSnapshotId = currentMainSnapshotId;
- indexedSequenceNumber = mainSnapshot.sequenceNumber();
- emitMainDataReadCommands(mainSnapshot);
+ rebuildIndex(mainSnapshot, reindex);
}
lastMainSnapshotId = currentMainSnapshotId;
+ return bootstrap || reindex;
+ }
+
+ /**
+ * Re-emits every data row on {@code mainSnapshot} so the worker's index
holds all their positions
+ * again, optionally preceded by a CLEAR_INDEX broadcast that evicts keyed
entries the re-emission
+ * will not re-add (e.g. a PK whose data file was removed by a CoW commit).
A bootstrap has no
+ * earlier index and so nothing to evict.
+ *
+ * <p>The worker detects stale state by comparing the generation stamped on
the commands it
+ * receives with the one it stored, so every rebuild hands out a higher
generation, including a
+ * rebuild while the target branch stands still.
+ */
+ private void rebuildIndex(Snapshot mainSnapshot, boolean evictStaleKeys) {
+ long generation = indexGeneration == null ? 1 : indexGeneration + 1;
Review Comment:
Can we remove the null check and ensure `indexGeneration` is initialized
properly from state?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java:
##########
@@ -1342,6 +1342,131 @@ void testReaderErrorSkipsCommit() throws Exception {
}
}
+ @Test
+ void testDeleteResolvedBeforeFailureIsRetained() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ // Two eq deletes in one commit. id=1's delete file stays readable so it
resolves, while id=2's
+ // is removed so the cycle aborts only after id=1 has been resolved out of
the index.
+ DeleteFile readableDelete = writeEqualityDelete(table, 1, "a");
+ DeleteFile missingDelete = writeEqualityDelete(table, 2, "b");
Review Comment:
Should we also add an insert for the "a"/"b" here in addition to the delete?
##########
docs/docs/flink-maintenance.md:
##########
@@ -113,8 +113,8 @@ Notes:
**State and restart**
-- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch. Its size scales with the number of
live rows in the table for the configured equality columns; size Flink state
backend, checkpoint storage, and TaskManager memory accordingly. **RocksDB is
the preferred state backend** because the PK index can grow beyond the
available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
-- Full (re)indexing from the target branch head is only triggered in a few
cases: cold start with no checkpoint, external commits that advance the target
past the currently-indexed snapshot (see below), or when the planner detects
that the target branch has diverged from the indexed snapshot (rollback /
replace-main / expired marker).
+- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch, unless the checkpoint was taken
while a conversion cycle was still in flight (see below). Its size scales with
the number of live rows in the table for the configured equality columns; size
Flink state backend, checkpoint storage, and TaskManager memory accordingly.
**RocksDB is the preferred state backend** because the PK index can grow beyond
the available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
+- Full (re)indexing from the target branch head is only triggered in a few
cases: cold start with no checkpoint, external commits that advance the target
past the currently-indexed snapshot (see below), when the planner detects that
the target branch has diverged from the indexed snapshot (rollback /
replace-main / expired marker), or when a staging snapshot is planned again
because the cycle that planned it did not commit. The last case covers both a
failed cycle and a restore from a checkpoint taken mid-cycle: resolving an
eq-delete consumes the index entries it matched, so those entries have to be
restored before the delete is resolved again.
Review Comment:
Maybe remove the last sentence, not sure we need that much detail here.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java:
##########
@@ -149,15 +150,15 @@ public void open(OpenContext context) throws Exception {
public void processElement(IndexCommand cmd, ReadOnlyContext ctx,
Collector<DVPosition> out)
throws Exception {
try {
- Long storedSequence = mainSequenceVersion.value();
- if (!Objects.equals(storedSequence, cmd.mainSequenceNumber())) {
+ Long storedGeneration = indexGeneration.value();
+ if (!Objects.equals(storedGeneration, cmd.indexGeneration())) {
LOG.info(
"Main sequence changed from {} to {} (snapshot {}), clearing
state",
Review Comment:
```suggestion
"Index generation changed from {} to {} (snapshot {}), clearing
state",
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -126,7 +130,10 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private transient Long lastMainSnapshotId;
private transient Long lastStagingSnapshotId;
private transient Long indexSnapshotId;
- private transient Long indexedSequenceNumber;
+ private transient Long indexGeneration;
+ // Staging snapshot the last emitted plan covered, checkpointed so it
survives a restore taken
+ // mid-cycle. Selecting it again means that cycle never committed.
+ private transient Long pendingStagingSnapshotId;
Review Comment:
Should we add a unit test to verify behavior regarding the new field?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
