You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by mx...@apache.org on 2026/08/30 09:23:42 UTC
(iceberg) 01/02: Flink: Backport eq-delete conversion replan reindex fix to 1.20 and 2.2 (#17630)
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 883ea5b748779f01f702c10ccb412b2335781696
Author: vishnuprakaz <nv...@gmail.com>
AuthorDate: Sat Aug 29 17:55:35 2026 +0530
Flink: Backport eq-delete conversion replan reindex fix to 1.20 and 2.2 (#17630)
---
.../operator/EqualityConvertPKIndex.java | 53 +++----
.../operator/EqualityConvertPlanner.java | 163 ++++++++++++++-------
.../operator/EqualityConvertReader.java | 16 +-
.../flink/maintenance/operator/IndexCommand.java | 23 ++-
.../flink/maintenance/operator/ReadCommand.java | 16 +-
.../api/TestConvertEqualityDeletes.java | 149 +++++++++++++++++--
.../operator/TestEqualityConvertPlanner.java | 82 +++++++++--
.../operator/TestEqualityConvertReader.java | 6 +-
.../operator/EqualityConvertPKIndex.java | 53 +++----
.../operator/EqualityConvertPlanner.java | 163 ++++++++++++++-------
.../operator/EqualityConvertReader.java | 16 +-
.../flink/maintenance/operator/IndexCommand.java | 23 ++-
.../flink/maintenance/operator/ReadCommand.java | 16 +-
.../api/TestConvertEqualityDeletes.java | 149 +++++++++++++++++--
.../operator/TestEqualityConvertPlanner.java | 82 +++++++++--
.../operator/TestEqualityConvertReader.java | 6 +-
16 files changed, 742 insertions(+), 274 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
index 5c949122d9..ac3b51bfdc 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
@@ -58,21 +58,22 @@ import org.slf4j.LoggerFactory;
* On a separate target branch the committer reassigns data sequence numbers, so every match is
* deleted; event-time ordering prevents over-deletion.
*
- * <p>Stale-index protection runs on two levels. Each key tracks the main sequence number its stored
- * positions were indexed against. The sequence number is unique and monotonic per snapshot, so it
- * serves both the equality test below and the ordering test:
+ * <p>Stale-index protection runs on two levels. Each key tracks the generation the commands
+ * carrying its stored positions were stamped with. The planner hands out a strictly higher
+ * generation on every rebuild of the index, so it serves both the equality test below and the
+ * ordering test:
*
* <ul>
* <li><b>Lazy (per-key)</b>: any keyed command at the top of {@link #processElement} whose {@link
- * IndexCommand#mainSequenceNumber()} differs from the stored one clears stale state and
- * adopts the command's sequence number. Equality suffices here. Required because the
- * broadcast and keyed inputs are independent streams with no ordering guarantee; without it,
- * an ADD_DATA_ROW that arrived before the broadcast would be wrongly evicted.
+ * IndexCommand#indexGeneration()} differs from the stored one clears stale state and adopts
+ * the command's generation. Equality suffices here. Required because the broadcast and keyed
+ * inputs are independent streams with no ordering guarantee; without it, an ADD_DATA_ROW that
+ * arrived before the broadcast would be wrongly evicted.
* <li><b>Eager (all keys)</b>: a CLEAR_INDEX broadcast iterates all keys on the subtask via
* {@link KeyedBroadcastProcessFunction.Context#applyToKeyedState} and clears any whose stored
- * sequence number is older than the broadcast's. Staleness is ordered by sequence number.
- * Bounds state size for PKs that were removed from main by an external CoW commit and won't
- * receive any keyed command next cycle.
+ * generation is older than the broadcast's. Staleness is ordered by generation. Bounds state
+ * size for PKs that were removed from main by an external CoW commit and won't receive any
+ * keyed command next cycle.
* </ul>
*/
@Internal
@@ -89,8 +90,8 @@ public class EqualityConvertPKIndex
public static final MapStateDescriptor<Void, Void> CLEAR_BROADCAST_DESCRIPTOR =
new MapStateDescriptor<>("eq-convert-clear-broadcast", Types.VOID, Types.VOID);
- private static final ValueStateDescriptor<Long> MAIN_SEQUENCE_VERSION_DESCRIPTOR =
- new ValueStateDescriptor<>("mainSequenceVersion", Types.LONG);
+ private static final ValueStateDescriptor<Long> INDEX_GENERATION_DESCRIPTOR =
+ new ValueStateDescriptor<>("indexGeneration", Types.LONG);
private static final ListStateDescriptor<DVPosition> DATA_ROW_POSITIONS_DESCRIPTOR =
new ListStateDescriptor<>("filePositions", TypeInformation.of(DVPosition.class));
private static final ListStateDescriptor<DVPosition> BUFFERED_ROWS_DESCRIPTOR =
@@ -102,7 +103,7 @@ public class EqualityConvertPKIndex
private static final ListStateDescriptor<Integer> RESOLVE_SPEC_IDS_DESCRIPTOR =
new ListStateDescriptor<>("resolveSpecIds", Types.INT);
- private transient ValueState<Long> mainSequenceVersion;
+ private transient ValueState<Long> indexGeneration;
// Resolvable rows for this key. Populated immediately for main data, or from onTimer for
// staging rows once their phase watermark passes, so a delete never resolves against a row from a
// later phase.
@@ -132,7 +133,7 @@ public class EqualityConvertPKIndex
@Override
public void open(OpenContext context) throws Exception {
super.open(context);
- mainSequenceVersion = getRuntimeContext().getState(MAIN_SEQUENCE_VERSION_DESCRIPTOR);
+ indexGeneration = getRuntimeContext().getState(INDEX_GENERATION_DESCRIPTOR);
dataRowPositions = getRuntimeContext().getListState(DATA_ROW_POSITIONS_DESCRIPTOR);
bufferedRows = getRuntimeContext().getListState(BUFFERED_ROWS_DESCRIPTOR);
resolveTimestamp = getRuntimeContext().getState(RESOLVE_TIMESTAMP_DESCRIPTOR);
@@ -149,15 +150,15 @@ public class EqualityConvertPKIndex
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",
- storedSequence,
- cmd.mainSequenceNumber(),
+ "Index generation changed from {} to {} (snapshot {}), clearing state",
+ storedGeneration,
+ cmd.indexGeneration(),
cmd.mainSnapshotId());
clearKeyState();
- mainSequenceVersion.update(cmd.mainSequenceNumber());
+ indexGeneration.update(cmd.indexGeneration());
}
long ts = ctx.timestamp();
@@ -204,15 +205,15 @@ public class EqualityConvertPKIndex
"Broadcast element must be %s",
IndexCommand.Type.CLEAR_INDEX);
- final long broadcastSequenceNumber = cmd.mainSequenceNumber();
+ final long broadcastGeneration = cmd.indexGeneration();
try {
ctx.applyToKeyedState(
- MAIN_SEQUENCE_VERSION_DESCRIPTOR,
- (key, sequenceState) -> {
- Long storedSequenceNumber = sequenceState.value();
- if (storedSequenceNumber != null && storedSequenceNumber < broadcastSequenceNumber) {
+ INDEX_GENERATION_DESCRIPTOR,
+ (key, generationState) -> {
+ Long storedGeneration = generationState.value();
+ if (storedGeneration != null && storedGeneration < broadcastGeneration) {
clearKeyState();
- sequenceState.update(broadcastSequenceNumber);
+ generationState.update(broadcastGeneration);
eagerlyEvictedKeyNumCounter.inc();
}
});
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
index 8499f689b2..82791bae4b 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
@@ -65,12 +65,14 @@ import org.slf4j.LoggerFactory;
* staging snapshot that hasn't been converted yet and emits {@link ReadCommand}s describing the
* files its downstream readers and workers must process.
*
- * <p>Each trigger runs two steps in order:
+ * <p>Each trigger runs three steps in order:
*
* <ol>
- * <li>{@link #ensureIndexCurrent}: updates {@link #lastStagingSnapshotId} from main's history,
- * bootstraps the worker index from main on first run, and reindexes when external commits
- * (e.g. compaction) have advanced main past the currently-indexed snapshot.
+ * <li>{@link #refreshStagingCursor}: updates {@link #lastStagingSnapshotId} from the most recent
+ * committer marker on the target branch.
+ * <li>{@link #ensureIndexCurrent}: rebuilds the worker index from main when there is no index
+ * yet, when external commits (e.g. compaction) have advanced main, or when the staging
+ * snapshot picked for this cycle is the one the previous cycle planned.
* <li>{@link #processStagingSnapshot}: resolve the chosen staging snapshot's eq deletes against
* the (now-current) index, pass through any DV files, and index the snapshot's new data files
* for the next cycle.
@@ -116,8 +118,10 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
// Main snapshot id the worker's index reflects.
private transient ListState<Long> indexSnapshotState;
- // Main sequence number the worker's index reflects.
- private transient ListState<Long> indexedSequenceNumberState;
+ // Generation of the worker's current index build.
+ private transient ListState<Long> indexGenerationState;
+ // Staging snapshot the last emitted plan covered.
+ private transient ListState<Long> pendingStagingSnapshotState;
// Equality field IDs the index was built with, allows to detect reconfiguration.
private transient ListState<Integer> eqFieldIdsState;
@@ -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;
private transient long nextPhaseTs;
@@ -189,19 +196,20 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
indexSnapshotId = stateValue;
}
- indexedSequenceNumberState =
+ indexGenerationState =
context
.getOperatorStateStore()
- .getListState(new ListStateDescriptor<>("indexedSequenceNumber", Types.LONG));
+ .getListState(new ListStateDescriptor<>("indexGeneration", Types.LONG));
- indexedSequenceNumber = null;
- for (Long stateValue : indexedSequenceNumberState.get()) {
+ Long restoredGeneration = null;
+ for (Long stateValue : indexGenerationState.get()) {
Preconditions.checkState(
- indexedSequenceNumber == null,
- "indexedSequenceNumber state should hold at most one value");
- indexedSequenceNumber = stateValue;
+ restoredGeneration == null, "indexGeneration state should hold at most one value");
+ restoredGeneration = stateValue;
}
+ indexGeneration = restoredGeneration != null ? restoredGeneration : 0L;
+
eqFieldIdsState =
context
.getOperatorStateStore()
@@ -214,6 +222,19 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
+ "restart from a clean state (no savepoint).",
restoredEqFieldIds,
eqFieldIds);
+
+ pendingStagingSnapshotState =
+ context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("pendingStagingSnapshotId", Types.LONG));
+
+ pendingStagingSnapshotId = null;
+ for (Long stateValue : pendingStagingSnapshotState.get()) {
+ Preconditions.checkState(
+ pendingStagingSnapshotId == null,
+ "pendingStagingSnapshotId state should hold at most one value");
+ pendingStagingSnapshotId = stateValue;
+ }
}
@Override
@@ -224,15 +245,18 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
indexSnapshotState.add(indexSnapshotId);
}
- indexedSequenceNumberState.clear();
- if (indexedSequenceNumber != null) {
- indexedSequenceNumberState.add(indexedSequenceNumber);
- }
+ indexGenerationState.clear();
+ indexGenerationState.add(indexGeneration);
eqFieldIdsState.clear();
for (int id : eqFieldIds) {
eqFieldIdsState.add(id);
}
+
+ pendingStagingSnapshotState.clear();
+ if (pendingStagingSnapshotId != null) {
+ pendingStagingSnapshotState.add(pendingStagingSnapshotId);
+ }
}
@Override
@@ -246,11 +270,13 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
Snapshot mainSnapshot = table.snapshot(targetBranch);
currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() : null;
- ensureIndexCurrent(mainSnapshot);
+ LastCommittedWork committedWork = refreshStagingCursor(mainSnapshot);
Snapshot nextToProcess =
nextUnprocessedStagingSnapshot(table.snapshot(stagingBranch), mainSnapshot);
+ ensureIndexCurrent(mainSnapshot, committedWork, nextToProcess);
+
if (nextToProcess == null) {
LOG.info("Nothing new to convert on staging branch '{}'.", stagingBranch);
emitNoOpResult(triggerTs, currentMainSnapshotId);
@@ -267,52 +293,81 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
}
/**
- * Brings the worker's index up to date with the current state of the target branch:
- *
- * <ul>
- * <li>Updates {@link #lastStagingSnapshotId} from the most recent committer marker on main.
- * <li>Bootstraps the index from main on the first trigger with a non-null main snapshot.
- * <li>Reindexes from main when external commits (e.g. compaction or direct writes) have
- * advanced main past the currently-indexed snapshot.
- * </ul>
- *
- * <p>No-op when main hasn't moved since the last trigger. Otherwise the history walk is bounded
- * to commits added since {@link #lastMainSnapshotId}.
+ * Updates {@link #lastStagingSnapshotId} from the most recent committer marker on the target
+ * branch. Returns the discovered work, or null when the target has not moved since the last
+ * trigger and the cursor therefore cannot have changed.
*/
- private void ensureIndexCurrent(Snapshot mainSnapshot) {
+ private LastCommittedWork refreshStagingCursor(Snapshot mainSnapshot) {
Long currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() : null;
if (Objects.equals(lastMainSnapshotId, currentMainSnapshotId)) {
- return;
+ return null;
}
LastCommittedWork info = discoverLastCommittedWork(mainSnapshot);
updateLastStagingSnapshotId(info);
+ return info;
+ }
- boolean bootstrap = mainSnapshot != null && indexSnapshotId == null;
- boolean reindex = indexSnapshotId != null && info.externalCommitCount() > 0;
- if (bootstrap || reindex) {
+ /**
+ * Rebuilds the worker index when it is missing, when external commits have advanced the target
+ * branch, or when {@code nextToProcess} is the staging snapshot the previous plan covered.
+ * 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 planning the same staging snapshot again means that cycle did not
+ * commit and nothing else has rebuilt the index.
+ */
+ private void ensureIndexCurrent(
+ Snapshot mainSnapshot, LastCommittedWork committedWork, Snapshot nextToProcess) {
+ if (mainSnapshot == null) {
+ lastMainSnapshotId = null;
+ return;
+ }
+
+ boolean bootstrap = indexSnapshotId == null;
+ boolean reindex =
+ !bootstrap && committedWork != null && committedWork.externalCommitCount() > 0;
+ boolean replan =
+ !bootstrap
+ && nextToProcess != null
+ && Objects.equals(pendingStagingSnapshotId, nextToProcess.snapshotId());
+
+ if (bootstrap || reindex || replan) {
LOG.info(
"{} worker index from main snapshot {} for field IDs {}.",
bootstrap ? "Bootstrapping" : "Reindexing",
- currentMainSnapshotId,
+ mainSnapshot.snapshotId(),
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();
- }
+ rebuildIndex(mainSnapshot, !bootstrap);
+ }
- indexSnapshotId = currentMainSnapshotId;
- indexedSequenceNumber = mainSnapshot.sequenceNumber();
- emitMainDataReadCommands(mainSnapshot);
+ lastMainSnapshotId = mainSnapshot.snapshotId();
+ }
+
+ /**
+ * 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 + 1;
+
+ if (evictStaleKeys) {
+ output.collect(
+ CLEAR_BROADCAST_STREAM,
+ new StreamRecord<>(
+ IndexCommand.clearBeforeReindex(mainSnapshot.snapshotId(), generation)));
+ reindexCounter.inc();
}
- lastMainSnapshotId = currentMainSnapshotId;
+ indexSnapshotId = mainSnapshot.snapshotId();
+ indexGeneration = generation;
+ emitMainDataReadCommands(mainSnapshot);
}
private void updateLastStagingSnapshotId(LastCommittedWork info) {
@@ -445,6 +500,10 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
"Staging snapshot %s has no convertible inputs; shouldSkip should have filtered it.",
stagingSnapshot.snapshotId());
+ // Recorded only once the inputs are known to be convertible: a snapshot that fails validation
+ // consumes no index entries, so it must not make later triggers rebuild the index.
+ pendingStagingSnapshotId = stagingSnapshot.snapshotId();
+
emitDeletePhase(inputs.eqDeleteFiles());
emitSnapshotDataPhase(inputs.newDataFiles());
@@ -572,7 +631,7 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
deleteFile,
spec,
indexSnapshotId,
- indexedSequenceNumber,
+ indexGeneration,
dataSequenceNumber(deleteFile)),
nextPhaseTs));
processedEqDeleteFileNumCounter.inc();
@@ -593,7 +652,7 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
ReadCommand.stagingDataFile(
new FlinkAddedRowsScanTask(dataFile, spec),
indexSnapshotId,
- indexedSequenceNumber,
+ indexGeneration,
dataSequenceNumber(dataFile)),
nextPhaseTs));
}
@@ -632,7 +691,7 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
output.collect(
new StreamRecord<>(
ReadCommand.dataFile(
- task, indexSnapshotId, indexedSequenceNumber, dataSequenceNumber(task.file())),
+ task, indexSnapshotId, indexGeneration, dataSequenceNumber(task.file())),
nextPhaseTs));
}
} catch (IOException e) {
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
index 809e60a8f8..a0c1752fdf 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
@@ -115,17 +115,13 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
processDataFile(
dataTask,
cmd.mainSnapshotId(),
- cmd.mainSequenceNumber(),
+ cmd.indexGeneration(),
cmd.dataSequenceNumber(),
cmd.staging(),
out);
} else if (task instanceof EqualityDeleteFileScanTask deleteTask) {
processDeleteFile(
- deleteTask,
- cmd.mainSnapshotId(),
- cmd.mainSequenceNumber(),
- cmd.dataSequenceNumber(),
- out);
+ deleteTask, cmd.mainSnapshotId(), cmd.indexGeneration(), cmd.dataSequenceNumber(), out);
} else {
throw new IllegalStateException(
"Unexpected ContentScanTask type: " + task.getClass().getName());
@@ -140,7 +136,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
private void processDataFile(
FileScanTask task,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber,
boolean staging,
Collector<IndexCommand> out)
@@ -173,7 +169,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
out.collect(
IndexCommand.addDataRow(
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
key,
file.location(),
position,
@@ -188,7 +184,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
private void processDeleteFile(
EqualityDeleteFileScanTask task,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber,
Collector<IndexCommand> out)
throws IOException {
@@ -208,7 +204,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
SerializedEqualityValues key = fieldSerializer.serializeKey(record, keySchema.asStruct());
out.collect(
IndexCommand.resolveDelete(
- mainSnapshotId, mainSequenceNumber, key, dataSequenceNumber, deleteSpecId));
+ mainSnapshotId, indexGeneration, key, dataSequenceNumber, deleteSpecId));
}
}
}
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
index 509812cbb3..0781fa2723 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
@@ -32,10 +32,10 @@ import org.apache.flink.annotation.Internal;
* emits {@link DVPosition}s for all matching rows. All three flow through the keyed stream and
* route via {@link #key}.
*
- * <p>{@link Type#CLEAR_INDEX} is emitted on the broadcast side when an external commit has advanced
- * the main branch and the worker must evict keyed entries that won't be re-added by the upcoming
- * reindex (e.g. PKs whose data file was removed by CoW). Has no {@link #key} or row position.
- * Carries the new {@link #mainSequenceNumber} as the staleness threshold.
+ * <p>{@link Type#CLEAR_INDEX} is emitted on the broadcast side when the index is rebuilt and the
+ * worker must evict keyed entries that won't be re-added by that rebuild (e.g. PKs whose data file
+ * was removed by CoW). Has no {@link #key} or row position. Carries the new {@link
+ * #indexGeneration} as the staleness threshold.
*
* <p>{@link #rowPosition} is the data row's location, set for the two add types and null otherwise;
* the data sequence number it carries lets the worker apply a delete only to older rows. {@code
@@ -49,7 +49,7 @@ import org.apache.flink.annotation.Internal;
public record IndexCommand(
Type type,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
SerializedEqualityValues key,
DVPosition rowPosition,
long deleteSequenceNumber,
@@ -68,7 +68,7 @@ public record IndexCommand(
public static IndexCommand addDataRow(
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
SerializedEqualityValues key,
String filePath,
long position,
@@ -79,7 +79,7 @@ public record IndexCommand(
return new IndexCommand(
staging ? Type.ADD_STAGING_DATA_ROW : Type.ADD_DATA_ROW,
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
key,
new DVPosition(filePath, position, specId, partition, dataSequenceNumber),
-1,
@@ -88,22 +88,21 @@ public record IndexCommand(
public static IndexCommand resolveDelete(
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
SerializedEqualityValues key,
long deleteSequenceNumber,
int deleteSpecId) {
return new IndexCommand(
Type.RESOLVE_DELETE,
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
key,
null,
deleteSequenceNumber,
deleteSpecId);
}
- public static IndexCommand clearBeforeReindex(long mainSnapshotId, long mainSequenceNumber) {
- return new IndexCommand(
- Type.CLEAR_INDEX, mainSnapshotId, mainSequenceNumber, null, null, -1, -1);
+ public static IndexCommand clearBeforeReindex(long mainSnapshotId, long indexGeneration) {
+ return new IndexCommand(Type.CLEAR_INDEX, mainSnapshotId, indexGeneration, null, null, -1, -1);
}
}
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
index e30e851600..5be25949cd 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
@@ -43,7 +43,7 @@ import org.apache.iceberg.PartitionSpec;
*
* <p>{@code mainSnapshotId} is sent for diagnostic output.
*
- * <p>{@code mainSequenceNumber} is used by the index to order eager evictions by.
+ * <p>{@code indexGeneration} is used by the index to order eager evictions by.
*
* <p>{@code dataSequenceNumber} is the wrapped file's sequence number (data file or equality
* delete), propagated to the worker so a delete only deletes rows older than itself.
@@ -55,31 +55,31 @@ import org.apache.iceberg.PartitionSpec;
public record ReadCommand(
ContentScanTask<?> task,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber,
boolean staging)
implements Serializable {
public static ReadCommand dataFile(
- FileScanTask task, Long mainSnapshotId, Long mainSequenceNumber, long dataSequenceNumber) {
- return new ReadCommand(task, mainSnapshotId, mainSequenceNumber, dataSequenceNumber, false);
+ FileScanTask task, Long mainSnapshotId, Long indexGeneration, long dataSequenceNumber) {
+ return new ReadCommand(task, mainSnapshotId, indexGeneration, dataSequenceNumber, false);
}
public static ReadCommand stagingDataFile(
- FileScanTask task, Long mainSnapshotId, Long mainSequenceNumber, long dataSequenceNumber) {
- return new ReadCommand(task, mainSnapshotId, mainSequenceNumber, dataSequenceNumber, true);
+ FileScanTask task, Long mainSnapshotId, Long indexGeneration, long dataSequenceNumber) {
+ return new ReadCommand(task, mainSnapshotId, indexGeneration, dataSequenceNumber, true);
}
public static ReadCommand eqDeleteFile(
DeleteFile file,
PartitionSpec spec,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber) {
return new ReadCommand(
new EqualityDeleteFileScanTask(file, spec),
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
dataSequenceNumber,
false);
}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
index beee65202d..2e03a0655e 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
@@ -64,8 +64,6 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.types.Types;
-import org.apache.iceberg.util.StructLikeSet;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -1342,6 +1340,139 @@ class TestConvertEqualityDeletes extends MaintenanceTaskTestBase {
}
}
+ @Test
+ void testDeleteResolvedBeforeFailureIsRetained() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ // Two eq deletes with their re-inserts, the usual upsert shape. id=1's delete file stays
+ // readable so it resolves; id=2's is removed so the cycle aborts after id=1 is resolved.
+ DataFile reinsertA = writeDataFile(table, createRecord(1, "a"));
+ DataFile reinsertB = writeDataFile(table, createRecord(2, "b"));
+ DeleteFile readableDelete = writeEqualityDelete(table, 1, "a");
+ DeleteFile missingDelete = writeEqualityDelete(table, 2, "b");
+ table
+ .newRowDelta()
+ .addRows(reinsertA)
+ .addRows(reinsertB)
+ .addDeletes(readableDelete)
+ .addDeletes(missingDelete)
+ .commit();
+ table.refresh();
+
+ // The eq deletes hide the original rows, but not the re-inserts.
+ assertRecords(table, ImmutableList.of(createRecord(1, "a"), createRecord(2, "b")));
+
+ long mainSnapshotBeforeConversion = table.currentSnapshot().snapshotId();
+ File missingDeleteLocalFile = new File(missingDelete.location().replace("file:", ""));
+ assertThat(missingDeleteLocalFile.delete()).isTrue();
+
+ appendConvertTask(SnapshotRef.MAIN_BRANCH);
+
+ JobClient jobClient = null;
+ try {
+ jobClient = infra.env().executeAsync();
+
+ long time1 = System.currentTimeMillis();
+ infra.source().sendRecord(Trigger.create(time1, 0), time1);
+ TaskResult result1 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result1.success()).isFalse();
+ table.refresh();
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotBeforeConversion);
+
+ // Rewrite an identical delete file and retry. The cursor did not advance, so the planner
+ // re-processes the same snapshot.
+ DeleteFile recreated =
+ FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(missingDeleteLocalFile),
+ new PartitionData(PartitionSpec.unpartitioned().partitionType()),
+ Lists.newArrayList(createRecord(2, "b")),
+ table.schema());
+ assertThat(recreated.location()).isEqualTo(missingDelete.location());
+
+ long time2 = time1 + 1;
+ infra.source().sendRecord(Trigger.create(time2, 0), time2);
+ TaskResult result2 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result2.exceptions()).isEmpty();
+ assertThat(result2.success()).isTrue();
+
+ table.refresh();
+ // The retried cycle converted both eq deletes, so only the re-inserts remain visible.
+ assertNoEqualityDeletesOnMain(table, 0);
+ assertRecords(table, ImmutableList.of(createRecord(1, "a"), createRecord(2, "b")));
+ } finally {
+ closeJobClient(jobClient);
+ }
+ }
+
+ @Test
+ void testDeleteResolvedBeforeFailureIsRetainedOnSeparateStagingBranch() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ long targetSnapshotBeforeConversion = table.currentSnapshot().snapshotId();
+
+ // Same shape as the in-place case, but the eq deletes live on a separate staging branch. The
+ // target only advances when the converter commits, so nothing else can rebuild the index.
+ DeleteFile readableDelete = writeEqualityDelete(table, 1, "a");
+ DeleteFile missingDelete = writeEqualityDelete(table, 2, "b");
+ table
+ .newRowDelta()
+ .addDeletes(readableDelete)
+ .addDeletes(missingDelete)
+ .toBranch(STAGING_BRANCH)
+ .commit();
+ table.refresh();
+
+ File missingDeleteLocalFile = new File(missingDelete.location().replace("file:", ""));
+ assertThat(missingDeleteLocalFile.delete()).isTrue();
+
+ appendConvertTask(STAGING_BRANCH);
+
+ JobClient jobClient = null;
+ try {
+ jobClient = infra.env().executeAsync();
+
+ long time1 = System.currentTimeMillis();
+ infra.source().sendRecord(Trigger.create(time1, 0), time1);
+ TaskResult result1 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result1.success()).isFalse();
+ table.refresh();
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(targetSnapshotBeforeConversion);
+
+ DeleteFile recreated =
+ FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(missingDeleteLocalFile),
+ new PartitionData(PartitionSpec.unpartitioned().partitionType()),
+ Lists.newArrayList(createRecord(2, "b")),
+ table.schema());
+ assertThat(recreated.location()).isEqualTo(missingDelete.location());
+
+ long time2 = time1 + 1;
+ infra.source().sendRecord(Trigger.create(time2, 0), time2);
+ TaskResult result2 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result2.exceptions()).isEmpty();
+ assertThat(result2.success()).isTrue();
+
+ table.refresh();
+ // Both deletes converted to DVs on the target, so neither row is visible there.
+ assertRecords(table, ImmutableList.of());
+ } finally {
+ closeJobClient(jobClient);
+ }
+ }
+
private void appendConvertTask() {
appendConvertTask(STAGING_BRANCH);
}
@@ -1365,22 +1496,16 @@ class TestConvertEqualityDeletes extends MaintenanceTaskTestBase {
private static void assertRecords(Table table, List<Record> expected) throws IOException {
table.refresh();
- Types.StructType type = SimpleDataUtil.SCHEMA.asStruct();
-
- StructLikeSet expectedSet = StructLikeSet.create(type);
- expectedSet.addAll(expected);
try (CloseableIterable<Record> iterable =
IcebergGenerics.read(table)
.useSnapshot(table.currentSnapshot().snapshotId())
.project(SimpleDataUtil.SCHEMA)
.build()) {
- StructLikeSet actualSet = StructLikeSet.create(type);
- for (Record record : iterable) {
- actualSet.add(record);
- }
-
- assertThat(actualSet).isEqualTo(expectedSet);
+ // rows from files with deletes applied carry an extra _pos field
+ assertThat(Lists.newArrayList(iterable))
+ .map(r -> createRecord((Integer) r.getField("id"), (String) r.getField("data")))
+ .containsExactlyInAnyOrderElementsOf(expected);
}
}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
index 87eb47ac2c..ed243695bc 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
@@ -630,7 +630,7 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
}
@Test
- void noMainReEmitWhenUnchanged() throws Exception {
+ void rebuildsIndexWhenReplanningUncommittedStagingSnapshot() throws Exception {
Table table = createTableWithDelete(3);
insert(table, 1, "a");
insert(table, 2, "b");
@@ -638,34 +638,84 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
table.refresh();
- DeleteFile eqDelete1 = writeEqualityDelete(table, 1, "a");
- table.newRowDelta().addDeletes(eqDelete1).toBranch(STAGING_BRANCH).commit();
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
table.refresh();
try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
createHarness(STAGING_BRANCH)) {
harness.open();
+ // First trigger bootstraps the index and resolves the staging snapshot's eq delete, which
+ // consumes the matching index entries.
sendTrigger(harness);
int firstTriggerCount = harness.extractOutputValues().size();
- // 2 DATA_FILE (main) + 1 EQ_DELETE_FILE
assertThat(firstTriggerCount).isEqualTo(3);
- assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1);
-
- DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
- table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
- table.refresh();
+ // No committer marker on main, so the cycle did not commit and the same staging snapshot is
+ // planned again. Main has not moved either, so the planner must rebuild the index itself.
sendTrigger(harness);
List<ReadCommand> allCommands = harness.extractOutputValues();
List<ReadCommand> trigger2Commands =
allCommands.subList(firstTriggerCount, allCommands.size());
- // Only 1 EQ_DELETE_FILE, no main re-emission
- assertThat(countDataFileTasks(trigger2Commands)).isEqualTo(0);
+ assertThat(countDataFileTasks(trigger2Commands)).isEqualTo(2);
assertThat(countEqDeleteTasks(trigger2Commands)).isEqualTo(1);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void rebuildsIndexAfterRestoreOfUncommittedCycle() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ OperatorSubtaskState state;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // First cycle converts S1 and commits, so the newest commit on the target is the converter's
+ // own marker. The existing reindex path counts only commits newer than that marker, so it
+ // cannot fire on the restore below and the replan check is the only thing that can rebuild.
+ sendTrigger(harness);
+ assertThat(harness.extractOutputValues()).hasSize(3);
+ simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
+ // Second cycle resolves S2's eq delete but never commits, then a checkpoint is taken.
+ DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
+ table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
- assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2);
+ int afterFirstCycle = harness.extractOutputValues().size();
+ sendTrigger(harness);
+ assertThat(harness.extractOutputValues().size()).isGreaterThan(afterFirstCycle);
+
+ state = harness.snapshot(1, System.currentTimeMillis());
+ }
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(state);
+ harness.open();
+
+ // The restored planner replans the same staging snapshot, so it must rebuild the index the
+ // uncommitted cycle consumed.
+ sendTrigger(harness);
+ List<ReadCommand> commands = harness.extractOutputValues();
+
+ // Two inserts plus simulateConvertCommit's marker file = 3 data files on the target.
+ assertThat(countDataFileTasks(commands)).isEqualTo(3);
+ assertThat(countEqDeleteTasks(commands)).isEqualTo(1);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
}
}
@@ -789,7 +839,6 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
table.newAppend().appendFile(externalFile).commit();
table.refresh();
long mainAfterExternal = table.snapshot(SnapshotRef.MAIN_BRANCH).snapshotId();
- long mainSeqAfterExternal = table.snapshot(SnapshotRef.MAIN_BRANCH).sequenceNumber();
DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
@@ -804,7 +853,8 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
assertThat(clears).hasSize(1);
assertThat(clears.get(0).type()).isEqualTo(IndexCommand.Type.CLEAR_INDEX);
assertThat(clears.get(0).mainSnapshotId()).isEqualTo(mainAfterExternal);
- assertThat(clears.get(0).mainSequenceNumber()).isEqualTo(mainSeqAfterExternal);
+ // The bootstrap on the first trigger took generation 1, so the reindex takes the next one.
+ assertThat(clears.get(0).indexGeneration()).isEqualTo(2L);
}
}
@@ -829,6 +879,10 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
int afterFirst = harness.extractOutputValues().size();
assertThat(afterFirst).isGreaterThan(0);
+ // The first cycle commits, so later triggers plan new staging snapshots rather than
+ // replanning this one.
+ simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
// External commit on main (no COMMITTED_STAGING_SNAPSHOT_PROPERTY).
DataFile externalFile =
new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
index d098b9e0b4..9c93b8122a 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
@@ -229,7 +229,7 @@ class TestEqualityConvertReader extends OperatorTestBase {
PartitionSpec spec = table.specs().get(dataFile.specId());
List<Integer> fieldIds = Lists.newArrayList(1);
long mainSnapshotId = 42L;
- long mainSequenceNumber = 7L;
+ long indexGeneration = 7L;
long dataSequenceNumber = 9L;
try (OneInputStreamOperatorTestHarness<ReadCommand, IndexCommand> harness =
@@ -240,14 +240,14 @@ class TestEqualityConvertReader extends OperatorTestBase {
ReadCommand.stagingDataFile(
new FlinkAddedRowsScanTask(dataFile, spec),
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
dataSequenceNumber);
harness.processElement(cmd, 0);
List<IndexCommand> output = harness.extractOutputValues();
assertThat(output).hasSize(1);
assertThat(output.get(0).mainSnapshotId()).isEqualTo(mainSnapshotId);
- assertThat(output.get(0).mainSequenceNumber()).isEqualTo(mainSequenceNumber);
+ assertThat(output.get(0).indexGeneration()).isEqualTo(indexGeneration);
assertThat(output.get(0).rowPosition().dataSequenceNumber()).isEqualTo(dataSequenceNumber);
}
}
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
index 5c949122d9..ac3b51bfdc 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
@@ -58,21 +58,22 @@ import org.slf4j.LoggerFactory;
* On a separate target branch the committer reassigns data sequence numbers, so every match is
* deleted; event-time ordering prevents over-deletion.
*
- * <p>Stale-index protection runs on two levels. Each key tracks the main sequence number its stored
- * positions were indexed against. The sequence number is unique and monotonic per snapshot, so it
- * serves both the equality test below and the ordering test:
+ * <p>Stale-index protection runs on two levels. Each key tracks the generation the commands
+ * carrying its stored positions were stamped with. The planner hands out a strictly higher
+ * generation on every rebuild of the index, so it serves both the equality test below and the
+ * ordering test:
*
* <ul>
* <li><b>Lazy (per-key)</b>: any keyed command at the top of {@link #processElement} whose {@link
- * IndexCommand#mainSequenceNumber()} differs from the stored one clears stale state and
- * adopts the command's sequence number. Equality suffices here. Required because the
- * broadcast and keyed inputs are independent streams with no ordering guarantee; without it,
- * an ADD_DATA_ROW that arrived before the broadcast would be wrongly evicted.
+ * IndexCommand#indexGeneration()} differs from the stored one clears stale state and adopts
+ * the command's generation. Equality suffices here. Required because the broadcast and keyed
+ * inputs are independent streams with no ordering guarantee; without it, an ADD_DATA_ROW that
+ * arrived before the broadcast would be wrongly evicted.
* <li><b>Eager (all keys)</b>: a CLEAR_INDEX broadcast iterates all keys on the subtask via
* {@link KeyedBroadcastProcessFunction.Context#applyToKeyedState} and clears any whose stored
- * sequence number is older than the broadcast's. Staleness is ordered by sequence number.
- * Bounds state size for PKs that were removed from main by an external CoW commit and won't
- * receive any keyed command next cycle.
+ * generation is older than the broadcast's. Staleness is ordered by generation. Bounds state
+ * size for PKs that were removed from main by an external CoW commit and won't receive any
+ * keyed command next cycle.
* </ul>
*/
@Internal
@@ -89,8 +90,8 @@ public class EqualityConvertPKIndex
public static final MapStateDescriptor<Void, Void> CLEAR_BROADCAST_DESCRIPTOR =
new MapStateDescriptor<>("eq-convert-clear-broadcast", Types.VOID, Types.VOID);
- private static final ValueStateDescriptor<Long> MAIN_SEQUENCE_VERSION_DESCRIPTOR =
- new ValueStateDescriptor<>("mainSequenceVersion", Types.LONG);
+ private static final ValueStateDescriptor<Long> INDEX_GENERATION_DESCRIPTOR =
+ new ValueStateDescriptor<>("indexGeneration", Types.LONG);
private static final ListStateDescriptor<DVPosition> DATA_ROW_POSITIONS_DESCRIPTOR =
new ListStateDescriptor<>("filePositions", TypeInformation.of(DVPosition.class));
private static final ListStateDescriptor<DVPosition> BUFFERED_ROWS_DESCRIPTOR =
@@ -102,7 +103,7 @@ public class EqualityConvertPKIndex
private static final ListStateDescriptor<Integer> RESOLVE_SPEC_IDS_DESCRIPTOR =
new ListStateDescriptor<>("resolveSpecIds", Types.INT);
- private transient ValueState<Long> mainSequenceVersion;
+ private transient ValueState<Long> indexGeneration;
// Resolvable rows for this key. Populated immediately for main data, or from onTimer for
// staging rows once their phase watermark passes, so a delete never resolves against a row from a
// later phase.
@@ -132,7 +133,7 @@ public class EqualityConvertPKIndex
@Override
public void open(OpenContext context) throws Exception {
super.open(context);
- mainSequenceVersion = getRuntimeContext().getState(MAIN_SEQUENCE_VERSION_DESCRIPTOR);
+ indexGeneration = getRuntimeContext().getState(INDEX_GENERATION_DESCRIPTOR);
dataRowPositions = getRuntimeContext().getListState(DATA_ROW_POSITIONS_DESCRIPTOR);
bufferedRows = getRuntimeContext().getListState(BUFFERED_ROWS_DESCRIPTOR);
resolveTimestamp = getRuntimeContext().getState(RESOLVE_TIMESTAMP_DESCRIPTOR);
@@ -149,15 +150,15 @@ public class EqualityConvertPKIndex
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",
- storedSequence,
- cmd.mainSequenceNumber(),
+ "Index generation changed from {} to {} (snapshot {}), clearing state",
+ storedGeneration,
+ cmd.indexGeneration(),
cmd.mainSnapshotId());
clearKeyState();
- mainSequenceVersion.update(cmd.mainSequenceNumber());
+ indexGeneration.update(cmd.indexGeneration());
}
long ts = ctx.timestamp();
@@ -204,15 +205,15 @@ public class EqualityConvertPKIndex
"Broadcast element must be %s",
IndexCommand.Type.CLEAR_INDEX);
- final long broadcastSequenceNumber = cmd.mainSequenceNumber();
+ final long broadcastGeneration = cmd.indexGeneration();
try {
ctx.applyToKeyedState(
- MAIN_SEQUENCE_VERSION_DESCRIPTOR,
- (key, sequenceState) -> {
- Long storedSequenceNumber = sequenceState.value();
- if (storedSequenceNumber != null && storedSequenceNumber < broadcastSequenceNumber) {
+ INDEX_GENERATION_DESCRIPTOR,
+ (key, generationState) -> {
+ Long storedGeneration = generationState.value();
+ if (storedGeneration != null && storedGeneration < broadcastGeneration) {
clearKeyState();
- sequenceState.update(broadcastSequenceNumber);
+ generationState.update(broadcastGeneration);
eagerlyEvictedKeyNumCounter.inc();
}
});
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
index 8499f689b2..82791bae4b 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
@@ -65,12 +65,14 @@ import org.slf4j.LoggerFactory;
* staging snapshot that hasn't been converted yet and emits {@link ReadCommand}s describing the
* files its downstream readers and workers must process.
*
- * <p>Each trigger runs two steps in order:
+ * <p>Each trigger runs three steps in order:
*
* <ol>
- * <li>{@link #ensureIndexCurrent}: updates {@link #lastStagingSnapshotId} from main's history,
- * bootstraps the worker index from main on first run, and reindexes when external commits
- * (e.g. compaction) have advanced main past the currently-indexed snapshot.
+ * <li>{@link #refreshStagingCursor}: updates {@link #lastStagingSnapshotId} from the most recent
+ * committer marker on the target branch.
+ * <li>{@link #ensureIndexCurrent}: rebuilds the worker index from main when there is no index
+ * yet, when external commits (e.g. compaction) have advanced main, or when the staging
+ * snapshot picked for this cycle is the one the previous cycle planned.
* <li>{@link #processStagingSnapshot}: resolve the chosen staging snapshot's eq deletes against
* the (now-current) index, pass through any DV files, and index the snapshot's new data files
* for the next cycle.
@@ -116,8 +118,10 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
// Main snapshot id the worker's index reflects.
private transient ListState<Long> indexSnapshotState;
- // Main sequence number the worker's index reflects.
- private transient ListState<Long> indexedSequenceNumberState;
+ // Generation of the worker's current index build.
+ private transient ListState<Long> indexGenerationState;
+ // Staging snapshot the last emitted plan covered.
+ private transient ListState<Long> pendingStagingSnapshotState;
// Equality field IDs the index was built with, allows to detect reconfiguration.
private transient ListState<Integer> eqFieldIdsState;
@@ -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;
private transient long nextPhaseTs;
@@ -189,19 +196,20 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
indexSnapshotId = stateValue;
}
- indexedSequenceNumberState =
+ indexGenerationState =
context
.getOperatorStateStore()
- .getListState(new ListStateDescriptor<>("indexedSequenceNumber", Types.LONG));
+ .getListState(new ListStateDescriptor<>("indexGeneration", Types.LONG));
- indexedSequenceNumber = null;
- for (Long stateValue : indexedSequenceNumberState.get()) {
+ Long restoredGeneration = null;
+ for (Long stateValue : indexGenerationState.get()) {
Preconditions.checkState(
- indexedSequenceNumber == null,
- "indexedSequenceNumber state should hold at most one value");
- indexedSequenceNumber = stateValue;
+ restoredGeneration == null, "indexGeneration state should hold at most one value");
+ restoredGeneration = stateValue;
}
+ indexGeneration = restoredGeneration != null ? restoredGeneration : 0L;
+
eqFieldIdsState =
context
.getOperatorStateStore()
@@ -214,6 +222,19 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
+ "restart from a clean state (no savepoint).",
restoredEqFieldIds,
eqFieldIds);
+
+ pendingStagingSnapshotState =
+ context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("pendingStagingSnapshotId", Types.LONG));
+
+ pendingStagingSnapshotId = null;
+ for (Long stateValue : pendingStagingSnapshotState.get()) {
+ Preconditions.checkState(
+ pendingStagingSnapshotId == null,
+ "pendingStagingSnapshotId state should hold at most one value");
+ pendingStagingSnapshotId = stateValue;
+ }
}
@Override
@@ -224,15 +245,18 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
indexSnapshotState.add(indexSnapshotId);
}
- indexedSequenceNumberState.clear();
- if (indexedSequenceNumber != null) {
- indexedSequenceNumberState.add(indexedSequenceNumber);
- }
+ indexGenerationState.clear();
+ indexGenerationState.add(indexGeneration);
eqFieldIdsState.clear();
for (int id : eqFieldIds) {
eqFieldIdsState.add(id);
}
+
+ pendingStagingSnapshotState.clear();
+ if (pendingStagingSnapshotId != null) {
+ pendingStagingSnapshotState.add(pendingStagingSnapshotId);
+ }
}
@Override
@@ -246,11 +270,13 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
Snapshot mainSnapshot = table.snapshot(targetBranch);
currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() : null;
- ensureIndexCurrent(mainSnapshot);
+ LastCommittedWork committedWork = refreshStagingCursor(mainSnapshot);
Snapshot nextToProcess =
nextUnprocessedStagingSnapshot(table.snapshot(stagingBranch), mainSnapshot);
+ ensureIndexCurrent(mainSnapshot, committedWork, nextToProcess);
+
if (nextToProcess == null) {
LOG.info("Nothing new to convert on staging branch '{}'.", stagingBranch);
emitNoOpResult(triggerTs, currentMainSnapshotId);
@@ -267,52 +293,81 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
}
/**
- * Brings the worker's index up to date with the current state of the target branch:
- *
- * <ul>
- * <li>Updates {@link #lastStagingSnapshotId} from the most recent committer marker on main.
- * <li>Bootstraps the index from main on the first trigger with a non-null main snapshot.
- * <li>Reindexes from main when external commits (e.g. compaction or direct writes) have
- * advanced main past the currently-indexed snapshot.
- * </ul>
- *
- * <p>No-op when main hasn't moved since the last trigger. Otherwise the history walk is bounded
- * to commits added since {@link #lastMainSnapshotId}.
+ * Updates {@link #lastStagingSnapshotId} from the most recent committer marker on the target
+ * branch. Returns the discovered work, or null when the target has not moved since the last
+ * trigger and the cursor therefore cannot have changed.
*/
- private void ensureIndexCurrent(Snapshot mainSnapshot) {
+ private LastCommittedWork refreshStagingCursor(Snapshot mainSnapshot) {
Long currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() : null;
if (Objects.equals(lastMainSnapshotId, currentMainSnapshotId)) {
- return;
+ return null;
}
LastCommittedWork info = discoverLastCommittedWork(mainSnapshot);
updateLastStagingSnapshotId(info);
+ return info;
+ }
- boolean bootstrap = mainSnapshot != null && indexSnapshotId == null;
- boolean reindex = indexSnapshotId != null && info.externalCommitCount() > 0;
- if (bootstrap || reindex) {
+ /**
+ * Rebuilds the worker index when it is missing, when external commits have advanced the target
+ * branch, or when {@code nextToProcess} is the staging snapshot the previous plan covered.
+ * 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 planning the same staging snapshot again means that cycle did not
+ * commit and nothing else has rebuilt the index.
+ */
+ private void ensureIndexCurrent(
+ Snapshot mainSnapshot, LastCommittedWork committedWork, Snapshot nextToProcess) {
+ if (mainSnapshot == null) {
+ lastMainSnapshotId = null;
+ return;
+ }
+
+ boolean bootstrap = indexSnapshotId == null;
+ boolean reindex =
+ !bootstrap && committedWork != null && committedWork.externalCommitCount() > 0;
+ boolean replan =
+ !bootstrap
+ && nextToProcess != null
+ && Objects.equals(pendingStagingSnapshotId, nextToProcess.snapshotId());
+
+ if (bootstrap || reindex || replan) {
LOG.info(
"{} worker index from main snapshot {} for field IDs {}.",
bootstrap ? "Bootstrapping" : "Reindexing",
- currentMainSnapshotId,
+ mainSnapshot.snapshotId(),
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();
- }
+ rebuildIndex(mainSnapshot, !bootstrap);
+ }
- indexSnapshotId = currentMainSnapshotId;
- indexedSequenceNumber = mainSnapshot.sequenceNumber();
- emitMainDataReadCommands(mainSnapshot);
+ lastMainSnapshotId = mainSnapshot.snapshotId();
+ }
+
+ /**
+ * 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 + 1;
+
+ if (evictStaleKeys) {
+ output.collect(
+ CLEAR_BROADCAST_STREAM,
+ new StreamRecord<>(
+ IndexCommand.clearBeforeReindex(mainSnapshot.snapshotId(), generation)));
+ reindexCounter.inc();
}
- lastMainSnapshotId = currentMainSnapshotId;
+ indexSnapshotId = mainSnapshot.snapshotId();
+ indexGeneration = generation;
+ emitMainDataReadCommands(mainSnapshot);
}
private void updateLastStagingSnapshotId(LastCommittedWork info) {
@@ -445,6 +500,10 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
"Staging snapshot %s has no convertible inputs; shouldSkip should have filtered it.",
stagingSnapshot.snapshotId());
+ // Recorded only once the inputs are known to be convertible: a snapshot that fails validation
+ // consumes no index entries, so it must not make later triggers rebuild the index.
+ pendingStagingSnapshotId = stagingSnapshot.snapshotId();
+
emitDeletePhase(inputs.eqDeleteFiles());
emitSnapshotDataPhase(inputs.newDataFiles());
@@ -572,7 +631,7 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
deleteFile,
spec,
indexSnapshotId,
- indexedSequenceNumber,
+ indexGeneration,
dataSequenceNumber(deleteFile)),
nextPhaseTs));
processedEqDeleteFileNumCounter.inc();
@@ -593,7 +652,7 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
ReadCommand.stagingDataFile(
new FlinkAddedRowsScanTask(dataFile, spec),
indexSnapshotId,
- indexedSequenceNumber,
+ indexGeneration,
dataSequenceNumber(dataFile)),
nextPhaseTs));
}
@@ -632,7 +691,7 @@ public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand>
output.collect(
new StreamRecord<>(
ReadCommand.dataFile(
- task, indexSnapshotId, indexedSequenceNumber, dataSequenceNumber(task.file())),
+ task, indexSnapshotId, indexGeneration, dataSequenceNumber(task.file())),
nextPhaseTs));
}
} catch (IOException e) {
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
index 809e60a8f8..a0c1752fdf 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
@@ -115,17 +115,13 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
processDataFile(
dataTask,
cmd.mainSnapshotId(),
- cmd.mainSequenceNumber(),
+ cmd.indexGeneration(),
cmd.dataSequenceNumber(),
cmd.staging(),
out);
} else if (task instanceof EqualityDeleteFileScanTask deleteTask) {
processDeleteFile(
- deleteTask,
- cmd.mainSnapshotId(),
- cmd.mainSequenceNumber(),
- cmd.dataSequenceNumber(),
- out);
+ deleteTask, cmd.mainSnapshotId(), cmd.indexGeneration(), cmd.dataSequenceNumber(), out);
} else {
throw new IllegalStateException(
"Unexpected ContentScanTask type: " + task.getClass().getName());
@@ -140,7 +136,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
private void processDataFile(
FileScanTask task,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber,
boolean staging,
Collector<IndexCommand> out)
@@ -173,7 +169,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
out.collect(
IndexCommand.addDataRow(
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
key,
file.location(),
position,
@@ -188,7 +184,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
private void processDeleteFile(
EqualityDeleteFileScanTask task,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber,
Collector<IndexCommand> out)
throws IOException {
@@ -208,7 +204,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
SerializedEqualityValues key = fieldSerializer.serializeKey(record, keySchema.asStruct());
out.collect(
IndexCommand.resolveDelete(
- mainSnapshotId, mainSequenceNumber, key, dataSequenceNumber, deleteSpecId));
+ mainSnapshotId, indexGeneration, key, dataSequenceNumber, deleteSpecId));
}
}
}
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
index 509812cbb3..0781fa2723 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
@@ -32,10 +32,10 @@ import org.apache.flink.annotation.Internal;
* emits {@link DVPosition}s for all matching rows. All three flow through the keyed stream and
* route via {@link #key}.
*
- * <p>{@link Type#CLEAR_INDEX} is emitted on the broadcast side when an external commit has advanced
- * the main branch and the worker must evict keyed entries that won't be re-added by the upcoming
- * reindex (e.g. PKs whose data file was removed by CoW). Has no {@link #key} or row position.
- * Carries the new {@link #mainSequenceNumber} as the staleness threshold.
+ * <p>{@link Type#CLEAR_INDEX} is emitted on the broadcast side when the index is rebuilt and the
+ * worker must evict keyed entries that won't be re-added by that rebuild (e.g. PKs whose data file
+ * was removed by CoW). Has no {@link #key} or row position. Carries the new {@link
+ * #indexGeneration} as the staleness threshold.
*
* <p>{@link #rowPosition} is the data row's location, set for the two add types and null otherwise;
* the data sequence number it carries lets the worker apply a delete only to older rows. {@code
@@ -49,7 +49,7 @@ import org.apache.flink.annotation.Internal;
public record IndexCommand(
Type type,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
SerializedEqualityValues key,
DVPosition rowPosition,
long deleteSequenceNumber,
@@ -68,7 +68,7 @@ public record IndexCommand(
public static IndexCommand addDataRow(
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
SerializedEqualityValues key,
String filePath,
long position,
@@ -79,7 +79,7 @@ public record IndexCommand(
return new IndexCommand(
staging ? Type.ADD_STAGING_DATA_ROW : Type.ADD_DATA_ROW,
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
key,
new DVPosition(filePath, position, specId, partition, dataSequenceNumber),
-1,
@@ -88,22 +88,21 @@ public record IndexCommand(
public static IndexCommand resolveDelete(
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
SerializedEqualityValues key,
long deleteSequenceNumber,
int deleteSpecId) {
return new IndexCommand(
Type.RESOLVE_DELETE,
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
key,
null,
deleteSequenceNumber,
deleteSpecId);
}
- public static IndexCommand clearBeforeReindex(long mainSnapshotId, long mainSequenceNumber) {
- return new IndexCommand(
- Type.CLEAR_INDEX, mainSnapshotId, mainSequenceNumber, null, null, -1, -1);
+ public static IndexCommand clearBeforeReindex(long mainSnapshotId, long indexGeneration) {
+ return new IndexCommand(Type.CLEAR_INDEX, mainSnapshotId, indexGeneration, null, null, -1, -1);
}
}
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
index e30e851600..5be25949cd 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
@@ -43,7 +43,7 @@ import org.apache.iceberg.PartitionSpec;
*
* <p>{@code mainSnapshotId} is sent for diagnostic output.
*
- * <p>{@code mainSequenceNumber} is used by the index to order eager evictions by.
+ * <p>{@code indexGeneration} is used by the index to order eager evictions by.
*
* <p>{@code dataSequenceNumber} is the wrapped file's sequence number (data file or equality
* delete), propagated to the worker so a delete only deletes rows older than itself.
@@ -55,31 +55,31 @@ import org.apache.iceberg.PartitionSpec;
public record ReadCommand(
ContentScanTask<?> task,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber,
boolean staging)
implements Serializable {
public static ReadCommand dataFile(
- FileScanTask task, Long mainSnapshotId, Long mainSequenceNumber, long dataSequenceNumber) {
- return new ReadCommand(task, mainSnapshotId, mainSequenceNumber, dataSequenceNumber, false);
+ FileScanTask task, Long mainSnapshotId, Long indexGeneration, long dataSequenceNumber) {
+ return new ReadCommand(task, mainSnapshotId, indexGeneration, dataSequenceNumber, false);
}
public static ReadCommand stagingDataFile(
- FileScanTask task, Long mainSnapshotId, Long mainSequenceNumber, long dataSequenceNumber) {
- return new ReadCommand(task, mainSnapshotId, mainSequenceNumber, dataSequenceNumber, true);
+ FileScanTask task, Long mainSnapshotId, Long indexGeneration, long dataSequenceNumber) {
+ return new ReadCommand(task, mainSnapshotId, indexGeneration, dataSequenceNumber, true);
}
public static ReadCommand eqDeleteFile(
DeleteFile file,
PartitionSpec spec,
Long mainSnapshotId,
- Long mainSequenceNumber,
+ Long indexGeneration,
long dataSequenceNumber) {
return new ReadCommand(
new EqualityDeleteFileScanTask(file, spec),
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
dataSequenceNumber,
false);
}
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
index beee65202d..2e03a0655e 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
@@ -64,8 +64,6 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.types.Types;
-import org.apache.iceberg.util.StructLikeSet;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -1342,6 +1340,139 @@ class TestConvertEqualityDeletes extends MaintenanceTaskTestBase {
}
}
+ @Test
+ void testDeleteResolvedBeforeFailureIsRetained() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ // Two eq deletes with their re-inserts, the usual upsert shape. id=1's delete file stays
+ // readable so it resolves; id=2's is removed so the cycle aborts after id=1 is resolved.
+ DataFile reinsertA = writeDataFile(table, createRecord(1, "a"));
+ DataFile reinsertB = writeDataFile(table, createRecord(2, "b"));
+ DeleteFile readableDelete = writeEqualityDelete(table, 1, "a");
+ DeleteFile missingDelete = writeEqualityDelete(table, 2, "b");
+ table
+ .newRowDelta()
+ .addRows(reinsertA)
+ .addRows(reinsertB)
+ .addDeletes(readableDelete)
+ .addDeletes(missingDelete)
+ .commit();
+ table.refresh();
+
+ // The eq deletes hide the original rows, but not the re-inserts.
+ assertRecords(table, ImmutableList.of(createRecord(1, "a"), createRecord(2, "b")));
+
+ long mainSnapshotBeforeConversion = table.currentSnapshot().snapshotId();
+ File missingDeleteLocalFile = new File(missingDelete.location().replace("file:", ""));
+ assertThat(missingDeleteLocalFile.delete()).isTrue();
+
+ appendConvertTask(SnapshotRef.MAIN_BRANCH);
+
+ JobClient jobClient = null;
+ try {
+ jobClient = infra.env().executeAsync();
+
+ long time1 = System.currentTimeMillis();
+ infra.source().sendRecord(Trigger.create(time1, 0), time1);
+ TaskResult result1 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result1.success()).isFalse();
+ table.refresh();
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotBeforeConversion);
+
+ // Rewrite an identical delete file and retry. The cursor did not advance, so the planner
+ // re-processes the same snapshot.
+ DeleteFile recreated =
+ FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(missingDeleteLocalFile),
+ new PartitionData(PartitionSpec.unpartitioned().partitionType()),
+ Lists.newArrayList(createRecord(2, "b")),
+ table.schema());
+ assertThat(recreated.location()).isEqualTo(missingDelete.location());
+
+ long time2 = time1 + 1;
+ infra.source().sendRecord(Trigger.create(time2, 0), time2);
+ TaskResult result2 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result2.exceptions()).isEmpty();
+ assertThat(result2.success()).isTrue();
+
+ table.refresh();
+ // The retried cycle converted both eq deletes, so only the re-inserts remain visible.
+ assertNoEqualityDeletesOnMain(table, 0);
+ assertRecords(table, ImmutableList.of(createRecord(1, "a"), createRecord(2, "b")));
+ } finally {
+ closeJobClient(jobClient);
+ }
+ }
+
+ @Test
+ void testDeleteResolvedBeforeFailureIsRetainedOnSeparateStagingBranch() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ long targetSnapshotBeforeConversion = table.currentSnapshot().snapshotId();
+
+ // Same shape as the in-place case, but the eq deletes live on a separate staging branch. The
+ // target only advances when the converter commits, so nothing else can rebuild the index.
+ DeleteFile readableDelete = writeEqualityDelete(table, 1, "a");
+ DeleteFile missingDelete = writeEqualityDelete(table, 2, "b");
+ table
+ .newRowDelta()
+ .addDeletes(readableDelete)
+ .addDeletes(missingDelete)
+ .toBranch(STAGING_BRANCH)
+ .commit();
+ table.refresh();
+
+ File missingDeleteLocalFile = new File(missingDelete.location().replace("file:", ""));
+ assertThat(missingDeleteLocalFile.delete()).isTrue();
+
+ appendConvertTask(STAGING_BRANCH);
+
+ JobClient jobClient = null;
+ try {
+ jobClient = infra.env().executeAsync();
+
+ long time1 = System.currentTimeMillis();
+ infra.source().sendRecord(Trigger.create(time1, 0), time1);
+ TaskResult result1 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result1.success()).isFalse();
+ table.refresh();
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(targetSnapshotBeforeConversion);
+
+ DeleteFile recreated =
+ FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(missingDeleteLocalFile),
+ new PartitionData(PartitionSpec.unpartitioned().partitionType()),
+ Lists.newArrayList(createRecord(2, "b")),
+ table.schema());
+ assertThat(recreated.location()).isEqualTo(missingDelete.location());
+
+ long time2 = time1 + 1;
+ infra.source().sendRecord(Trigger.create(time2, 0), time2);
+ TaskResult result2 = infra.sink().poll(Duration.ofSeconds(10));
+
+ assertThat(result2.exceptions()).isEmpty();
+ assertThat(result2.success()).isTrue();
+
+ table.refresh();
+ // Both deletes converted to DVs on the target, so neither row is visible there.
+ assertRecords(table, ImmutableList.of());
+ } finally {
+ closeJobClient(jobClient);
+ }
+ }
+
private void appendConvertTask() {
appendConvertTask(STAGING_BRANCH);
}
@@ -1365,22 +1496,16 @@ class TestConvertEqualityDeletes extends MaintenanceTaskTestBase {
private static void assertRecords(Table table, List<Record> expected) throws IOException {
table.refresh();
- Types.StructType type = SimpleDataUtil.SCHEMA.asStruct();
-
- StructLikeSet expectedSet = StructLikeSet.create(type);
- expectedSet.addAll(expected);
try (CloseableIterable<Record> iterable =
IcebergGenerics.read(table)
.useSnapshot(table.currentSnapshot().snapshotId())
.project(SimpleDataUtil.SCHEMA)
.build()) {
- StructLikeSet actualSet = StructLikeSet.create(type);
- for (Record record : iterable) {
- actualSet.add(record);
- }
-
- assertThat(actualSet).isEqualTo(expectedSet);
+ // rows from files with deletes applied carry an extra _pos field
+ assertThat(Lists.newArrayList(iterable))
+ .map(r -> createRecord((Integer) r.getField("id"), (String) r.getField("data")))
+ .containsExactlyInAnyOrderElementsOf(expected);
}
}
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
index 87eb47ac2c..ed243695bc 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
@@ -630,7 +630,7 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
}
@Test
- void noMainReEmitWhenUnchanged() throws Exception {
+ void rebuildsIndexWhenReplanningUncommittedStagingSnapshot() throws Exception {
Table table = createTableWithDelete(3);
insert(table, 1, "a");
insert(table, 2, "b");
@@ -638,34 +638,84 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
table.refresh();
- DeleteFile eqDelete1 = writeEqualityDelete(table, 1, "a");
- table.newRowDelta().addDeletes(eqDelete1).toBranch(STAGING_BRANCH).commit();
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
table.refresh();
try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
createHarness(STAGING_BRANCH)) {
harness.open();
+ // First trigger bootstraps the index and resolves the staging snapshot's eq delete, which
+ // consumes the matching index entries.
sendTrigger(harness);
int firstTriggerCount = harness.extractOutputValues().size();
- // 2 DATA_FILE (main) + 1 EQ_DELETE_FILE
assertThat(firstTriggerCount).isEqualTo(3);
- assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1);
-
- DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
- table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
- table.refresh();
+ // No committer marker on main, so the cycle did not commit and the same staging snapshot is
+ // planned again. Main has not moved either, so the planner must rebuild the index itself.
sendTrigger(harness);
List<ReadCommand> allCommands = harness.extractOutputValues();
List<ReadCommand> trigger2Commands =
allCommands.subList(firstTriggerCount, allCommands.size());
- // Only 1 EQ_DELETE_FILE, no main re-emission
- assertThat(countDataFileTasks(trigger2Commands)).isEqualTo(0);
+ assertThat(countDataFileTasks(trigger2Commands)).isEqualTo(2);
assertThat(countEqDeleteTasks(trigger2Commands)).isEqualTo(1);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void rebuildsIndexAfterRestoreOfUncommittedCycle() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ OperatorSubtaskState state;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // First cycle converts S1 and commits, so the newest commit on the target is the converter's
+ // own marker. The existing reindex path counts only commits newer than that marker, so it
+ // cannot fire on the restore below and the replan check is the only thing that can rebuild.
+ sendTrigger(harness);
+ assertThat(harness.extractOutputValues()).hasSize(3);
+ simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
+ // Second cycle resolves S2's eq delete but never commits, then a checkpoint is taken.
+ DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
+ table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
- assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2);
+ int afterFirstCycle = harness.extractOutputValues().size();
+ sendTrigger(harness);
+ assertThat(harness.extractOutputValues().size()).isGreaterThan(afterFirstCycle);
+
+ state = harness.snapshot(1, System.currentTimeMillis());
+ }
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(state);
+ harness.open();
+
+ // The restored planner replans the same staging snapshot, so it must rebuild the index the
+ // uncommitted cycle consumed.
+ sendTrigger(harness);
+ List<ReadCommand> commands = harness.extractOutputValues();
+
+ // Two inserts plus simulateConvertCommit's marker file = 3 data files on the target.
+ assertThat(countDataFileTasks(commands)).isEqualTo(3);
+ assertThat(countEqDeleteTasks(commands)).isEqualTo(1);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
}
}
@@ -789,7 +839,6 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
table.newAppend().appendFile(externalFile).commit();
table.refresh();
long mainAfterExternal = table.snapshot(SnapshotRef.MAIN_BRANCH).snapshotId();
- long mainSeqAfterExternal = table.snapshot(SnapshotRef.MAIN_BRANCH).sequenceNumber();
DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
@@ -804,7 +853,8 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
assertThat(clears).hasSize(1);
assertThat(clears.get(0).type()).isEqualTo(IndexCommand.Type.CLEAR_INDEX);
assertThat(clears.get(0).mainSnapshotId()).isEqualTo(mainAfterExternal);
- assertThat(clears.get(0).mainSequenceNumber()).isEqualTo(mainSeqAfterExternal);
+ // The bootstrap on the first trigger took generation 1, so the reindex takes the next one.
+ assertThat(clears.get(0).indexGeneration()).isEqualTo(2L);
}
}
@@ -829,6 +879,10 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
int afterFirst = harness.extractOutputValues().size();
assertThat(afterFirst).isGreaterThan(0);
+ // The first cycle commits, so later triggers plan new staging snapshots rather than
+ // replanning this one.
+ simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
// External commit on main (no COMMITTED_STAGING_SNAPSHOT_PROPERTY).
DataFile externalFile =
new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
index d098b9e0b4..9c93b8122a 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
@@ -229,7 +229,7 @@ class TestEqualityConvertReader extends OperatorTestBase {
PartitionSpec spec = table.specs().get(dataFile.specId());
List<Integer> fieldIds = Lists.newArrayList(1);
long mainSnapshotId = 42L;
- long mainSequenceNumber = 7L;
+ long indexGeneration = 7L;
long dataSequenceNumber = 9L;
try (OneInputStreamOperatorTestHarness<ReadCommand, IndexCommand> harness =
@@ -240,14 +240,14 @@ class TestEqualityConvertReader extends OperatorTestBase {
ReadCommand.stagingDataFile(
new FlinkAddedRowsScanTask(dataFile, spec),
mainSnapshotId,
- mainSequenceNumber,
+ indexGeneration,
dataSequenceNumber);
harness.processElement(cmd, 0);
List<IndexCommand> output = harness.extractOutputValues();
assertThat(output).hasSize(1);
assertThat(output.get(0).mainSnapshotId()).isEqualTo(mainSnapshotId);
- assertThat(output.get(0).mainSequenceNumber()).isEqualTo(mainSequenceNumber);
+ assertThat(output.get(0).indexGeneration()).isEqualTo(indexGeneration);
assertThat(output.get(0).rowPosition().dataSequenceNumber()).isEqualTo(dataSequenceNumber);
}
}