(iceberg) branch main updated: Flink: Backport: ConvertEqualityDeletes: Fix index consistency after a failure or restore (#18674) (#18777)

[mxm.apache.org via commits]

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 5f2976d1ed Flink: Backport: ConvertEqualityDeletes: Fix index 
consistency after a failure or restore (#18674) (#18777)
5f2976d1ed is described below

commit 5f2976d1edddff7dfb8c69f9ddf0f06059155054
Author: Maximilian Michels <m...@apache.org>
AuthorDate: Wed Nov 4 16:18:24 2026 +0100

    Flink: Backport: ConvertEqualityDeletes: Fix index consistency after a 
failure or restore (#18674) (#18777)
---
 .../operator/EqualityConvertCommitter.java         |  56 +++-
 .../maintenance/operator/EqualityConvertPlan.java  |  19 +-
 .../operator/EqualityConvertPlanner.java           | 132 +++++---
 .../api/TestConvertEqualityDeletes.java            | 108 +++++++
 .../api/TestConvertEqualityDeletesE2E.java         | 109 ++++++-
 .../operator/TestEqualityConvertCommitter.java     | 198 +++++++++++-
 .../operator/TestEqualityConvertDVWriter.java      |  16 +-
 .../operator/TestEqualityConvertPlanner.java       | 338 ++++++++++++++++++++-
 .../operator/EqualityConvertCommitter.java         |  56 +++-
 .../maintenance/operator/EqualityConvertPlan.java  |  19 +-
 .../operator/EqualityConvertPlanner.java           | 132 +++++---
 .../api/TestConvertEqualityDeletes.java            | 108 +++++++
 .../api/TestConvertEqualityDeletesE2E.java         | 111 ++++++-
 .../operator/TestEqualityConvertCommitter.java     | 198 +++++++++++-
 .../operator/TestEqualityConvertDVWriter.java      |  16 +-
 .../operator/TestEqualityConvertPlanner.java       | 338 ++++++++++++++++++++-
 16 files changed, 1794 insertions(+), 160 deletions(-)

diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
index 34d996c3d9..0e2fe43df4 100644
--- 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
@@ -22,8 +22,13 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
 import org.apache.flink.streaming.api.watermark.Watermark;
@@ -36,6 +41,7 @@ import 
org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.maintenance.api.Trigger;
 import 
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.ContentFileUtil;
@@ -66,7 +72,13 @@ import org.slf4j.LoggerFactory;
  * completes cleanly. Errors are reported via {@link 
TaskResultAggregator#ERROR_STREAM} side output;
  * the Aggregator collects them and surfaces failure on its own watermark.
  *
- * <p>The committer is intentionally stateless: {@code bufferedResults} and 
{@code planResult} are
+ * <p>The committer keeps the index generation of the last failed cycle as 
checkpointed state. A
+ * failure in the readers or workers while the index is updated in a cycle 
with nothing to convert
+ * may have left the index without some rows, but that cycle has no commit to 
withhold. The
+ * committer withholds the next conversion commit on that generation instead: 
the next trigger then
+ * plans that staging snapshot again, which makes the planner rebuild the 
index on a new generation.
+ * The failure was reported where it happened, so the withheld cycle completes 
without an error.
+ * Apart from that the committer is stateless: {@code bufferedResults} and 
{@code planResult} are
  * not checkpointed. The maintenance framework ensures mutual exclusive tasks, 
and on restart the
  * planner re-derives its position from {@link 
#COMMITTED_STAGING_SNAPSHOT_PROPERTY} on main, so the
  * committer never receives a plan for an already-committed staging snapshot.
@@ -93,6 +105,8 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
   private transient Table table;
   private transient List<DVWriteResult> bufferedResults;
   private transient EqualityConvertPlan planResult;
+  private transient ListState<Long> failedIndexGenerationState;
+  private transient long failedIndexGeneration;
 
   private transient Counter errorCounter;
   private transient Counter addedDataFileNumCounter;
@@ -136,6 +150,32 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
     this.commitDurationMsCounter = 
taskMetricGroup.counter(COMMIT_DURATION_MS_METRIC);
   }
 
+  @Override
+  public void initializeState(StateInitializationContext context) throws 
Exception {
+    super.initializeState(context);
+    failedIndexGenerationState =
+        context
+            .getOperatorStateStore()
+            .getListState(new ListStateDescriptor<>("failedIndexGeneration", 
Types.LONG));
+
+    failedIndexGeneration = EqualityConvertPlan.NO_INDEX_GENERATION;
+    for (Long stateValue : failedIndexGenerationState.get()) {
+      Preconditions.checkState(
+          failedIndexGeneration == EqualityConvertPlan.NO_INDEX_GENERATION,
+          "failedIndexGeneration state should hold at most one value");
+      failedIndexGeneration = stateValue;
+    }
+  }
+
+  @Override
+  public void snapshotState(StateSnapshotContext context) throws Exception {
+    super.snapshotState(context);
+    failedIndexGenerationState.clear();
+    if (failedIndexGeneration != EqualityConvertPlan.NO_INDEX_GENERATION) {
+      failedIndexGenerationState.add(failedIndexGeneration);
+    }
+  }
+
   @Override
   public void processElement1(StreamRecord<DVWriteResult> record) {
     bufferedResults.add(record.getValue());
@@ -196,6 +236,7 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
             "Skipping commit for table {} task {}: a DV writer reported an 
error.",
             tableName,
             taskName);
+        failedIndexGeneration = planResult.indexGeneration();
         deleteUncommittedDVs();
         return;
       }
@@ -209,6 +250,19 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
       return;
     }
 
+    if (planResult.indexGeneration() == failedIndexGeneration) {
+      LOG.warn(
+          "Not committing staging snapshot {} for table {} task {}: an index 
update failed on "
+              + "index generation {}, so the next trigger rebuilds the index 
and converts the "
+              + "snapshot again.",
+          planResult.stagingSnapshotId(),
+          tableName,
+          taskName,
+          failedIndexGeneration);
+      deleteUncommittedDVs();
+      return;
+    }
+
     table.refresh();
 
     List<DeleteFile> allDvFiles = Lists.newArrayList();
diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
index 4aa326d19a..3226fbf4a2 100644
--- 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
@@ -31,8 +31,8 @@ import 
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
  * EqualityConvertCommitter} (for data files and staging deletes to commit).
  *
  * <p>A no-op cycle is encoded by {@link #stagingSnapshotId()} == {@link
- * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to 
construct one and {@link
- * #noOp()} to check.
+ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long, long)} to 
construct one and
+ * {@link #noOp()} to check.
  *
  * <p>A cycle-start marker is a no-op plan whose done timestamp is never 
reached. Use {@link
  * #cycleStart(long)} to construct one.
@@ -50,6 +50,9 @@ import 
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
  * @param triggerTimestamp original trigger timestamp, forwarded by the 
Committer to the Aggregator
  * @param doneTimestamp timestamp after which all phase watermarks have been 
emitted; the DVWriter
  *     should only process the result when the watermark reaches or exceeds 
this value
+ * @param indexGeneration generation of the worker index the cycle ran on. The 
committer does not
+ *     commit a cycle on a generation whose update failed. {@link 
#NO_INDEX_GENERATION} for the
+ *     cycle-start marker.
  */
 @Internal
 public record EqualityConvertPlan(
@@ -59,11 +62,14 @@ public record EqualityConvertPlan(
     long stagingSnapshotId,
     Long mainSnapshotId,
     long triggerTimestamp,
-    long doneTimestamp)
+    long doneTimestamp,
+    long indexGeneration)
     implements Serializable {
 
   public static final long NO_OP_STAGING_SNAPSHOT_ID = -1L;
 
+  public static final long NO_INDEX_GENERATION = -1L;
+
   /** Returns {@code true} if this cycle has nothing to commit. */
   public boolean noOp() {
     return stagingSnapshotId == NO_OP_STAGING_SNAPSHOT_ID;
@@ -71,7 +77,7 @@ public record EqualityConvertPlan(
 
   /** No-op cycle: empty file lists, sentinel staging snapshot id. */
   public static EqualityConvertPlan noOp(
-      Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) {
+      Long mainSnapshotId, long triggerTimestamp, long doneTimestamp, long 
indexGeneration) {
     return new EqualityConvertPlan(
         ImmutableList.of(),
         ImmutableList.of(),
@@ -79,7 +85,8 @@ public record EqualityConvertPlan(
         NO_OP_STAGING_SNAPSHOT_ID,
         mainSnapshotId,
         triggerTimestamp,
-        doneTimestamp);
+        doneTimestamp,
+        indexGeneration);
   }
 
   /**
@@ -87,6 +94,6 @@ public record EqualityConvertPlan(
    * until the cycle's actual plan replaces it.
    */
   public static EqualityConvertPlan cycleStart(long triggerTimestamp) {
-    return noOp(null, triggerTimestamp, Long.MAX_VALUE);
+    return noOp(null, triggerTimestamp, Long.MAX_VALUE, NO_INDEX_GENERATION);
   }
 }
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 96e5861f1c..1bf08b0976 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
@@ -70,9 +70,8 @@ import org.slf4j.LoggerFactory;
  * <ol>
  *   <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 #ensureIndexCurrent}: rebuilds the worker index from main 
unless it reflects the
+ *       target branch head, see {@link #indexReflects}.
  *   <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.
@@ -109,6 +108,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   private static final String SKIPPED_NO_OP_CYCLES_METRIC = 
"skippedNoOpCycles";
   private static final String REINDEX_COUNT_METRIC = "reindexCount";
 
+  private static final int STATE_VERSION = 1;
+
   private final String tableName;
   private final String taskName;
   private final TableLoader tableLoader;
@@ -123,10 +124,12 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   private transient ListState<Long> indexSnapshotState;
   // Generation of the worker's current index build.
   private transient ListState<Long> indexGenerationState;
-  // Staging snapshot the last emitted plan covered.
+  // Staging snapshot the index changed for until its conversion commit shows 
up on the target.
   private transient ListState<Long> pendingStagingSnapshotState;
   // Equality field IDs the index was built with, allows to detect 
reconfiguration.
   private transient ListState<Integer> eqFieldIdsState;
+  // Absent in state of a version that recorded the index as current before it 
was complete.
+  private transient ListState<Integer> stateVersionState;
 
   private transient Table table;
 
@@ -134,8 +137,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   private transient Long lastStagingSnapshotId;
   private transient Long indexSnapshotId;
   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.
+  // Staging snapshot the index changed for until its conversion commit shows 
up on the target
+  // branch, checkpointed so it survives a restore taken mid-cycle.
   private transient Long pendingStagingSnapshotId;
 
   private transient long nextPhaseTs;
@@ -238,6 +241,20 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
           "pendingStagingSnapshotId state should hold at most one value");
       pendingStagingSnapshotId = stateValue;
     }
+
+    stateVersionState =
+        context
+            .getOperatorStateStore()
+            .getListState(new ListStateDescriptor<>("stateVersion", 
Types.INT));
+
+    // The index recorded by an older version can be incomplete while it looks 
current.
+    if (context.isRestored() && !stateVersionState.get().iterator().hasNext()) 
{
+      LOG.info(
+          "Restored state of an older version for table {} task {}, rebuilding 
the worker index.",
+          tableName,
+          taskName);
+      indexSnapshotId = null;
+    }
   }
 
   @Override
@@ -260,6 +277,9 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
     if (pendingStagingSnapshotId != null) {
       pendingStagingSnapshotState.add(pendingStagingSnapshotId);
     }
+
+    stateVersionState.clear();
+    stateVersionState.add(STATE_VERSION);
   }
 
   @Override
@@ -279,12 +299,12 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
       Snapshot mainSnapshot = table.snapshot(targetBranch);
       currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() 
: null;
 
-      LastCommittedWork committedWork = refreshStagingCursor(mainSnapshot);
+      refreshStagingCursor(mainSnapshot);
 
       Snapshot nextToProcess =
           nextUnprocessedStagingSnapshot(table.snapshot(stagingBranch), 
mainSnapshot);
 
-      ensureIndexCurrent(mainSnapshot, committedWork, nextToProcess);
+      ensureIndexCurrent(mainSnapshot);
 
       if (nextToProcess == null) {
         LOG.info("Nothing new to convert on staging branch '{}'.", 
stagingBranch);
@@ -303,45 +323,35 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
 
   /**
    * 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.
+   * branch. Does nothing when the target has not moved since the last trigger 
and the cursor
+   * therefore cannot have changed.
    */
-  private LastCommittedWork refreshStagingCursor(Snapshot mainSnapshot) {
+  private void refreshStagingCursor(Snapshot mainSnapshot) {
     Long currentMainSnapshotId = mainSnapshot != null ? 
mainSnapshot.snapshotId() : null;
 
     if (Objects.equals(lastMainSnapshotId, currentMainSnapshotId)) {
-      return null;
+      return;
     }
 
-    LastCommittedWork info = discoverLastCommittedWork(mainSnapshot);
-    updateLastStagingSnapshotId(info);
-    return info;
+    updateLastStagingSnapshotId(discoverLastCommittedWork(mainSnapshot));
   }
 
   /**
-   * 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.
+   * Rebuilds the worker index unless it reflects {@code mainSnapshot}, see 
{@link #indexReflects}.
    */
-  private void ensureIndexCurrent(
-      Snapshot mainSnapshot, LastCommittedWork committedWork, Snapshot 
nextToProcess) {
+  private void ensureIndexCurrent(Snapshot mainSnapshot) {
     if (mainSnapshot == null) {
+      // The target branch has no rows, so the index must not have any either. 
A new generation
+      // makes the worker drop the rows indexed so far, e.g. by a cycle that 
did not commit.
+      indexGeneration++;
+      indexSnapshotId = null;
+      pendingStagingSnapshotId = 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) {
+    if (!indexReflects(mainSnapshot)) {
+      boolean bootstrap = indexSnapshotId == null;
       LOG.info(
           "{} worker index from main snapshot {} for field IDs {}.",
           bootstrap ? "Bootstrapping" : "Reindexing",
@@ -350,20 +360,58 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
       rebuildIndex(mainSnapshot, !bootstrap);
     }
 
+    indexSnapshotId = mainSnapshot.snapshotId();
+    pendingStagingSnapshotId = null;
     lastMainSnapshotId = mainSnapshot.snapshotId();
   }
 
+  /**
+   * Returns whether the index reflects {@code mainSnapshot}.
+   *
+   * <p>Planning a staging snapshot changes the index before its conversion 
commit exists: its eq
+   * deletes consume the entries they match and its data files are added. So 
with a snapshot
+   * planned, the index reflects the head only if that snapshot's conversion 
commit sits directly on
+   * top of the indexed snapshot. Without one, the head must be the indexed 
snapshot. Any other head
+   * holds commits the index never saw (external commits, rollbacks, a restore 
from before a
+   * conversion commit) or follows a cycle that did not commit.
+   *
+   * <p>After a restore, the conversion commit can come from a retry of the 
planned cycle. A cycle
+   * that only failed to commit leaves the same index as its retry. A rebuild 
that fails part way
+   * clears {@link #indexSnapshotId}, and the committer does not commit a 
conversion on a generation
+   * whose update failed downstream.
+   */
+  private boolean indexReflects(Snapshot mainSnapshot) {
+    if (indexSnapshotId == null) {
+      return false;
+    }
+
+    if (pendingStagingSnapshotId == null) {
+      return mainSnapshot.snapshotId() == indexSnapshotId;
+    }
+
+    String committedStaging =
+        
mainSnapshot.summary().get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+    return committedStaging != null
+        && Long.parseLong(committedStaging) == pendingStagingSnapshotId
+        && Objects.equals(mainSnapshot.parentId(), indexSnapshotId);
+  }
+
   /**
    * 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.
+   * complete earlier index. Entries of an incomplete one are ignored at the 
new generation.
    *
    * <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.
+   *
+   * <p>Clears {@link #indexSnapshotId} before emitting anything, and the 
caller records the head
+   * only after this returns. Neither the next trigger nor a restore from a 
checkpoint taken after a
+   * rebuild that failed part way trusts the index it left.
    */
   private void rebuildIndex(Snapshot mainSnapshot, boolean evictStaleKeys) {
+    indexSnapshotId = null;
     long generation = indexGeneration + 1;
 
     if (evictStaleKeys) {
@@ -374,7 +422,6 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
       reindexCounter.inc();
     }
 
-    indexSnapshotId = mainSnapshot.snapshotId();
     indexGeneration = generation;
     emitMainDataReadCommands(mainSnapshot);
   }
@@ -399,7 +446,7 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   /**
    * Walks main back from head looking for the most recent snapshot tagged 
with {@link
    * EqualityConvertCommitter#COMMITTED_STAGING_SNAPSHOT_PROPERTY}. Returns 
the staging snapshot id
-   * recorded there (or {@code null} if not reached) and the count of 
intervening external commits.
+   * recorded there, or {@code null} if not reached.
    *
    * <p>The walk stops at whichever comes first:
    *
@@ -410,7 +457,6 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
    */
   private LastCommittedWork discoverLastCommittedWork(Snapshot mainSnapshot) {
     Long lastCommittedStaging = null;
-    int externalCount = 0;
     boolean reachedLastInspected = false;
     Snapshot current = mainSnapshot;
     while (current != null) {
@@ -426,15 +472,13 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
         break;
       }
 
-      externalCount++;
       current = parentOf(current);
     }
 
-    return new LastCommittedWork(lastCommittedStaging, externalCount, 
reachedLastInspected);
+    return new LastCommittedWork(lastCommittedStaging, reachedLastInspected);
   }
 
-  private record LastCommittedWork(
-      Long lastCommittedStaging, int externalCommitCount, boolean 
reachedLastInspected) {}
+  private record LastCommittedWork(Long lastCommittedStaging, boolean 
reachedLastInspected) {}
 
   /**
    * Walks staging history from head back to the stop point and returns the 
oldest unprocessed
@@ -533,7 +577,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
                 stagingSnapshot.snapshotId(),
                 currentMainSnapshotId,
                 triggerTs,
-                nextPhaseTs)));
+                nextPhaseTs,
+                indexGeneration)));
 
     advancePhase();
   }
@@ -680,7 +725,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
     output.collect(
         METADATA_STREAM,
         new StreamRecord<>(
-            EqualityConvertPlan.noOp(currentMainSnapshotId, triggerTimestamp, 
nextPhaseTs)));
+            EqualityConvertPlan.noOp(
+                currentMainSnapshotId, triggerTimestamp, nextPhaseTs, 
indexGeneration)));
     advancePhase();
   }
 
@@ -700,7 +746,7 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
         output.collect(
             new StreamRecord<>(
                 ReadCommand.dataFile(
-                    task, indexSnapshotId, indexGeneration, 
dataSequenceNumber(task.file())),
+                    task, commitSnapshotId, indexGeneration, 
dataSequenceNumber(task.file())),
                 nextPhaseTs));
       }
     } catch (IOException e) {
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 2e03a0655e..486ab0fe42 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
@@ -1473,6 +1473,114 @@ class TestConvertEqualityDeletes extends 
MaintenanceTaskTestBase {
     }
   }
 
+  @Test
+  void testIndexUpdateFailureOnIdleTriggerIsRepaired() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile firstDelete = writeEqualityDelete(table, 1, "a");
+    
table.newRowDelta().addDeletes(firstDelete).toBranch(STAGING_BRANCH).commit();
+
+    appendConvertTask(STAGING_BRANCH);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = infra.env().executeAsync();
+
+      long time1 = System.currentTimeMillis();
+      infra.source().sendRecord(Trigger.create(time1, 0), time1);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+      // An external commit lands while staging is idle, and its data file is 
unreadable while the
+      // next trigger updates the index.
+      DataFile externalData = writeDataFile(table, createRecord(2, "b"));
+      table.newAppend().appendFile(externalData).commit();
+      File externalFile = new File(externalData.location().replace("file:", 
""));
+      File hiddenFile = new File(externalFile.getPath() + ".hidden");
+      assertThat(externalFile.renameTo(hiddenFile)).isTrue();
+
+      long time2 = time1 + 1;
+      infra.source().sendRecord(Trigger.create(time2, 0), time2);
+      
assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isFalse();
+
+      assertThat(hiddenFile.renameTo(externalFile)).isTrue();
+
+      DeleteFile secondDelete = writeEqualityDelete(table, 2, "b");
+      
table.newRowDelta().addDeletes(secondDelete).toBranch(STAGING_BRANCH).commit();
+
+      // The index may miss the external row, so the next conversion is not 
committed.
+      table.refresh();
+      long mainSnapshotId = table.currentSnapshot().snapshotId();
+      long time3 = time2 + 1;
+      infra.source().sendRecord(Trigger.create(time3, 0), time3);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+      table.refresh();
+      
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+
+      // Planned again, the conversion rebuilds the index and deletes the 
external row.
+      long time4 = time3 + 1;
+      infra.source().sendRecord(Trigger.create(time4, 0), time4);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+      assertRecords(table, ImmutableList.of());
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  @Test
+  void testFailedConversionOnEmptyTargetIsRetried() throws Exception {
+    Table table = createTableWithDelete(3);
+
+    // The target branch has no snapshot until the first conversion commit.
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    // An upsert of (1, "a") and an insert whose data file is unreadable 
during the first cycle.
+    DataFile reinsert = writeDataFile(table, createRecord(1, "a"));
+    DataFile insert = writeDataFile(table, createRecord(2, "b"));
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table
+        .newRowDelta()
+        .addRows(reinsert)
+        .addRows(insert)
+        .addDeletes(eqDelete)
+        .toBranch(STAGING_BRANCH)
+        .commit();
+
+    File insertFile = new File(insert.location().replace("file:", ""));
+    File hiddenFile = new File(insertFile.getPath() + ".hidden");
+    assertThat(insertFile.renameTo(hiddenFile)).isTrue();
+
+    appendConvertTask(STAGING_BRANCH);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = infra.env().executeAsync();
+
+      // The first cycle indexes the re-insert, fails on the unreadable file 
and does not commit.
+      long time1 = System.currentTimeMillis();
+      infra.source().sendRecord(Trigger.create(time1, 0), time1);
+      
assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isFalse();
+      table.refresh();
+      assertThat(table.snapshot(SnapshotRef.MAIN_BRANCH)).isNull();
+
+      assertThat(hiddenFile.renameTo(insertFile)).isTrue();
+
+      // The retry must not delete the re-insert indexed by the first cycle.
+      long time2 = time1 + 1;
+      infra.source().sendRecord(Trigger.create(time2, 0), time2);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+      assertRecords(table, ImmutableList.of(createRecord(1, "a"), 
createRecord(2, "b")));
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
   private void appendConvertTask() {
     appendConvertTask(STAGING_BRANCH);
   }
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
index e39d05b66f..edc0f0da43 100644
--- 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.flink.maintenance.api;
 
 import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
 import static 
org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME;
+import static 
org.apache.iceberg.flink.maintenance.operator.EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY;
 import static 
org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -30,12 +31,18 @@ import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.test.junit5.InjectClusterClient;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileContent;
@@ -104,18 +111,7 @@ class TestConvertEqualityDeletesE2E extends 
OperatorTestBase {
     table.refresh();
     assertThat(dvCountOnMain(table)).isZero();
 
-    TableMaintenance.forTable(env, tableLoader(), LOCK_FACTORY)
-        .uidSuffix("ConvertEqualityDeletesE2EUID-" + stagingBranch)
-        .rateLimit(Duration.ofMillis(50))
-        .lockCheckDelay(Duration.ofMillis(50))
-        .add(
-            ConvertEqualityDeletes.builder()
-                .scheduleOnInterval(Duration.ofMillis(100))
-                .stagingBranch(stagingBranch)
-                .targetBranch(SnapshotRef.MAIN_BRANCH)
-                .equalityFieldColumns(ImmutableList.of("id", "data"))
-                .parallelism(2))
-        .append();
+    appendConvertTask(env, stagingBranch, "ConvertEqualityDeletesE2EUID-" + 
stagingBranch);
 
     JobClient jobClient = null;
     try {
@@ -211,6 +207,95 @@ class TestConvertEqualityDeletesE2E extends 
OperatorTestBase {
     }
   }
 
+  /**
+   * Restoring from a savepoint taken before a committed cycle leaves the 
converter's index without
+   * the rows that cycle copied to the target. The next conversion must 
rebuild it, or a delete for
+   * one of those rows resolves to nothing.
+   */
+  @Test
+  void reindexesAfterRestoreFromBeforeCommittedCycle(
+      @TempDir File savepointDir, @InjectClusterClient ClusterClient<?> 
clusterClient)
+      throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table
+        .newRowDelta()
+        .addDeletes(writeEqualityDelete(table, 1, "a"))
+        .toBranch(STAGING_BRANCH)
+        .commit();
+    table.refresh();
+
+    appendConvertTask(env, STAGING_BRANCH, "ConvertEqualityDeletesRestoreUID");
+    String savepointPath;
+    JobClient jobClient = null;
+    try {
+      jobClient = env.executeAsync();
+      awaitConverted(table, table.snapshot(STAGING_BRANCH).snapshotId());
+      savepointPath =
+          jobClient.triggerSavepoint(savepointDir.getPath(), 
SavepointFormatType.CANONICAL).get();
+
+      // This cycle copies S2's row to the target after the savepoint.
+      table.newAppend().appendFile(writeDataFile(table, 2, 
"b")).toBranch(STAGING_BRANCH).commit();
+      table.refresh();
+      awaitConverted(table, table.snapshot(STAGING_BRANCH).snapshotId());
+    } finally {
+      closeJobClient(jobClient);
+    }
+
+    table
+        .newRowDelta()
+        .addDeletes(writeEqualityDelete(table, 2, "b"))
+        .toBranch(STAGING_BRANCH)
+        .commit();
+    table.refresh();
+    long deleteSnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+    StreamExecutionEnvironment restoredEnv = 
StreamExecutionEnvironment.getExecutionEnvironment();
+    appendConvertTask(restoredEnv, STAGING_BRANCH, 
"ConvertEqualityDeletesRestoreUID");
+    StreamGraph streamGraph = restoredEnv.getStreamGraph();
+    
streamGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath));
+    JobID restoredJobId = 
clusterClient.submitJob(streamGraph.getJobGraph()).get();
+    try {
+      awaitConverted(table, deleteSnapshotId);
+      SimpleDataUtil.assertTableRecords(table, ImmutableList.of());
+    } finally {
+      clusterClient.cancel(restoredJobId).get();
+      Awaitility.await()
+          .until(() -> 
clusterClient.getJobStatus(restoredJobId).get().isTerminalState());
+    }
+  }
+
+  private void appendConvertTask(
+      StreamExecutionEnvironment environment, String stagingBranch, String 
uidSuffix)
+      throws IOException {
+    TableMaintenance.forTable(environment, tableLoader(), LOCK_FACTORY)
+        .uidSuffix(uidSuffix)
+        .rateLimit(Duration.ofMillis(50))
+        .lockCheckDelay(Duration.ofMillis(50))
+        .add(
+            ConvertEqualityDeletes.builder()
+                .scheduleOnInterval(Duration.ofMillis(100))
+                .stagingBranch(stagingBranch)
+                .targetBranch(SnapshotRef.MAIN_BRANCH)
+                .equalityFieldColumns(ImmutableList.of("id", "data"))
+                .parallelism(2))
+        .append();
+  }
+
+  private static void awaitConverted(Table table, long stagingSnapshotId) {
+    Awaitility.await()
+        .atMost(Duration.ofMinutes(5))
+        .pollInterval(Duration.ofMillis(200))
+        .untilAsserted(
+            () -> {
+              table.refresh();
+              assertThat(table.currentSnapshot().summary())
+                  .containsEntry(
+                      COMMITTED_STAGING_SNAPSHOT_PROPERTY, 
String.valueOf(stagingSnapshotId));
+            });
+  }
+
   private static void awaitSucceeded(List<String> counterKey, long expected) {
     Awaitility.await()
         .atMost(Duration.ofMinutes(3))
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
index 35aa9908bc..e52f53f942 100644
--- 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
@@ -25,6 +25,7 @@ import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.stream.Collectors;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
@@ -54,6 +55,8 @@ import org.junit.jupiter.api.io.TempDir;
 
 class TestEqualityConvertCommitter extends OperatorTestBase {
 
+  private static final long INDEX_GENERATION = 1L;
+
   @TempDir private Path tempDir;
 
   @Test
@@ -78,7 +81,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
               stagingSnapshotId,
               snapshotIdBefore,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
       harness.processBothWatermarks(new Watermark(doneTs));
@@ -116,7 +120,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               doneTs - 2,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 2));
 
@@ -165,7 +170,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               triggerTs,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
       harness.processElement2(new StreamRecord<>(planResult));
       harness.processBothWatermarks(new Watermark(doneTs));
 
@@ -203,7 +209,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
       harness.open();
 
       long doneTs = System.currentTimeMillis();
-      EqualityConvertPlan planResult = EqualityConvertPlan.noOp(null, doneTs - 
1, doneTs);
+      EqualityConvertPlan planResult =
+          EqualityConvertPlan.noOp(null, doneTs - 1, doneTs, INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
       harness.processBothWatermarks(new Watermark(doneTs));
@@ -235,7 +242,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       // Receive an abort signal from the DVWriter followed by the planning 
planResult.
       harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, doneTs));
@@ -250,6 +258,129 @@ class TestEqualityConvertCommitter extends 
OperatorTestBase {
     }
   }
 
+  @Test
+  void withholdsNextConversionCommitAfterFailedIdleCycle() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+    DeleteFile writtenDv = writeDV(table, stagingDataFile.location(), 0L);
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      failIdleRebuild(harness, mainSnapshotId);
+
+      // The next conversion is not committed, and the DV it wrote is deleted.
+      harness.processElement1(
+          new StreamRecord<>(
+              new DVWriteResult(Lists.newArrayList(writtenDv), 
Lists.newArrayList()), 20L));
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
20L, 1L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      table.refresh();
+      
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+      
assertThat(table.io().newInputFile(writtenDv.location()).exists()).isFalse();
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+
+      // Planned again with a rebuilt index on a new generation, the 
conversion commits.
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
30L, 2L), 29L));
+      harness.processBothWatermarks(new Watermark(30L));
+
+      assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+    }
+  }
+
+  @Test
+  void withholdsNextConversionCommitAfterRestore() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+    OperatorSubtaskState state;
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      failIdleRebuild(harness, mainSnapshotId);
+
+      state = harness.snapshot(1, 11L);
+    }
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.initializeState(state);
+      harness.open();
+
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
20L, 1L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      table.refresh();
+      
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+    }
+  }
+
+  @Test
+  void commitsReplannedConversionAfterFailedConversion() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      // A failed conversion is planned again with a rebuilt index, so nothing 
is withheld.
+      harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, 10L));
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
10L, 1L), 9L));
+      harness.processBothWatermarks(new Watermark(10L));
+
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
20L, 2L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+    }
+  }
+
+  @Test
+  void commitsNextConversionAfterCleanRebuild() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      failIdleRebuild(harness, mainSnapshotId);
+
+      // An external commit makes the next idle cycle rebuild the index, which 
repairs the failure.
+      harness.processElement2(
+          new StreamRecord<>(EqualityConvertPlan.noOp(mainSnapshotId, 19L, 
20L, 2L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
30L, 2L), 29L));
+      harness.processBothWatermarks(new Watermark(30L));
+
+      assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+    }
+  }
+
   @Test
   void failsCommitWhenExternalCommitLandsAfterPlan() throws Exception {
     Table table = createTable(3, FileFormat.PARQUET);
@@ -272,7 +403,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               stagingSnapshotId,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
 
@@ -330,7 +462,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               stagingSnapshotId,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       // Cycle 1 commits, advancing the target past mainSnapshotIdAtPlan and 
writing the marker.
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
@@ -353,7 +486,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               stagingSnapshotId,
               mainSnapshotIdAtPlan,
               doneTs2 - 1,
-              doneTs2);
+              doneTs2,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(replay, doneTs2 - 1));
       harness.processBothWatermarks(new Watermark(doneTs2));
@@ -399,7 +533,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               777L,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -450,7 +585,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       // One writer task wrote a DV (rewriting an existing one); a sibling 
task aborted.
       harness.processElement1(
@@ -505,7 +641,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               888L,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -557,7 +694,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               123L,
               mainSnapshotId,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -616,7 +754,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               123L,
               mainSnapshotId,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -655,6 +794,39 @@ class TestEqualityConvertCommitter extends 
OperatorTestBase {
             SnapshotRef.MAIN_BRANCH));
   }
 
+  // The planner rebuilt the index to generation 1 in a cycle with nothing to 
convert, and a worker
+  // failed to update it.
+  private static void failIdleRebuild(
+      TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness,
+      long mainSnapshotId)
+      throws Exception {
+    harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, 10L));
+    harness.processElement2(
+        new StreamRecord<>(EqualityConvertPlan.noOp(mainSnapshotId, 9L, 10L, 
1L), 9L));
+    harness.processBothWatermarks(new Watermark(10L));
+  }
+
+  private static String committedStagingSnapshot(Table table) {
+    table.refresh();
+    return table
+        .currentSnapshot()
+        .summary()
+        .get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+  }
+
+  private static EqualityConvertPlan conversionPlan(
+      DataFile stagingDataFile, long mainSnapshotId, long doneTs, long 
indexGeneration) {
+    return new EqualityConvertPlan(
+        Lists.newArrayList(stagingDataFile),
+        Lists.newArrayList(),
+        Lists.newArrayList(),
+        42L,
+        mainSnapshotId,
+        doneTs - 1,
+        doneTs,
+        indexGeneration);
+  }
+
   private static List<Watermark> 
watermarks(TwoInputStreamOperatorTestHarness<?, ?, ?> harness) {
     return harness.getOutput().stream()
         .filter(Watermark.class::isInstance)
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
index 93614fe8ab..b7dd1e308d 100644
--- 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
@@ -40,6 +40,7 @@ import org.junit.jupiter.api.Test;
 class TestEqualityConvertDVWriter extends OperatorTestBase {
 
   private static final byte[] EMPTY_PARTITION = new byte[0];
+  private static final long INDEX_GENERATION = 1L;
 
   @Test
   void writesDVFileForSinglePosition() throws Exception {
@@ -247,7 +248,8 @@ class TestEqualityConvertDVWriter extends OperatorTestBase {
               1L,
               null,
               0L,
-              0L);
+              0L,
+              INDEX_GENERATION);
 
       // New position 1 in the same file. Must merge it with the staged 
position 0.
       harness.processElement1(
@@ -288,7 +290,8 @@ class TestEqualityConvertDVWriter extends OperatorTestBase {
               1L,
               plannedSnapshotId,
               0L,
-              0L);
+              0L,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, 
EMPTY_PARTITION, 0L), time));
@@ -491,7 +494,14 @@ class TestEqualityConvertDVWriter extends OperatorTestBase 
{
 
   private static EqualityConvertPlan emptyEqualityConvertPlan() {
     return new EqualityConvertPlan(
-        Lists.newArrayList(), Lists.newArrayList(), Lists.newArrayList(), 1L, 
null, 0L, 0L);
+        Lists.newArrayList(),
+        Lists.newArrayList(),
+        Lists.newArrayList(),
+        1L,
+        null,
+        0L,
+        0L,
+        INDEX_GENERATION);
   }
 
   private static String getFirstDataFilePath(Table table) {
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 1482ce8726..c98de623e7 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
@@ -26,9 +26,13 @@ import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.OperatorStreamStateHandle;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
@@ -45,6 +49,7 @@ import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.RewriteFiles;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SnapshotSummary;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.data.FileHelpers;
 import org.apache.iceberg.data.GenericAppenderHelper;
@@ -53,9 +58,12 @@ import org.apache.iceberg.data.Record;
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.flink.maintenance.api.Trigger;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 class TestEqualityConvertPlanner extends OperatorTestBase {
 
@@ -353,6 +361,13 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
       sendTrigger(harness);
       assertThat(harness.extractOutputValues()).hasSize(firstTriggerCount);
       assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1);
+
+      // Each trigger emits a cycle start marker followed by the plan. The 
bootstrap built
+      // generation 1, and the second trigger reused it.
+      List<StreamRecord<EqualityConvertPlan>> metadata =
+          
Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM));
+      assertThat(metadata.get(1).getValue().indexGeneration()).isEqualTo(1L);
+      assertThat(metadata.get(3).getValue().indexGeneration()).isEqualTo(1L);
     }
   }
 
@@ -668,6 +683,38 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
     }
   }
 
+  @Test
+  void startsNewIndexGenerationWhenReplanningOnEmptyTarget() throws Exception {
+    Table table = createTableWithDelete(3);
+
+    // The target branch has no snapshot until the first conversion commit.
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DataFile reinsert = writeDataFile(table, createRecord(1, "a"));
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    
table.newRowDelta().addRows(reinsert).addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The first cycle indexes the re-insert but does not commit, so the 
same staging snapshot is
+      // planned again. Its eq delete must not resolve against the re-insert 
of the first cycle.
+      sendTrigger(harness);
+      sendTrigger(harness);
+
+      List<ReadCommand> eqDeletes =
+          harness.extractOutputValues().stream()
+              .filter(cmd -> cmd.task() instanceof EqualityDeleteFileScanTask)
+              .collect(Collectors.toList());
+      assertThat(eqDeletes).hasSize(2);
+      assertThat(eqDeletes.get(1).indexGeneration())
+          .isGreaterThan(eqDeletes.get(0).indexGeneration());
+    }
+  }
+
   @Test
   void rebuildsIndexAfterRestoreOfUncommittedCycle() throws Exception {
     Table table = createTableWithDelete(3);
@@ -687,8 +734,7 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
       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.
+      // own marker, which the planner sees on the next trigger.
       sendTrigger(harness);
       assertThat(harness.extractOutputValues()).hasSize(3);
       simulateConvertCommit(table, 
table.snapshot(STAGING_BRANCH).snapshotId());
@@ -722,6 +768,206 @@ class TestEqualityConvertPlanner extends OperatorTestBase 
{
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  void rebuildsIndexAfterRestoreFromBeforeCommittedCycle(boolean 
plannerSawConversionCommit)
+      throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DataFile stagingData;
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The checkpoint follows the first cycle, which converts S1 and 
commits. The planner sees
+      // that commit only if another trigger follows.
+      sendTrigger(harness);
+      simulateConvertCommit(table, 
table.snapshot(STAGING_BRANCH).snapshotId());
+      if (plannerSawConversionCommit) {
+        sendTrigger(harness);
+      }
+
+      state = harness.snapshot(1, System.currentTimeMillis());
+
+      // The second cycle indexes S2's data file as staging data and copies it 
to the target. No
+      // checkpoint follows it.
+      stagingData = writeDataFile(table, createRecord(2, "b"));
+      
table.newAppend().appendFile(stagingData).toBranch(STAGING_BRANCH).commit();
+      table.refresh();
+      long s2SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+      sendTrigger(harness);
+      table
+          .newRowDelta()
+          .addRows(stagingData)
+          .set(
+              EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY,
+              String.valueOf(s2SnapshotId))
+          .commit();
+      table.refresh();
+    }
+
+    DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
+    
table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The index restored with the planner never held S2's row, so the 
planner rebuilds it from
+      // the target before resolving the delete for that row.
+      sendTrigger(harness);
+      List<ReadCommand> commands = harness.extractOutputValues();
+
+      assertThat(commands).anyMatch(cmd -> 
filePath(cmd).equals(stagingData.location()));
+      assertThat(countEqDeleteTasks(commands)).isEqualTo(1);
+      assertThat(planner(harness).reindexCount()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  void rebuildsIndexAfterRestoreWhenExternalCommitIsBelowConversionCommit() 
throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+    long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The first cycle plans S1 but does not commit, then a checkpoint is 
taken.
+      sendTrigger(harness);
+      state = harness.snapshot(1, System.currentTimeMillis());
+
+      // An external commit lands, the planner indexes it and the second cycle 
commits S1 on top.
+      insert(table, 2, "b");
+      sendTrigger(harness);
+      assertThat(planner(harness).reindexCount()).isEqualTo(1);
+      simulateConvertCommit(table, s1SnapshotId);
+    }
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The newest conversion commit is for the planned snapshot, but the 
external commit below it
+      // is not in the restored index.
+      sendTrigger(harness);
+
+      // Two inserts plus simulateConvertCommit's marker file = 3 data files 
on the target.
+      
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(3);
+      assertThat(planner(harness).reindexCount()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  void rebuildsIndexAfterRestoreFromFailedRebuild() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+    long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+    List<File> manifests = dataManifestFiles(table);
+
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The first cycle plans S1 but its commit fails.
+      sendTrigger(harness);
+
+      // The retry's rebuild fails part way, then a checkpoint is taken.
+      hide(manifests);
+      sendTrigger(harness);
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+      restore(manifests);
+      state = harness.snapshot(1, System.currentTimeMillis());
+
+      // The next retry rebuilds the index and commits S1 on the same parent.
+      sendTrigger(harness);
+      simulateConvertCommit(table, s1SnapshotId);
+    }
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The commit has the planned snapshot and parent, but the restored 
index is the one the
+      // failed rebuild left.
+      sendTrigger(harness);
+
+      // One insert plus simulateConvertCommit's marker file = 2 data files on 
the target.
+      
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2);
+    }
+  }
+
+  @Test
+  void bootstrapsAgainAfterRestoreFromFailedBootstrap() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    // The newest commit is a conversion commit, so no external commit makes 
the planner rebuild.
+    simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
+    List<File> manifests = dataManifestFiles(table);
+
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      hide(manifests);
+      sendTrigger(harness);
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+      state = harness.snapshot(1, System.currentTimeMillis());
+    }
+
+    restore(manifests);
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The checkpoint must not claim the index covers the head: the 
bootstrap did not finish.
+      sendTrigger(harness);
+
+      // One insert plus simulateConvertCommit's marker file = 2 data files on 
the target.
+      
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2);
+    }
+  }
+
   @Test
   void noMainReEmitAfterOwnCommit() throws Exception {
     Table table = createTableWithDelete(3);
@@ -963,6 +1209,55 @@ class TestEqualityConvertPlanner extends OperatorTestBase 
{
     }
   }
 
+  @Test
+  void rebuildsIndexAfterRestoreFromOlderVersion() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    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;
+    long generationBeforeRestore;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The same state as in stateRestoredAfterCheckpoint, where the index is 
reused.
+      sendTrigger(harness);
+      simulateConvertCommit(table, 
table.snapshot(STAGING_BRANCH).snapshotId());
+      sendTrigger(harness);
+      generationBeforeRestore =
+          harness.extractOutputValues().stream()
+              .mapToLong(ReadCommand::indexGeneration)
+              .max()
+              .getAsLong();
+      state = harness.snapshot(1, System.currentTimeMillis());
+    }
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(stateOfOlderVersion(state));
+      harness.open();
+
+      // An older version can record the head as indexed while the index is 
incomplete.
+      sendTrigger(harness);
+      List<ReadCommand> commands = harness.extractOutputValues();
+
+      long dataFileCount =
+          Long.parseLong(
+              
table.currentSnapshot().summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP));
+      assertThat(countDataFileTasks(commands)).isEqualTo(dataFileCount);
+
+      // The workers ignore index entries of an older generation.
+      assertThat(commands).allMatch(cmd -> cmd.indexGeneration() > 
generationBeforeRestore);
+    }
+  }
+
   @Test
   void failsOnV2PosDeleteOnStagingBranch() throws Exception {
     Table table = createTableWithDelete(2);
@@ -1175,10 +1470,49 @@ class TestEqualityConvertPlanner extends 
OperatorTestBase {
         3);
   }
 
+  private static List<File> dataManifestFiles(Table table) {
+    return table.currentSnapshot().dataManifests(table.io()).stream()
+        .map(manifest -> new File(manifest.path().replace("file:", "")))
+        .collect(Collectors.toList());
+  }
+
+  // Makes reads of the files fail until restore is called.
+  private static void hide(List<File> files) {
+    assertThat(files.stream().map(file -> 
file.renameTo(hidden(file))).collect(Collectors.toList()))
+        .containsOnly(true);
+  }
+
+  private static void restore(List<File> files) {
+    assertThat(files.stream().map(file -> 
hidden(file).renameTo(file)).collect(Collectors.toList()))
+        .containsOnly(true);
+  }
+
+  private static File hidden(File file) {
+    return new File(file.getPath() + ".hidden");
+  }
+
   private static long countDataFileTasks(List<ReadCommand> commands) {
     return commands.stream().filter(c -> c.task() instanceof 
FileScanTask).count();
   }
 
+  // Keeps the operator state entries that the version before the stateVersion 
entry wrote.
+  private static OperatorSubtaskState stateOfOlderVersion(OperatorSubtaskState 
state) {
+    Set<String> olderStateNames =
+        Sets.newHashSet(
+            "indexSnapshotId", "indexGeneration", "eqFieldIds", 
"pendingStagingSnapshotId");
+    List<OperatorStateHandle> handles = Lists.newArrayList();
+    for (OperatorStateHandle handle : state.getManagedOperatorState()) {
+      Map<String, OperatorStateHandle.StateMetaInfo> stateMetaInfos =
+          Maps.newHashMap(handle.getStateNameToPartitionOffsets());
+      stateMetaInfos.keySet().retainAll(olderStateNames);
+      handles.add(new OperatorStreamStateHandle(stateMetaInfos, 
handle.getDelegateStateHandle()));
+    }
+
+    return OperatorSubtaskState.builder()
+        .setManagedOperatorState(new StateObjectCollection<>(handles))
+        .build();
+  }
+
   private static long countEqDeleteTasks(List<ReadCommand> commands) {
     return 
commands.stream().filter(TestEqualityConvertPlanner::isEqDelete).count();
   }
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
index 34d996c3d9..0e2fe43df4 100644
--- 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
@@ -22,8 +22,13 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
 import org.apache.flink.streaming.api.watermark.Watermark;
@@ -36,6 +41,7 @@ import 
org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.maintenance.api.Trigger;
 import 
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.ContentFileUtil;
@@ -66,7 +72,13 @@ import org.slf4j.LoggerFactory;
  * completes cleanly. Errors are reported via {@link 
TaskResultAggregator#ERROR_STREAM} side output;
  * the Aggregator collects them and surfaces failure on its own watermark.
  *
- * <p>The committer is intentionally stateless: {@code bufferedResults} and 
{@code planResult} are
+ * <p>The committer keeps the index generation of the last failed cycle as 
checkpointed state. A
+ * failure in the readers or workers while the index is updated in a cycle 
with nothing to convert
+ * may have left the index without some rows, but that cycle has no commit to 
withhold. The
+ * committer withholds the next conversion commit on that generation instead: 
the next trigger then
+ * plans that staging snapshot again, which makes the planner rebuild the 
index on a new generation.
+ * The failure was reported where it happened, so the withheld cycle completes 
without an error.
+ * Apart from that the committer is stateless: {@code bufferedResults} and 
{@code planResult} are
  * not checkpointed. The maintenance framework ensures mutual exclusive tasks, 
and on restart the
  * planner re-derives its position from {@link 
#COMMITTED_STAGING_SNAPSHOT_PROPERTY} on main, so the
  * committer never receives a plan for an already-committed staging snapshot.
@@ -93,6 +105,8 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
   private transient Table table;
   private transient List<DVWriteResult> bufferedResults;
   private transient EqualityConvertPlan planResult;
+  private transient ListState<Long> failedIndexGenerationState;
+  private transient long failedIndexGeneration;
 
   private transient Counter errorCounter;
   private transient Counter addedDataFileNumCounter;
@@ -136,6 +150,32 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
     this.commitDurationMsCounter = 
taskMetricGroup.counter(COMMIT_DURATION_MS_METRIC);
   }
 
+  @Override
+  public void initializeState(StateInitializationContext context) throws 
Exception {
+    super.initializeState(context);
+    failedIndexGenerationState =
+        context
+            .getOperatorStateStore()
+            .getListState(new ListStateDescriptor<>("failedIndexGeneration", 
Types.LONG));
+
+    failedIndexGeneration = EqualityConvertPlan.NO_INDEX_GENERATION;
+    for (Long stateValue : failedIndexGenerationState.get()) {
+      Preconditions.checkState(
+          failedIndexGeneration == EqualityConvertPlan.NO_INDEX_GENERATION,
+          "failedIndexGeneration state should hold at most one value");
+      failedIndexGeneration = stateValue;
+    }
+  }
+
+  @Override
+  public void snapshotState(StateSnapshotContext context) throws Exception {
+    super.snapshotState(context);
+    failedIndexGenerationState.clear();
+    if (failedIndexGeneration != EqualityConvertPlan.NO_INDEX_GENERATION) {
+      failedIndexGenerationState.add(failedIndexGeneration);
+    }
+  }
+
   @Override
   public void processElement1(StreamRecord<DVWriteResult> record) {
     bufferedResults.add(record.getValue());
@@ -196,6 +236,7 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
             "Skipping commit for table {} task {}: a DV writer reported an 
error.",
             tableName,
             taskName);
+        failedIndexGeneration = planResult.indexGeneration();
         deleteUncommittedDVs();
         return;
       }
@@ -209,6 +250,19 @@ public class EqualityConvertCommitter extends 
AbstractStreamOperator<Trigger>
       return;
     }
 
+    if (planResult.indexGeneration() == failedIndexGeneration) {
+      LOG.warn(
+          "Not committing staging snapshot {} for table {} task {}: an index 
update failed on "
+              + "index generation {}, so the next trigger rebuilds the index 
and converts the "
+              + "snapshot again.",
+          planResult.stagingSnapshotId(),
+          tableName,
+          taskName,
+          failedIndexGeneration);
+      deleteUncommittedDVs();
+      return;
+    }
+
     table.refresh();
 
     List<DeleteFile> allDvFiles = Lists.newArrayList();
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
index 4aa326d19a..3226fbf4a2 100644
--- 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
@@ -31,8 +31,8 @@ import 
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
  * EqualityConvertCommitter} (for data files and staging deletes to commit).
  *
  * <p>A no-op cycle is encoded by {@link #stagingSnapshotId()} == {@link
- * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to 
construct one and {@link
- * #noOp()} to check.
+ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long, long)} to 
construct one and
+ * {@link #noOp()} to check.
  *
  * <p>A cycle-start marker is a no-op plan whose done timestamp is never 
reached. Use {@link
  * #cycleStart(long)} to construct one.
@@ -50,6 +50,9 @@ import 
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
  * @param triggerTimestamp original trigger timestamp, forwarded by the 
Committer to the Aggregator
  * @param doneTimestamp timestamp after which all phase watermarks have been 
emitted; the DVWriter
  *     should only process the result when the watermark reaches or exceeds 
this value
+ * @param indexGeneration generation of the worker index the cycle ran on. The 
committer does not
+ *     commit a cycle on a generation whose update failed. {@link 
#NO_INDEX_GENERATION} for the
+ *     cycle-start marker.
  */
 @Internal
 public record EqualityConvertPlan(
@@ -59,11 +62,14 @@ public record EqualityConvertPlan(
     long stagingSnapshotId,
     Long mainSnapshotId,
     long triggerTimestamp,
-    long doneTimestamp)
+    long doneTimestamp,
+    long indexGeneration)
     implements Serializable {
 
   public static final long NO_OP_STAGING_SNAPSHOT_ID = -1L;
 
+  public static final long NO_INDEX_GENERATION = -1L;
+
   /** Returns {@code true} if this cycle has nothing to commit. */
   public boolean noOp() {
     return stagingSnapshotId == NO_OP_STAGING_SNAPSHOT_ID;
@@ -71,7 +77,7 @@ public record EqualityConvertPlan(
 
   /** No-op cycle: empty file lists, sentinel staging snapshot id. */
   public static EqualityConvertPlan noOp(
-      Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) {
+      Long mainSnapshotId, long triggerTimestamp, long doneTimestamp, long 
indexGeneration) {
     return new EqualityConvertPlan(
         ImmutableList.of(),
         ImmutableList.of(),
@@ -79,7 +85,8 @@ public record EqualityConvertPlan(
         NO_OP_STAGING_SNAPSHOT_ID,
         mainSnapshotId,
         triggerTimestamp,
-        doneTimestamp);
+        doneTimestamp,
+        indexGeneration);
   }
 
   /**
@@ -87,6 +94,6 @@ public record EqualityConvertPlan(
    * until the cycle's actual plan replaces it.
    */
   public static EqualityConvertPlan cycleStart(long triggerTimestamp) {
-    return noOp(null, triggerTimestamp, Long.MAX_VALUE);
+    return noOp(null, triggerTimestamp, Long.MAX_VALUE, NO_INDEX_GENERATION);
   }
 }
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 96e5861f1c..1bf08b0976 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
@@ -70,9 +70,8 @@ import org.slf4j.LoggerFactory;
  * <ol>
  *   <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 #ensureIndexCurrent}: rebuilds the worker index from main 
unless it reflects the
+ *       target branch head, see {@link #indexReflects}.
  *   <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.
@@ -109,6 +108,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   private static final String SKIPPED_NO_OP_CYCLES_METRIC = 
"skippedNoOpCycles";
   private static final String REINDEX_COUNT_METRIC = "reindexCount";
 
+  private static final int STATE_VERSION = 1;
+
   private final String tableName;
   private final String taskName;
   private final TableLoader tableLoader;
@@ -123,10 +124,12 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   private transient ListState<Long> indexSnapshotState;
   // Generation of the worker's current index build.
   private transient ListState<Long> indexGenerationState;
-  // Staging snapshot the last emitted plan covered.
+  // Staging snapshot the index changed for until its conversion commit shows 
up on the target.
   private transient ListState<Long> pendingStagingSnapshotState;
   // Equality field IDs the index was built with, allows to detect 
reconfiguration.
   private transient ListState<Integer> eqFieldIdsState;
+  // Absent in state of a version that recorded the index as current before it 
was complete.
+  private transient ListState<Integer> stateVersionState;
 
   private transient Table table;
 
@@ -134,8 +137,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   private transient Long lastStagingSnapshotId;
   private transient Long indexSnapshotId;
   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.
+  // Staging snapshot the index changed for until its conversion commit shows 
up on the target
+  // branch, checkpointed so it survives a restore taken mid-cycle.
   private transient Long pendingStagingSnapshotId;
 
   private transient long nextPhaseTs;
@@ -238,6 +241,20 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
           "pendingStagingSnapshotId state should hold at most one value");
       pendingStagingSnapshotId = stateValue;
     }
+
+    stateVersionState =
+        context
+            .getOperatorStateStore()
+            .getListState(new ListStateDescriptor<>("stateVersion", 
Types.INT));
+
+    // The index recorded by an older version can be incomplete while it looks 
current.
+    if (context.isRestored() && !stateVersionState.get().iterator().hasNext()) 
{
+      LOG.info(
+          "Restored state of an older version for table {} task {}, rebuilding 
the worker index.",
+          tableName,
+          taskName);
+      indexSnapshotId = null;
+    }
   }
 
   @Override
@@ -260,6 +277,9 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
     if (pendingStagingSnapshotId != null) {
       pendingStagingSnapshotState.add(pendingStagingSnapshotId);
     }
+
+    stateVersionState.clear();
+    stateVersionState.add(STATE_VERSION);
   }
 
   @Override
@@ -279,12 +299,12 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
       Snapshot mainSnapshot = table.snapshot(targetBranch);
       currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() 
: null;
 
-      LastCommittedWork committedWork = refreshStagingCursor(mainSnapshot);
+      refreshStagingCursor(mainSnapshot);
 
       Snapshot nextToProcess =
           nextUnprocessedStagingSnapshot(table.snapshot(stagingBranch), 
mainSnapshot);
 
-      ensureIndexCurrent(mainSnapshot, committedWork, nextToProcess);
+      ensureIndexCurrent(mainSnapshot);
 
       if (nextToProcess == null) {
         LOG.info("Nothing new to convert on staging branch '{}'.", 
stagingBranch);
@@ -303,45 +323,35 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
 
   /**
    * 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.
+   * branch. Does nothing when the target has not moved since the last trigger 
and the cursor
+   * therefore cannot have changed.
    */
-  private LastCommittedWork refreshStagingCursor(Snapshot mainSnapshot) {
+  private void refreshStagingCursor(Snapshot mainSnapshot) {
     Long currentMainSnapshotId = mainSnapshot != null ? 
mainSnapshot.snapshotId() : null;
 
     if (Objects.equals(lastMainSnapshotId, currentMainSnapshotId)) {
-      return null;
+      return;
     }
 
-    LastCommittedWork info = discoverLastCommittedWork(mainSnapshot);
-    updateLastStagingSnapshotId(info);
-    return info;
+    updateLastStagingSnapshotId(discoverLastCommittedWork(mainSnapshot));
   }
 
   /**
-   * 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.
+   * Rebuilds the worker index unless it reflects {@code mainSnapshot}, see 
{@link #indexReflects}.
    */
-  private void ensureIndexCurrent(
-      Snapshot mainSnapshot, LastCommittedWork committedWork, Snapshot 
nextToProcess) {
+  private void ensureIndexCurrent(Snapshot mainSnapshot) {
     if (mainSnapshot == null) {
+      // The target branch has no rows, so the index must not have any either. 
A new generation
+      // makes the worker drop the rows indexed so far, e.g. by a cycle that 
did not commit.
+      indexGeneration++;
+      indexSnapshotId = null;
+      pendingStagingSnapshotId = 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) {
+    if (!indexReflects(mainSnapshot)) {
+      boolean bootstrap = indexSnapshotId == null;
       LOG.info(
           "{} worker index from main snapshot {} for field IDs {}.",
           bootstrap ? "Bootstrapping" : "Reindexing",
@@ -350,20 +360,58 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
       rebuildIndex(mainSnapshot, !bootstrap);
     }
 
+    indexSnapshotId = mainSnapshot.snapshotId();
+    pendingStagingSnapshotId = null;
     lastMainSnapshotId = mainSnapshot.snapshotId();
   }
 
+  /**
+   * Returns whether the index reflects {@code mainSnapshot}.
+   *
+   * <p>Planning a staging snapshot changes the index before its conversion 
commit exists: its eq
+   * deletes consume the entries they match and its data files are added. So 
with a snapshot
+   * planned, the index reflects the head only if that snapshot's conversion 
commit sits directly on
+   * top of the indexed snapshot. Without one, the head must be the indexed 
snapshot. Any other head
+   * holds commits the index never saw (external commits, rollbacks, a restore 
from before a
+   * conversion commit) or follows a cycle that did not commit.
+   *
+   * <p>After a restore, the conversion commit can come from a retry of the 
planned cycle. A cycle
+   * that only failed to commit leaves the same index as its retry. A rebuild 
that fails part way
+   * clears {@link #indexSnapshotId}, and the committer does not commit a 
conversion on a generation
+   * whose update failed downstream.
+   */
+  private boolean indexReflects(Snapshot mainSnapshot) {
+    if (indexSnapshotId == null) {
+      return false;
+    }
+
+    if (pendingStagingSnapshotId == null) {
+      return mainSnapshot.snapshotId() == indexSnapshotId;
+    }
+
+    String committedStaging =
+        
mainSnapshot.summary().get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+    return committedStaging != null
+        && Long.parseLong(committedStaging) == pendingStagingSnapshotId
+        && Objects.equals(mainSnapshot.parentId(), indexSnapshotId);
+  }
+
   /**
    * 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.
+   * complete earlier index. Entries of an incomplete one are ignored at the 
new generation.
    *
    * <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.
+   *
+   * <p>Clears {@link #indexSnapshotId} before emitting anything, and the 
caller records the head
+   * only after this returns. Neither the next trigger nor a restore from a 
checkpoint taken after a
+   * rebuild that failed part way trusts the index it left.
    */
   private void rebuildIndex(Snapshot mainSnapshot, boolean evictStaleKeys) {
+    indexSnapshotId = null;
     long generation = indexGeneration + 1;
 
     if (evictStaleKeys) {
@@ -374,7 +422,6 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
       reindexCounter.inc();
     }
 
-    indexSnapshotId = mainSnapshot.snapshotId();
     indexGeneration = generation;
     emitMainDataReadCommands(mainSnapshot);
   }
@@ -399,7 +446,7 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
   /**
    * Walks main back from head looking for the most recent snapshot tagged 
with {@link
    * EqualityConvertCommitter#COMMITTED_STAGING_SNAPSHOT_PROPERTY}. Returns 
the staging snapshot id
-   * recorded there (or {@code null} if not reached) and the count of 
intervening external commits.
+   * recorded there, or {@code null} if not reached.
    *
    * <p>The walk stops at whichever comes first:
    *
@@ -410,7 +457,6 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
    */
   private LastCommittedWork discoverLastCommittedWork(Snapshot mainSnapshot) {
     Long lastCommittedStaging = null;
-    int externalCount = 0;
     boolean reachedLastInspected = false;
     Snapshot current = mainSnapshot;
     while (current != null) {
@@ -426,15 +472,13 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
         break;
       }
 
-      externalCount++;
       current = parentOf(current);
     }
 
-    return new LastCommittedWork(lastCommittedStaging, externalCount, 
reachedLastInspected);
+    return new LastCommittedWork(lastCommittedStaging, reachedLastInspected);
   }
 
-  private record LastCommittedWork(
-      Long lastCommittedStaging, int externalCommitCount, boolean 
reachedLastInspected) {}
+  private record LastCommittedWork(Long lastCommittedStaging, boolean 
reachedLastInspected) {}
 
   /**
    * Walks staging history from head back to the stop point and returns the 
oldest unprocessed
@@ -533,7 +577,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
                 stagingSnapshot.snapshotId(),
                 currentMainSnapshotId,
                 triggerTs,
-                nextPhaseTs)));
+                nextPhaseTs,
+                indexGeneration)));
 
     advancePhase();
   }
@@ -680,7 +725,8 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
     output.collect(
         METADATA_STREAM,
         new StreamRecord<>(
-            EqualityConvertPlan.noOp(currentMainSnapshotId, triggerTimestamp, 
nextPhaseTs)));
+            EqualityConvertPlan.noOp(
+                currentMainSnapshotId, triggerTimestamp, nextPhaseTs, 
indexGeneration)));
     advancePhase();
   }
 
@@ -700,7 +746,7 @@ public class EqualityConvertPlanner extends 
AbstractStreamOperator<ReadCommand>
         output.collect(
             new StreamRecord<>(
                 ReadCommand.dataFile(
-                    task, indexSnapshotId, indexGeneration, 
dataSequenceNumber(task.file())),
+                    task, commitSnapshotId, indexGeneration, 
dataSequenceNumber(task.file())),
                 nextPhaseTs));
       }
     } catch (IOException e) {
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 2e03a0655e..486ab0fe42 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
@@ -1473,6 +1473,114 @@ class TestConvertEqualityDeletes extends 
MaintenanceTaskTestBase {
     }
   }
 
+  @Test
+  void testIndexUpdateFailureOnIdleTriggerIsRepaired() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile firstDelete = writeEqualityDelete(table, 1, "a");
+    
table.newRowDelta().addDeletes(firstDelete).toBranch(STAGING_BRANCH).commit();
+
+    appendConvertTask(STAGING_BRANCH);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = infra.env().executeAsync();
+
+      long time1 = System.currentTimeMillis();
+      infra.source().sendRecord(Trigger.create(time1, 0), time1);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+      // An external commit lands while staging is idle, and its data file is 
unreadable while the
+      // next trigger updates the index.
+      DataFile externalData = writeDataFile(table, createRecord(2, "b"));
+      table.newAppend().appendFile(externalData).commit();
+      File externalFile = new File(externalData.location().replace("file:", 
""));
+      File hiddenFile = new File(externalFile.getPath() + ".hidden");
+      assertThat(externalFile.renameTo(hiddenFile)).isTrue();
+
+      long time2 = time1 + 1;
+      infra.source().sendRecord(Trigger.create(time2, 0), time2);
+      
assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isFalse();
+
+      assertThat(hiddenFile.renameTo(externalFile)).isTrue();
+
+      DeleteFile secondDelete = writeEqualityDelete(table, 2, "b");
+      
table.newRowDelta().addDeletes(secondDelete).toBranch(STAGING_BRANCH).commit();
+
+      // The index may miss the external row, so the next conversion is not 
committed.
+      table.refresh();
+      long mainSnapshotId = table.currentSnapshot().snapshotId();
+      long time3 = time2 + 1;
+      infra.source().sendRecord(Trigger.create(time3, 0), time3);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+      table.refresh();
+      
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+
+      // Planned again, the conversion rebuilds the index and deletes the 
external row.
+      long time4 = time3 + 1;
+      infra.source().sendRecord(Trigger.create(time4, 0), time4);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+      assertRecords(table, ImmutableList.of());
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  @Test
+  void testFailedConversionOnEmptyTargetIsRetried() throws Exception {
+    Table table = createTableWithDelete(3);
+
+    // The target branch has no snapshot until the first conversion commit.
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    // An upsert of (1, "a") and an insert whose data file is unreadable 
during the first cycle.
+    DataFile reinsert = writeDataFile(table, createRecord(1, "a"));
+    DataFile insert = writeDataFile(table, createRecord(2, "b"));
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table
+        .newRowDelta()
+        .addRows(reinsert)
+        .addRows(insert)
+        .addDeletes(eqDelete)
+        .toBranch(STAGING_BRANCH)
+        .commit();
+
+    File insertFile = new File(insert.location().replace("file:", ""));
+    File hiddenFile = new File(insertFile.getPath() + ".hidden");
+    assertThat(insertFile.renameTo(hiddenFile)).isTrue();
+
+    appendConvertTask(STAGING_BRANCH);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = infra.env().executeAsync();
+
+      // The first cycle indexes the re-insert, fails on the unreadable file 
and does not commit.
+      long time1 = System.currentTimeMillis();
+      infra.source().sendRecord(Trigger.create(time1, 0), time1);
+      
assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isFalse();
+      table.refresh();
+      assertThat(table.snapshot(SnapshotRef.MAIN_BRANCH)).isNull();
+
+      assertThat(hiddenFile.renameTo(insertFile)).isTrue();
+
+      // The retry must not delete the re-insert indexed by the first cycle.
+      long time2 = time1 + 1;
+      infra.source().sendRecord(Trigger.create(time2, 0), time2);
+      assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+      assertRecords(table, ImmutableList.of(createRecord(1, "a"), 
createRecord(2, "b")));
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
   private void appendConvertTask() {
     appendConvertTask(STAGING_BRANCH);
   }
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
index e39d05b66f..d18f81a54a 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.flink.maintenance.api;
 
 import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
 import static 
org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME;
+import static 
org.apache.iceberg.flink.maintenance.operator.EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY;
 import static 
org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -30,12 +31,18 @@ import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.test.junit5.InjectClusterClient;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileContent;
@@ -104,18 +111,7 @@ class TestConvertEqualityDeletesE2E extends 
OperatorTestBase {
     table.refresh();
     assertThat(dvCountOnMain(table)).isZero();
 
-    TableMaintenance.forTable(env, tableLoader(), LOCK_FACTORY)
-        .uidSuffix("ConvertEqualityDeletesE2EUID-" + stagingBranch)
-        .rateLimit(Duration.ofMillis(50))
-        .lockCheckDelay(Duration.ofMillis(50))
-        .add(
-            ConvertEqualityDeletes.builder()
-                .scheduleOnInterval(Duration.ofMillis(100))
-                .stagingBranch(stagingBranch)
-                .targetBranch(SnapshotRef.MAIN_BRANCH)
-                .equalityFieldColumns(ImmutableList.of("id", "data"))
-                .parallelism(2))
-        .append();
+    appendConvertTask(env, stagingBranch, "ConvertEqualityDeletesE2EUID-" + 
stagingBranch);
 
     JobClient jobClient = null;
     try {
@@ -211,6 +207,97 @@ class TestConvertEqualityDeletesE2E extends 
OperatorTestBase {
     }
   }
 
+  /**
+   * Restoring from a savepoint taken before a committed cycle leaves the 
converter's index without
+   * the rows that cycle copied to the target. The next conversion must 
rebuild it, or a delete for
+   * one of those rows resolves to nothing.
+   */
+  @Test
+  void reindexesAfterRestoreFromBeforeCommittedCycle(
+      @TempDir File savepointDir, @InjectClusterClient ClusterClient<?> 
clusterClient)
+      throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table
+        .newRowDelta()
+        .addDeletes(writeEqualityDelete(table, 1, "a"))
+        .toBranch(STAGING_BRANCH)
+        .commit();
+    table.refresh();
+
+    appendConvertTask(env, STAGING_BRANCH, "ConvertEqualityDeletesRestoreUID");
+    String savepointPath;
+    JobClient jobClient = null;
+    try {
+      jobClient = env.executeAsync();
+      awaitConverted(table, table.snapshot(STAGING_BRANCH).snapshotId());
+      savepointPath =
+          jobClient.triggerSavepoint(savepointDir.getPath(), 
SavepointFormatType.CANONICAL).get();
+
+      // This cycle copies S2's row to the target after the savepoint.
+      table.newAppend().appendFile(writeDataFile(table, 2, 
"b")).toBranch(STAGING_BRANCH).commit();
+      table.refresh();
+      awaitConverted(table, table.snapshot(STAGING_BRANCH).snapshotId());
+    } finally {
+      closeJobClient(jobClient);
+    }
+
+    table
+        .newRowDelta()
+        .addDeletes(writeEqualityDelete(table, 2, "b"))
+        .toBranch(STAGING_BRANCH)
+        .commit();
+    table.refresh();
+    long deleteSnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+    StreamExecutionEnvironment restoredEnv = 
StreamExecutionEnvironment.getExecutionEnvironment();
+    appendConvertTask(restoredEnv, STAGING_BRANCH, 
"ConvertEqualityDeletesRestoreUID");
+    StreamGraph streamGraph = restoredEnv.getStreamGraph();
+    // Submitting the graph directly skips the executor, which is what 
normally assigns the job id
+    streamGraph.setJobId(JobID.generate());
+    
streamGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath));
+    JobID restoredJobId = clusterClient.submitJob(streamGraph).get();
+    try {
+      awaitConverted(table, deleteSnapshotId);
+      SimpleDataUtil.assertTableRecords(table, ImmutableList.of());
+    } finally {
+      clusterClient.cancel(restoredJobId).get();
+      Awaitility.await()
+          .until(() -> 
clusterClient.getJobStatus(restoredJobId).get().isTerminalState());
+    }
+  }
+
+  private void appendConvertTask(
+      StreamExecutionEnvironment environment, String stagingBranch, String 
uidSuffix)
+      throws IOException {
+    TableMaintenance.forTable(environment, tableLoader(), LOCK_FACTORY)
+        .uidSuffix(uidSuffix)
+        .rateLimit(Duration.ofMillis(50))
+        .lockCheckDelay(Duration.ofMillis(50))
+        .add(
+            ConvertEqualityDeletes.builder()
+                .scheduleOnInterval(Duration.ofMillis(100))
+                .stagingBranch(stagingBranch)
+                .targetBranch(SnapshotRef.MAIN_BRANCH)
+                .equalityFieldColumns(ImmutableList.of("id", "data"))
+                .parallelism(2))
+        .append();
+  }
+
+  private static void awaitConverted(Table table, long stagingSnapshotId) {
+    Awaitility.await()
+        .atMost(Duration.ofMinutes(5))
+        .pollInterval(Duration.ofMillis(200))
+        .untilAsserted(
+            () -> {
+              table.refresh();
+              assertThat(table.currentSnapshot().summary())
+                  .containsEntry(
+                      COMMITTED_STAGING_SNAPSHOT_PROPERTY, 
String.valueOf(stagingSnapshotId));
+            });
+  }
+
   private static void awaitSucceeded(List<String> counterKey, long expected) {
     Awaitility.await()
         .atMost(Duration.ofMinutes(3))
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
index 35aa9908bc..e52f53f942 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
@@ -25,6 +25,7 @@ import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.stream.Collectors;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
@@ -54,6 +55,8 @@ import org.junit.jupiter.api.io.TempDir;
 
 class TestEqualityConvertCommitter extends OperatorTestBase {
 
+  private static final long INDEX_GENERATION = 1L;
+
   @TempDir private Path tempDir;
 
   @Test
@@ -78,7 +81,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
               stagingSnapshotId,
               snapshotIdBefore,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
       harness.processBothWatermarks(new Watermark(doneTs));
@@ -116,7 +120,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               doneTs - 2,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 2));
 
@@ -165,7 +170,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               triggerTs,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
       harness.processElement2(new StreamRecord<>(planResult));
       harness.processBothWatermarks(new Watermark(doneTs));
 
@@ -203,7 +209,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
       harness.open();
 
       long doneTs = System.currentTimeMillis();
-      EqualityConvertPlan planResult = EqualityConvertPlan.noOp(null, doneTs - 
1, doneTs);
+      EqualityConvertPlan planResult =
+          EqualityConvertPlan.noOp(null, doneTs - 1, doneTs, INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
       harness.processBothWatermarks(new Watermark(doneTs));
@@ -235,7 +242,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       // Receive an abort signal from the DVWriter followed by the planning 
planResult.
       harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, doneTs));
@@ -250,6 +258,129 @@ class TestEqualityConvertCommitter extends 
OperatorTestBase {
     }
   }
 
+  @Test
+  void withholdsNextConversionCommitAfterFailedIdleCycle() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+    DeleteFile writtenDv = writeDV(table, stagingDataFile.location(), 0L);
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      failIdleRebuild(harness, mainSnapshotId);
+
+      // The next conversion is not committed, and the DV it wrote is deleted.
+      harness.processElement1(
+          new StreamRecord<>(
+              new DVWriteResult(Lists.newArrayList(writtenDv), 
Lists.newArrayList()), 20L));
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
20L, 1L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      table.refresh();
+      
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+      
assertThat(table.io().newInputFile(writtenDv.location()).exists()).isFalse();
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+
+      // Planned again with a rebuilt index on a new generation, the 
conversion commits.
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
30L, 2L), 29L));
+      harness.processBothWatermarks(new Watermark(30L));
+
+      assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+    }
+  }
+
+  @Test
+  void withholdsNextConversionCommitAfterRestore() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+    OperatorSubtaskState state;
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      failIdleRebuild(harness, mainSnapshotId);
+
+      state = harness.snapshot(1, 11L);
+    }
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.initializeState(state);
+      harness.open();
+
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
20L, 1L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      table.refresh();
+      
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+    }
+  }
+
+  @Test
+  void commitsReplannedConversionAfterFailedConversion() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      // A failed conversion is planned again with a rebuilt index, so nothing 
is withheld.
+      harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, 10L));
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
10L, 1L), 9L));
+      harness.processBothWatermarks(new Watermark(10L));
+
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
20L, 2L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+    }
+  }
+
+  @Test
+  void commitsNextConversionAfterCleanRebuild() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      failIdleRebuild(harness, mainSnapshotId);
+
+      // An external commit makes the next idle cycle rebuild the index, which 
repairs the failure.
+      harness.processElement2(
+          new StreamRecord<>(EqualityConvertPlan.noOp(mainSnapshotId, 19L, 
20L, 2L), 19L));
+      harness.processBothWatermarks(new Watermark(20L));
+
+      harness.processElement2(
+          new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId, 
30L, 2L), 29L));
+      harness.processBothWatermarks(new Watermark(30L));
+
+      assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+    }
+  }
+
   @Test
   void failsCommitWhenExternalCommitLandsAfterPlan() throws Exception {
     Table table = createTable(3, FileFormat.PARQUET);
@@ -272,7 +403,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               stagingSnapshotId,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
 
@@ -330,7 +462,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               stagingSnapshotId,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       // Cycle 1 commits, advancing the target past mainSnapshotIdAtPlan and 
writing the marker.
       harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
@@ -353,7 +486,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               stagingSnapshotId,
               mainSnapshotIdAtPlan,
               doneTs2 - 1,
-              doneTs2);
+              doneTs2,
+              INDEX_GENERATION);
 
       harness.processElement2(new StreamRecord<>(replay, doneTs2 - 1));
       harness.processBothWatermarks(new Watermark(doneTs2));
@@ -399,7 +533,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               777L,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -450,7 +585,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               42L,
               snapshotIdBefore,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       // One writer task wrote a DV (rewriting an existing one); a sibling 
task aborted.
       harness.processElement1(
@@ -505,7 +641,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               888L,
               mainSnapshotIdAtPlan,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -557,7 +694,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               123L,
               mainSnapshotId,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -616,7 +754,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase 
{
               123L,
               mainSnapshotId,
               doneTs - 1,
-              doneTs);
+              doneTs,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(
@@ -655,6 +794,39 @@ class TestEqualityConvertCommitter extends 
OperatorTestBase {
             SnapshotRef.MAIN_BRANCH));
   }
 
+  // The planner rebuilt the index to generation 1 in a cycle with nothing to 
convert, and a worker
+  // failed to update it.
+  private static void failIdleRebuild(
+      TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, 
Trigger> harness,
+      long mainSnapshotId)
+      throws Exception {
+    harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, 10L));
+    harness.processElement2(
+        new StreamRecord<>(EqualityConvertPlan.noOp(mainSnapshotId, 9L, 10L, 
1L), 9L));
+    harness.processBothWatermarks(new Watermark(10L));
+  }
+
+  private static String committedStagingSnapshot(Table table) {
+    table.refresh();
+    return table
+        .currentSnapshot()
+        .summary()
+        .get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+  }
+
+  private static EqualityConvertPlan conversionPlan(
+      DataFile stagingDataFile, long mainSnapshotId, long doneTs, long 
indexGeneration) {
+    return new EqualityConvertPlan(
+        Lists.newArrayList(stagingDataFile),
+        Lists.newArrayList(),
+        Lists.newArrayList(),
+        42L,
+        mainSnapshotId,
+        doneTs - 1,
+        doneTs,
+        indexGeneration);
+  }
+
   private static List<Watermark> 
watermarks(TwoInputStreamOperatorTestHarness<?, ?, ?> harness) {
     return harness.getOutput().stream()
         .filter(Watermark.class::isInstance)
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
index 93614fe8ab..b7dd1e308d 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
@@ -40,6 +40,7 @@ import org.junit.jupiter.api.Test;
 class TestEqualityConvertDVWriter extends OperatorTestBase {
 
   private static final byte[] EMPTY_PARTITION = new byte[0];
+  private static final long INDEX_GENERATION = 1L;
 
   @Test
   void writesDVFileForSinglePosition() throws Exception {
@@ -247,7 +248,8 @@ class TestEqualityConvertDVWriter extends OperatorTestBase {
               1L,
               null,
               0L,
-              0L);
+              0L,
+              INDEX_GENERATION);
 
       // New position 1 in the same file. Must merge it with the staged 
position 0.
       harness.processElement1(
@@ -288,7 +290,8 @@ class TestEqualityConvertDVWriter extends OperatorTestBase {
               1L,
               plannedSnapshotId,
               0L,
-              0L);
+              0L,
+              INDEX_GENERATION);
 
       harness.processElement1(
           new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, 
EMPTY_PARTITION, 0L), time));
@@ -491,7 +494,14 @@ class TestEqualityConvertDVWriter extends OperatorTestBase 
{
 
   private static EqualityConvertPlan emptyEqualityConvertPlan() {
     return new EqualityConvertPlan(
-        Lists.newArrayList(), Lists.newArrayList(), Lists.newArrayList(), 1L, 
null, 0L, 0L);
+        Lists.newArrayList(),
+        Lists.newArrayList(),
+        Lists.newArrayList(),
+        1L,
+        null,
+        0L,
+        0L,
+        INDEX_GENERATION);
   }
 
   private static String getFirstDataFilePath(Table table) {
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 1482ce8726..c98de623e7 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
@@ -26,9 +26,13 @@ import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.OperatorStreamStateHandle;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
@@ -45,6 +49,7 @@ import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.RewriteFiles;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SnapshotSummary;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.data.FileHelpers;
 import org.apache.iceberg.data.GenericAppenderHelper;
@@ -53,9 +58,12 @@ import org.apache.iceberg.data.Record;
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.flink.maintenance.api.Trigger;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 class TestEqualityConvertPlanner extends OperatorTestBase {
 
@@ -353,6 +361,13 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
       sendTrigger(harness);
       assertThat(harness.extractOutputValues()).hasSize(firstTriggerCount);
       assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1);
+
+      // Each trigger emits a cycle start marker followed by the plan. The 
bootstrap built
+      // generation 1, and the second trigger reused it.
+      List<StreamRecord<EqualityConvertPlan>> metadata =
+          
Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM));
+      assertThat(metadata.get(1).getValue().indexGeneration()).isEqualTo(1L);
+      assertThat(metadata.get(3).getValue().indexGeneration()).isEqualTo(1L);
     }
   }
 
@@ -668,6 +683,38 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
     }
   }
 
+  @Test
+  void startsNewIndexGenerationWhenReplanningOnEmptyTarget() throws Exception {
+    Table table = createTableWithDelete(3);
+
+    // The target branch has no snapshot until the first conversion commit.
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DataFile reinsert = writeDataFile(table, createRecord(1, "a"));
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    
table.newRowDelta().addRows(reinsert).addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The first cycle indexes the re-insert but does not commit, so the 
same staging snapshot is
+      // planned again. Its eq delete must not resolve against the re-insert 
of the first cycle.
+      sendTrigger(harness);
+      sendTrigger(harness);
+
+      List<ReadCommand> eqDeletes =
+          harness.extractOutputValues().stream()
+              .filter(cmd -> cmd.task() instanceof EqualityDeleteFileScanTask)
+              .collect(Collectors.toList());
+      assertThat(eqDeletes).hasSize(2);
+      assertThat(eqDeletes.get(1).indexGeneration())
+          .isGreaterThan(eqDeletes.get(0).indexGeneration());
+    }
+  }
+
   @Test
   void rebuildsIndexAfterRestoreOfUncommittedCycle() throws Exception {
     Table table = createTableWithDelete(3);
@@ -687,8 +734,7 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
       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.
+      // own marker, which the planner sees on the next trigger.
       sendTrigger(harness);
       assertThat(harness.extractOutputValues()).hasSize(3);
       simulateConvertCommit(table, 
table.snapshot(STAGING_BRANCH).snapshotId());
@@ -722,6 +768,206 @@ class TestEqualityConvertPlanner extends OperatorTestBase 
{
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  void rebuildsIndexAfterRestoreFromBeforeCommittedCycle(boolean 
plannerSawConversionCommit)
+      throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DataFile stagingData;
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The checkpoint follows the first cycle, which converts S1 and 
commits. The planner sees
+      // that commit only if another trigger follows.
+      sendTrigger(harness);
+      simulateConvertCommit(table, 
table.snapshot(STAGING_BRANCH).snapshotId());
+      if (plannerSawConversionCommit) {
+        sendTrigger(harness);
+      }
+
+      state = harness.snapshot(1, System.currentTimeMillis());
+
+      // The second cycle indexes S2's data file as staging data and copies it 
to the target. No
+      // checkpoint follows it.
+      stagingData = writeDataFile(table, createRecord(2, "b"));
+      
table.newAppend().appendFile(stagingData).toBranch(STAGING_BRANCH).commit();
+      table.refresh();
+      long s2SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+      sendTrigger(harness);
+      table
+          .newRowDelta()
+          .addRows(stagingData)
+          .set(
+              EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY,
+              String.valueOf(s2SnapshotId))
+          .commit();
+      table.refresh();
+    }
+
+    DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
+    
table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The index restored with the planner never held S2's row, so the 
planner rebuilds it from
+      // the target before resolving the delete for that row.
+      sendTrigger(harness);
+      List<ReadCommand> commands = harness.extractOutputValues();
+
+      assertThat(commands).anyMatch(cmd -> 
filePath(cmd).equals(stagingData.location()));
+      assertThat(countEqDeleteTasks(commands)).isEqualTo(1);
+      assertThat(planner(harness).reindexCount()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  void rebuildsIndexAfterRestoreWhenExternalCommitIsBelowConversionCommit() 
throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+    long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The first cycle plans S1 but does not commit, then a checkpoint is 
taken.
+      sendTrigger(harness);
+      state = harness.snapshot(1, System.currentTimeMillis());
+
+      // An external commit lands, the planner indexes it and the second cycle 
commits S1 on top.
+      insert(table, 2, "b");
+      sendTrigger(harness);
+      assertThat(planner(harness).reindexCount()).isEqualTo(1);
+      simulateConvertCommit(table, s1SnapshotId);
+    }
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The newest conversion commit is for the planned snapshot, but the 
external commit below it
+      // is not in the restored index.
+      sendTrigger(harness);
+
+      // Two inserts plus simulateConvertCommit's marker file = 3 data files 
on the target.
+      
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(3);
+      assertThat(planner(harness).reindexCount()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  void rebuildsIndexAfterRestoreFromFailedRebuild() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+    table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+    table.refresh();
+    long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+    List<File> manifests = dataManifestFiles(table);
+
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The first cycle plans S1 but its commit fails.
+      sendTrigger(harness);
+
+      // The retry's rebuild fails part way, then a checkpoint is taken.
+      hide(manifests);
+      sendTrigger(harness);
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+      restore(manifests);
+      state = harness.snapshot(1, System.currentTimeMillis());
+
+      // The next retry rebuilds the index and commits S1 on the same parent.
+      sendTrigger(harness);
+      simulateConvertCommit(table, s1SnapshotId);
+    }
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The commit has the planned snapshot and parent, but the restored 
index is the one the
+      // failed rebuild left.
+      sendTrigger(harness);
+
+      // One insert plus simulateConvertCommit's marker file = 2 data files on 
the target.
+      
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2);
+    }
+  }
+
+  @Test
+  void bootstrapsAgainAfterRestoreFromFailedBootstrap() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+    table.refresh();
+
+    // The newest commit is a conversion commit, so no external commit makes 
the planner rebuild.
+    simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
+    List<File> manifests = dataManifestFiles(table);
+
+    OperatorSubtaskState state;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      hide(manifests);
+      sendTrigger(harness);
+      
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+      state = harness.snapshot(1, System.currentTimeMillis());
+    }
+
+    restore(manifests);
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(state);
+      harness.open();
+
+      // The checkpoint must not claim the index covers the head: the 
bootstrap did not finish.
+      sendTrigger(harness);
+
+      // One insert plus simulateConvertCommit's marker file = 2 data files on 
the target.
+      
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2);
+    }
+  }
+
   @Test
   void noMainReEmitAfterOwnCommit() throws Exception {
     Table table = createTableWithDelete(3);
@@ -963,6 +1209,55 @@ class TestEqualityConvertPlanner extends OperatorTestBase 
{
     }
   }
 
+  @Test
+  void rebuildsIndexAfterRestoreFromOlderVersion() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    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;
+    long generationBeforeRestore;
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.open();
+
+      // The same state as in stateRestoredAfterCheckpoint, where the index is 
reused.
+      sendTrigger(harness);
+      simulateConvertCommit(table, 
table.snapshot(STAGING_BRANCH).snapshotId());
+      sendTrigger(harness);
+      generationBeforeRestore =
+          harness.extractOutputValues().stream()
+              .mapToLong(ReadCommand::indexGeneration)
+              .max()
+              .getAsLong();
+      state = harness.snapshot(1, System.currentTimeMillis());
+    }
+
+    try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+        createHarness(STAGING_BRANCH)) {
+      harness.initializeState(stateOfOlderVersion(state));
+      harness.open();
+
+      // An older version can record the head as indexed while the index is 
incomplete.
+      sendTrigger(harness);
+      List<ReadCommand> commands = harness.extractOutputValues();
+
+      long dataFileCount =
+          Long.parseLong(
+              
table.currentSnapshot().summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP));
+      assertThat(countDataFileTasks(commands)).isEqualTo(dataFileCount);
+
+      // The workers ignore index entries of an older generation.
+      assertThat(commands).allMatch(cmd -> cmd.indexGeneration() > 
generationBeforeRestore);
+    }
+  }
+
   @Test
   void failsOnV2PosDeleteOnStagingBranch() throws Exception {
     Table table = createTableWithDelete(2);
@@ -1175,10 +1470,49 @@ class TestEqualityConvertPlanner extends 
OperatorTestBase {
         3);
   }
 
+  private static List<File> dataManifestFiles(Table table) {
+    return table.currentSnapshot().dataManifests(table.io()).stream()
+        .map(manifest -> new File(manifest.path().replace("file:", "")))
+        .collect(Collectors.toList());
+  }
+
+  // Makes reads of the files fail until restore is called.
+  private static void hide(List<File> files) {
+    assertThat(files.stream().map(file -> 
file.renameTo(hidden(file))).collect(Collectors.toList()))
+        .containsOnly(true);
+  }
+
+  private static void restore(List<File> files) {
+    assertThat(files.stream().map(file -> 
hidden(file).renameTo(file)).collect(Collectors.toList()))
+        .containsOnly(true);
+  }
+
+  private static File hidden(File file) {
+    return new File(file.getPath() + ".hidden");
+  }
+
   private static long countDataFileTasks(List<ReadCommand> commands) {
     return commands.stream().filter(c -> c.task() instanceof 
FileScanTask).count();
   }
 
+  // Keeps the operator state entries that the version before the stateVersion 
entry wrote.
+  private static OperatorSubtaskState stateOfOlderVersion(OperatorSubtaskState 
state) {
+    Set<String> olderStateNames =
+        Sets.newHashSet(
+            "indexSnapshotId", "indexGeneration", "eqFieldIds", 
"pendingStagingSnapshotId");
+    List<OperatorStateHandle> handles = Lists.newArrayList();
+    for (OperatorStateHandle handle : state.getManagedOperatorState()) {
+      Map<String, OperatorStateHandle.StateMetaInfo> stateMetaInfos =
+          Maps.newHashMap(handle.getStateNameToPartitionOffsets());
+      stateMetaInfos.keySet().retainAll(olderStateNames);
+      handles.add(new OperatorStreamStateHandle(stateMetaInfos, 
handle.getDelegateStateHandle()));
+    }
+
+    return OperatorSubtaskState.builder()
+        .setManagedOperatorState(new StateObjectCollection<>(handles))
+        .build();
+  }
+
   private static long countEqDeleteTasks(List<ReadCommand> commands) {
     return 
commands.stream().filter(TestEqualityConvertPlanner::isEqDelete).count();
   }