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