Skip to content

Flink: Backport eq-delete conversion replan reindex fix to 1.20, 2.2 and 2.3 #17875

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vishnuprakaz wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Flink: Backport eq-delete conversion replan reindex fix to 1.20, 2.2 and 2.3 #17875

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
50aa601
Flink: Backport eq-delete conversion replan reindex fix to 1.20 and 2…
vishnuprakaz Aug 29, 2026
fe9f980
Flink: Forward-port eq-delete conversion replan reindex fix to 2.3 (#…
vishnuprakaz Aug 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 27 additions & 26 deletions ...k/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,21 +58,22 @@
* 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
Expand All @@ -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 =
Expand All @@ -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.
Expand Down Expand Up @@ -132,7 +133,7 @@ public EqualityConvertPKIndex(boolean stagingOnTargetBranch) {
@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);
Expand All @@ -149,15 +150,15 @@ public void open(OpenContext context) throws Exception {
public void processElement(IndexCommand cmd, ReadOnlyContext ctx, Collector<DVPosition> out)
throws Exception {
try {
Long storedSequence = mainSequenceVersion.value();
if (!Objects.equals(storedSequence, cmd.mainSequenceNumber())) {
Long storedGeneration = indexGeneration.value();
if (!Objects.equals(storedGeneration, cmd.indexGeneration())) {
LOG.info(
"Main sequence changed from {} to {} (snapshot {}), clearing state",
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();
Expand Down Expand Up @@ -204,15 +205,15 @@ public void processBroadcastElement(IndexCommand cmd, Context ctx, Collector<DVP
"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();
}
});
Expand Down
163 changes: 111 additions & 52 deletions ...k/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,12 +65,14 @@
* 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.
Expand Down Expand Up @@ -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;

Expand All @@ -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;

Expand Down Expand Up @@ -189,19 +196,20 @@ public void initializeState(StateInitializationContext context) throws Exception
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()
Expand All @@ -214,6 +222,19 @@ public void initializeState(StateInitializationContext context) throws Exception
+ "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
Expand All @@ -224,15 +245,18 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
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
Expand All @@ -246,11 +270,13 @@ public void processElement(StreamRecord<Trigger> element) throws Exception {
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);
Expand All @@ -267,52 +293,81 @@ public void processElement(StreamRecord<Trigger> element) throws Exception {
}

/**
* 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) {
Expand Down Expand Up @@ -445,6 +500,10 @@ private void processStagingSnapshot(
"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());

Expand Down Expand Up @@ -572,7 +631,7 @@ private void emitDeletePhase(List<DeleteFile> eqDeleteFiles) {
deleteFile,
spec,
indexSnapshotId,
indexedSequenceNumber,
indexGeneration,
dataSequenceNumber(deleteFile)),
nextPhaseTs));
processedEqDeleteFileNumCounter.inc();
Expand All @@ -593,7 +652,7 @@ private void emitSnapshotDataPhase(List<DataFile> snapshotDataFiles) {
ReadCommand.stagingDataFile(
new FlinkAddedRowsScanTask(dataFile, spec),
indexSnapshotId,
indexedSequenceNumber,
indexGeneration,
dataSequenceNumber(dataFile)),
nextPhaseTs));
}
Expand Down Expand Up @@ -632,7 +691,7 @@ private void emitMainDataReadCommands(Snapshot mainSnapshot) {
output.collect(
new StreamRecord<>(
ReadCommand.dataFile(
task, indexSnapshotId, indexedSequenceNumber, dataSequenceNumber(task.file())),
task, indexSnapshotId, indexGeneration, dataSequenceNumber(task.file())),
nextPhaseTs));
}
} catch (IOException e) {
Expand Down
Loading
Loading