mxm commented on code in PR #17630:
URL: https://github.com/apache/iceberg/pull/17630#discussion_r3776259412
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -127,6 +131,9 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private transient Long lastStagingSnapshotId;
private transient Long indexSnapshotId;
private transient Long indexedSequenceNumber;
Review Comment:
```suggestion
private transient Long indexGeneration;
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -297,22 +334,41 @@ 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 sequence number
stamped on the commands it
+ * receives with the one it stored, so a rebuild while the target branch
stands still must carry a
+ * value above the last one. The index only ever compares this number; it is
never matched against
+ * a data or delete sequence number, which the delete semantics use instead.
+ */
+ private void rebuildIndex(Snapshot mainSnapshot, boolean evictStaleKeys) {
+ long rebuildSequenceNumber =
+ indexedSequenceNumber == null
+ ? mainSnapshot.sequenceNumber()
+ : Math.max(mainSnapshot.sequenceNumber(), indexedSequenceNumber +
1);
Review Comment:
Let's rename `sequenceNumber` to `generation` and remove the `Max.max`
logic, i.e. `indexSequenceNumber = mainSnapshot.generation() + 1`. The sequence
number we used initially but the logic has diverted from it.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -127,6 +131,9 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private transient Long lastStagingSnapshotId;
private transient Long indexSnapshotId;
private transient Long indexedSequenceNumber;
+ // 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 plannedStagingSnapshotId;
Review Comment:
Should this be:
```suggestion
private transient Long pendingStagingSnapshotId;
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -127,6 +131,9 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private transient Long lastStagingSnapshotId;
private transient Long indexSnapshotId;
private transient Long indexedSequenceNumber;
+ // 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 plannedStagingSnapshotId;
Review Comment:
Should this be:
```suggestion
private transient Long pendingStagingSnapshotId;
```
--
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]
