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/29 06:19:49 UTC
(iceberg) branch main updated: Flink: Fix deleted rows reappearing after a failed eq-delete conversion cycle (#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
The following commit(s) were added to refs/heads/main by this push:
new 0abea5bc1a Flink: Fix deleted rows reappearing after a failed eq-delete conversion cycle (#17630)
0abea5bc1a is described below
commit 0abea5bc1aa5cfa5919805b43574916833b5a371
Author: vishnu prakash <nv...@gmail.com>
AuthorDate: Sat Aug 29 11:49:42 2026 +0530
Flink: Fix deleted rows reappearing after a failed eq-delete conversion cycle (#17630)
---
docs/docs/flink-maintenance.md | 2 +-
.../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 +-
9 files changed, 372 insertions(+), 138 deletions(-)
diff --git a/docs/docs/flink-maintenance.md b/docs/docs/flink-maintenance.md
index 57354ee9ff..11a43943c4 100644
--- a/docs/docs/flink-maintenance.md
+++ b/docs/docs/flink-maintenance.md
@@ -114,7 +114,7 @@ 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 th [...]
-- 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).
+- 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 `equalityFieldColumns` set is persisted in operator state. Reconfiguring it across a restart from savepoint is **not supported** and the job will fail fast on restore. Restart from a clean state (no savepoint) if the equality columns change.
- The committer is intentionally stateless. On restart, the planner rediscovers its position by walking the target branch for the marker property `equality-convert-staging-snapshot` written by the committer. If that marker is no longer reachable on the target branch (target was rolled back, replace-main'ed, or the marker snapshot was expired) the planner fails and requires manual intervention.
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
index 5c949122d9..ac3b51bfdc 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
+++ b/flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
index 8499f689b2..82791bae4b 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
+++ b/flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
index 809e60a8f8..a0c1752fdf 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
+++ b/flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
index 509812cbb3..0781fa2723 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
+++ b/flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
index e30e851600..5be25949cd 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ReadCommand.java
+++ b/flink/v2.1/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.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
index beee65202d..2e03a0655e 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
+++ b/flink/v2.1/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.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
index 87eb47ac2c..ed243695bc 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
+++ b/flink/v2.1/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.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
index d098b9e0b4..9c93b8122a 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
+++ b/flink/v2.1/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);
}
}