[Bug] Fix data evolution self-merge ABA across rollback snapshot lineage

paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java

@@ -25,5 +25,9 @@ public class ErrorMessages {
             "For Data Evolution table, multiple 'MERGE INTO' operations have encountered conflicts,"
                     + " updating the same file, which can render some updates ineffective.";
 
+    public static final String DATA_EVOLUTION_SNAPSHOT_LINEAGE_CONFLICT_MESSAGE =
+            "For Data Evolution table, the base snapshot lineage has changed, possibly due to a"
+                    + " rollback. Staged updates from the old snapshot lineage cannot be committed.";
+
     private ErrorMessages() {}
 }

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java

@@ -46,6 +46,9 @@ public interface FileStoreCommit extends AutoCloseable {
 
     FileStoreCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
 
+    FileStoreCommit rowIdCheckConflict(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid);
+
     FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction(
             @Nullable Long rowIdCheckFromSnapshot);
 

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -267,6 +267,13 @@ public FileStoreCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot)
         return this;
     }
 
+    @Override
+    public FileStoreCommit rowIdCheckConflict(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid) {
+        this.conflictDetection.setRowIdCheckFromSnapshot(rowIdCheckFromSnapshot, baseSnapshotUuid);
+        return this;
+    }
+
     @Override
     public FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction(
             @Nullable Long rowIdCheckFromSnapshot) {

paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java

@@ -164,6 +164,11 @@ public void setRowIdCheckFromSnapshot(@Nullable Long rowIdCheckFromSnapshot) {
         // Only Data Evolution tables support Row ID conflict detection.
     }
 
+    public void setRowIdCheckFromSnapshot(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid) {
+        // Only Data Evolution tables support Row ID conflict detection.
+    }
+
     public void setRowIdCheckFromSnapshotForMaterializeDvCompaction(
             @Nullable Long rowIdCheckFromSnapshot) {
         // Only Data Evolution tables support Row ID conflict detection.

paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java

@@ -75,6 +75,7 @@ public class DataEvolutionConflictDetection extends ConflictDetection {
     private final boolean nestedFieldEnabled;
 
     private @Nullable Long rowIdCheckFromSnapshot;
+    private @Nullable String baseSnapshotUuid;
     private @Nullable RowIdConflictCheckStrategy rowIdConflictCheckStrategy;
 
     public DataEvolutionConflictDetection(
@@ -105,20 +106,41 @@ public DataEvolutionConflictDetection(
 
     @Override
     public void setRowIdCheckFromSnapshot(@Nullable Long rowIdCheckFromSnapshot) {
+        String uuid = null;
+        if (rowIdCheckFromSnapshot != null) {
+            try {
+                Snapshot snapshot = snapshotManager.snapshot(rowIdCheckFromSnapshot);
+                uuid = snapshot.uuid();
+            } catch (RuntimeException e) {
+                // snapshot file missing, leave uuid as null
+            }
+        }
+        setRowIdCheckFromSnapshot(
+                rowIdCheckFromSnapshot, uuid, DataEvolutionDmlRowIdConflictCheck.INSTANCE);
+    }
+
+    @Override
+    public void setRowIdCheckFromSnapshot(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid) {
         setRowIdCheckFromSnapshot(
-                rowIdCheckFromSnapshot, DataEvolutionDmlRowIdConflictCheck.INSTANCE);
+                rowIdCheckFromSnapshot,
+                baseSnapshotUuid,
+                DataEvolutionDmlRowIdConflictCheck.INSTANCE);
     }
 
     @Override
     public void setRowIdCheckFromSnapshotForMaterializeDvCompaction(
             @Nullable Long rowIdCheckFromSnapshot) {
-        setRowIdCheckFromSnapshot(rowIdCheckFromSnapshot, MaterializeDvRowIdConflictCheck.INSTANCE);
+        setRowIdCheckFromSnapshot(
+                rowIdCheckFromSnapshot, null, MaterializeDvRowIdConflictCheck.INSTANCE);
     }
 
     private void setRowIdCheckFromSnapshot(
             @Nullable Long rowIdCheckFromSnapshot,
+            @Nullable String baseSnapshotUuid,
             RowIdConflictCheckStrategy conflictCheckStrategy) {
         this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
+        this.baseSnapshotUuid = baseSnapshotUuid;
         this.rowIdConflictCheckStrategy =
                 rowIdCheckFromSnapshot == null ? null : conflictCheckStrategy;
     }
@@ -354,20 +376,58 @@ private Optional<RuntimeException> checkForRowIdFromSnapshot(
             List<SimpleFileEntry> deltaEntries,
             List<IndexManifestEntry> deltaIndexEntries,
             @Nullable RowIdConflictChecker conflictChecker) {
-        if (rowIdCheckFromSnapshot == null
-                || conflictChecker == null
-                || conflictChecker.isEmpty()) {
+        if (rowIdCheckFromSnapshot == null) {
+            return Optional.empty();
+        }
+
+        // Run lineage validation BEFORE empty checker check so that DV-only and
+        // index-only commits are also protected against rollback/ABA.
+        // Fail closed when the latest snapshot ID is less than the base snapshot ID.
+        // This indicates a rollback has deleted newer snapshots, and the staged update
+        // is based on a snapshot lineage that no longer exists.
+        if (latestSnapshot.id() < rowIdCheckFromSnapshot) {
+            return Optional.of(
+                    new RuntimeException(
+                            ErrorMessages.DATA_EVOLUTION_SNAPSHOT_LINEAGE_CONFLICT_MESSAGE));
+        }
+
+        // Detect equal snapshot IDs with different snapshot UUIDs (ABA problem).
+        // A rollback can delete a snapshot and a new commit can reuse the same numeric ID.
+        // If the base snapshot UUID differs from the current snapshot UUID at that ID,
+        // the staged update is based on a different snapshot lineage.
+        // Invalidate cache before reading to avoid stale entries after rollback.
+        Snapshot baseSnapshot;
+        try {
+            snapshotManager.invalidateCache();
+            baseSnapshot = snapshotManager.snapshot(rowIdCheckFromSnapshot);
+        } catch (RuntimeException e) {
+            // snapshotManager.snapshot() throws RuntimeException when file is missing
+            // (e.g., snapshot was deleted by rollback or expiration).
+            return Optional.of(
+                    new RuntimeException(
+                            ErrorMessages.DATA_EVOLUTION_SNAPSHOT_LINEAGE_CONFLICT_MESSAGE));
+        }
+        if (baseSnapshotUuid != null && !baseSnapshotUuid.equals(baseSnapshot.uuid())) {
+            return Optional.of(
+                    new RuntimeException(
+                            ErrorMessages.DATA_EVOLUTION_SNAPSHOT_LINEAGE_CONFLICT_MESSAGE));
+        }
+
+        if (conflictChecker == null || conflictChecker.isEmpty()) {
             return Optional.empty();
         }
 
         List<BinaryRow> changedPartitions = changedPartitions(deltaEntries, deltaIndexEntries);
-        Long checkNextRowId = snapshotManager.snapshot(rowIdCheckFromSnapshot).nextRowId();
+        Long checkNextRowId = baseSnapshot.nextRowId();
         checkState(
                 checkNextRowId != null,
                 "Next row id cannot be null for snapshot %s.",
                 rowIdCheckFromSnapshot);
         for (long i = rowIdCheckFromSnapshot + 1; i <= latestSnapshot.id(); i++) {
             Snapshot snapshot = snapshotManager.snapshot(i);
+            if (snapshot == null) {
+                continue;
+            }
             if (snapshot.commitKind() == CommitKind.COMPACT) {
                 continue;
             }

paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java

@@ -40,6 +40,7 @@ public class BatchWriteBuilderImpl implements BatchWriteBuilder {
 
     private Map<String, String> staticPartition;
     private @Nullable Long rowIdCheckFromSnapshot = null;
+    private @Nullable String baseSnapshotUuid = null;
 
     public BatchWriteBuilderImpl(InnerTable table) {
         this.table = table;
@@ -77,7 +78,7 @@ public BatchTableCommit newCommit() {
         InnerTableCommit commit =
                 table.newCommit(commitUser)
                         .withOverwrite(staticPartition)
-                        .rowIdCheckConflict(rowIdCheckFromSnapshot);
+                        .rowIdCheckConflict(rowIdCheckFromSnapshot, baseSnapshotUuid);
         commit.ignoreEmptyCommit(
                 Options.fromMap(table.options())
                         .getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
@@ -86,7 +87,13 @@ public BatchTableCommit newCommit() {
     }
 
     public BatchWriteBuilderImpl rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot) {
+        return rowIdCheckConflict(rowIdCheckFromSnapshot, null);
+    }
+
+    public BatchWriteBuilderImpl rowIdCheckConflict(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid) {
         this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
+        this.baseSnapshotUuid = baseSnapshotUuid;
         return this;
     }
 }

paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java

@@ -58,6 +58,9 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
 
     InnerTableCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
 
+    InnerTableCommit rowIdCheckConflict(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid);
+
     InnerTableCommit rowIdCheckConflictForMaterializeDvCompaction(
             @Nullable Long rowIdCheckFromSnapshot);
 

paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java

@@ -181,6 +181,13 @@ public TableCommitImpl rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot)
         return this;
     }
 
+    @Override
+    public TableCommitImpl rowIdCheckConflict(
+            @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid) {
+        commit.rowIdCheckConflict(rowIdCheckFromSnapshot, baseSnapshotUuid);
+        return this;
+    }
+
     @Override
     public TableCommitImpl rowIdCheckConflictForMaterializeDvCompaction(
             @Nullable Long rowIdCheckFromSnapshot) {

paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.errors.ErrorMessages;
 import org.apache.paimon.index.DeletionVectorMeta;
 import org.apache.paimon.index.GlobalIndexMeta;
 import org.apache.paimon.index.IndexFileMeta;
@@ -1638,4 +1639,73 @@ private Snapshot snapshot(long id) {
                 null,
                 null);
     }
+
+    @Test
+    void testRowIdCheckConflictAbaDetectsRollback() {
+        CommitScanner scanner = mock(CommitScanner.class);
+        SnapshotManager snapshotManager = mock(SnapshotManager.class);
+        DataEvolutionConflictDetection detection =
+                (DataEvolutionConflictDetection)
+                        createConflictDetection(scanner, true, false, false, snapshotManager);
+
+        String baseUuid = "uuid-v1";
+        detection.setRowIdCheckFromSnapshot(1L, baseUuid);
+
+        Snapshot baseSnapshot = mock(Snapshot.class);
+        Snapshot latestSnapshot = mock(Snapshot.class);
+        when(baseSnapshot.uuid()).thenReturn("uuid-v2");
+        when(baseSnapshot.nextRowId()).thenReturn(100L);
+        when(latestSnapshot.id()).thenReturn(2L);
+        when(latestSnapshot.commitUser()).thenReturn("test-user");
+        when(snapshotManager.snapshot(1L)).thenReturn(baseSnapshot);
+
+        RowIdConflictChecker checker = mock(RowIdConflictChecker.class);
+        when(checker.isEmpty()).thenReturn(false);
+
+        Optional<RuntimeException> conflict =
+                detection.checkConflicts(
+                        latestSnapshot,
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        checker,
+                        Snapshot.CommitKind.APPEND);
+        assertThat(conflict).isPresent();
+        assertThat(conflict.get())
+                .hasMessageContaining(
+                        ErrorMessages.DATA_EVOLUTION_SNAPSHOT_LINEAGE_CONFLICT_MESSAGE);
+    }
+
+    @Test
+    void testRowIdCheckConflictNoAbaWhenUuidMatches() {
+        CommitScanner scanner = mock(CommitScanner.class);
+        SnapshotManager snapshotManager = mock(SnapshotManager.class);
+        DataEvolutionConflictDetection detection =
+                (DataEvolutionConflictDetection)
+                        createConflictDetection(scanner, true, false, false, snapshotManager);
+
+        String baseUuid = "uuid-v1";
+        detection.setRowIdCheckFromSnapshot(1L, baseUuid);
+
+        Snapshot baseSnapshot = mock(Snapshot.class);
+        Snapshot latestSnapshot = mock(Snapshot.class);
+        when(baseSnapshot.uuid()).thenReturn("uuid-v1");
+        when(baseSnapshot.nextRowId()).thenReturn(100L);
+        when(latestSnapshot.id()).thenReturn(2L);
+        when(latestSnapshot.commitUser()).thenReturn("test-user");
+        when(snapshotManager.snapshot(1L)).thenReturn(baseSnapshot);
+
+        RowIdConflictChecker checker = mock(RowIdConflictChecker.class);
+        when(checker.isEmpty()).thenReturn(false);
+
+        assertThat(
+                        detection.checkConflicts(
+                                latestSnapshot,
+                                Collections.emptyList(),
+                                Collections.emptyList(),
+                                Collections.emptyList(),
+                                checker,
+                                Snapshot.CommitKind.APPEND))
+                .isEmpty();
+    }
 }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.flink.action;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.flink.FlinkRowWrapper;
@@ -711,6 +712,8 @@ public DataStream<Committable> commit(
         FileStoreTable storeTable = (FileStoreTable) table;
         // copy to avoid serialization issue
         long baseSnapshotId = this.baseSnapshotId;
+        Snapshot baseSnapshot = ((FileStoreTable) table).snapshotManager().snapshot(baseSnapshotId);
+        String baseSnapshotUuid = baseSnapshot != null ? baseSnapshot.uuid() : null;
 
         // Check if some global-indexed columns are updated
         DataStream<Committable> checked =
@@ -731,7 +734,8 @@ public DataStream<Committable> commit(
                                         storeTable,
                                         storeTable
                                                 .newCommit(context.commitUser())
-                                                .rowIdCheckConflict(baseSnapshotId),
+                                                .rowIdCheckConflict(
+                                                        baseSnapshotId, baseSnapshotUuid),
                                         context),
                         new NoopCommittableStateManager());
 

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java

@@ -121,6 +121,8 @@ public DataStreamSink<?> sinkFrom(DataStream<Long> rowIds) {
                         .setParallelism(sinkParallelism);
 
         String commitUser = CoreOptions.createCommitUser(table.coreOptions().toConfiguration());
+        Snapshot baseSnapshot = table.snapshotManager().snapshot(baseSnapshotId);
+        String baseSnapshotUuid = baseSnapshot != null ? baseSnapshot.uuid() : null;
         CommitterOperatorFactory<Committable, ManifestCommittable> committerOperator =
                 new CommitterOperatorFactory<>(
                         false,
@@ -131,7 +133,8 @@ public DataStreamSink<?> sinkFrom(DataStream<Long> rowIds) {
                                         table,
                                         table.newCommit(context.commitUser())
                                                 .withOperation(Snapshot.Operation.DELETE)
-                                                .rowIdCheckConflict(baseSnapshotId),
+                                                .rowIdCheckConflict(
+                                                        baseSnapshotId, baseSnapshotUuid),
                                         context),
                         new NoopCommittableStateManager());
 

paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala

@@ -390,7 +390,7 @@ case class MergeIntoPaimonDataEvolutionTable(
         else Nil
 
       if (readSnapshot != null) {
-        writer.rowIdCheckConflict(readSnapshot.id())
+        writer.rowIdCheckConflict(readSnapshot.id(), readSnapshot.uuid())
       }
       DataEvolutionRowIdConflictCommitter.commit(
         sparkSession,

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala

@@ -390,7 +390,7 @@ case class MergeIntoPaimonDataEvolutionTable(
         else Nil
 
       if (readSnapshot != null) {
-        writer.rowIdCheckConflict(readSnapshot.id())
+        writer.rowIdCheckConflict(readSnapshot.id(), readSnapshot.uuid())
       }
       DataEvolutionRowIdConflictCommitter.commit(
         sparkSession,

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -451,7 +451,13 @@ case class PaimonSparkWriter(
   }
 
   def rowIdCheckConflict(rowIdCheckFromSnapshot: Long): Unit = {
-    writeBuilder.asInstanceOf[BatchWriteBuilderImpl].rowIdCheckConflict(rowIdCheckFromSnapshot)
+    rowIdCheckConflict(rowIdCheckFromSnapshot, null)
+  }
+
+  def rowIdCheckConflict(rowIdCheckFromSnapshot: Long, baseSnapshotUuid: String): Unit = {
+    writeBuilder
+      .asInstanceOf[BatchWriteBuilderImpl]
+      .rowIdCheckConflict(rowIdCheckFromSnapshot, baseSnapshotUuid)
   }
 
   def commit(commitMessages: Seq[CommitMessage]): Unit = {