From 248b87652c9533d034dfbb5a69e3e486ec62dbd7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=B8=87=E4=B9=89?= Date: Sun, 23 Aug 2026 14:34:29 +0800 Subject: [PATCH 1/5] [Bug] Fix data evolution self-merge ABA across rollback snapshot lineage Add snapshot UUID-based lineage validation to prevent staged row-ID partial updates from being applied to wrong snapshots after a rollback reuses the same numeric snapshot ID. Three-layer validation in checkForRowIdFromSnapshot: 1. Fail closed when latest snapshot ID < base snapshot ID (rollback deleted the update's base snapshot) 2. Detect missing base snapshot (race with concurrent cleanup) 3. ABA detection: compare base snapshot UUID with current snapshot UUID at the same ID (different lineage) The baseSnapshotUuid field is nullable for backward compatibility. Callers that don't pass UUID get existing behavior without the ABA protection. Closes #9352 --- .../apache/paimon/errors/ErrorMessages.java | 4 ++ .../paimon/operation/FileStoreCommit.java | 3 + .../paimon/operation/FileStoreCommitImpl.java | 7 ++ .../operation/commit/ConflictDetection.java | 5 ++ .../DataEvolutionConflictDetection.java | 44 +++++++++++- .../table/sink/BatchWriteBuilderImpl.java | 9 ++- .../paimon/table/sink/InnerTableCommit.java | 3 + .../paimon/table/sink/TableCommitImpl.java | 7 ++ .../commit/ConflictDetectionTest.java | 70 +++++++++++++++++++ .../action/DataEvolutionMergeIntoAction.java | 6 +- .../DataEvolutionDeleteSink.java | 5 +- 11 files changed, 157 insertions(+), 6 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java b/paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java index 8fd9012b1bb9..9d33437b0d46 100644 --- a/paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java +++ b/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() {} } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java index b039ffb9e9fc..f78cae84dfbc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java +++ b/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); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 0f60d6a3cb7c..f7ea0213395e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/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) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java index 1a202ac95eea..d5fea24aa08a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java +++ b/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. diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index 056c6b8dca86..5899f714c2d4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/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( @@ -106,19 +107,31 @@ public DataEvolutionConflictDetection( @Override public void setRowIdCheckFromSnapshot(@Nullable Long rowIdCheckFromSnapshot) { setRowIdCheckFromSnapshot( - rowIdCheckFromSnapshot, DataEvolutionDmlRowIdConflictCheck.INSTANCE); + rowIdCheckFromSnapshot, null, DataEvolutionDmlRowIdConflictCheck.INSTANCE); + } + + @Override + public void setRowIdCheckFromSnapshot( + @Nullable Long rowIdCheckFromSnapshot, @Nullable String baseSnapshotUuid) { + setRowIdCheckFromSnapshot( + 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; } @@ -360,8 +373,33 @@ private Optional checkForRowIdFromSnapshot( return Optional.empty(); } + // 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. + Snapshot baseSnapshot = snapshotManager.snapshot(rowIdCheckFromSnapshot); + if (baseSnapshot == null) { + 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)); + } + List 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.", diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java index d8c97405e2b0..da709eb366ef 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java @@ -40,6 +40,7 @@ public class BatchWriteBuilderImpl implements BatchWriteBuilder { private Map 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; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java index 43f98d0e7933..b7918974833d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java +++ b/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); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java index 014b5e64daa1..847e609c6caf 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java +++ b/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) { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java index 787d0761ea3e..4db3994e088a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java +++ b/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); + + assertThat( + detection.checkConflicts( + latestSnapshot, + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + checker, + Snapshot.CommitKind.APPEND)) + .isPresent() + .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(); + } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java index 4407ca3a8581..00cfc38d5fb3 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java +++ b/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 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 checked = @@ -731,7 +734,8 @@ public DataStream commit( storeTable, storeTable .newCommit(context.commitUser()) - .rowIdCheckConflict(baseSnapshotId), + .rowIdCheckConflict( + baseSnapshotId, baseSnapshotUuid), context), new NoopCommittableStateManager()); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java index 55d6814c739c..3e9cf126f9d8 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java @@ -121,6 +121,8 @@ public DataStreamSink sinkFrom(DataStream rowIds) { .setParallelism(sinkParallelism); String commitUser = CoreOptions.createCommitUser(table.coreOptions().toConfiguration()); + Snapshot baseSnapshot = table.snapshotManager().snapshot(baseSnapshotId); + String baseSnapshotUuid = baseSnapshot != null ? baseSnapshot.uuid() : null; CommitterOperatorFactory committerOperator = new CommitterOperatorFactory<>( false, @@ -131,7 +133,8 @@ public DataStreamSink sinkFrom(DataStream rowIds) { table, table.newCommit(context.commitUser()) .withOperation(Snapshot.Operation.DELETE) - .rowIdCheckConflict(baseSnapshotId), + .rowIdCheckConflict( + baseSnapshotId, baseSnapshotUuid), context), new NoopCommittableStateManager()); From e7ffb2939c57263fb3dadce615d8c1b5720aa45e Mon Sep 17 00:00:00 2001 From: zhang-arvin Date: Sun, 30 Aug 2026 08:47:44 +0800 Subject: [PATCH 2/5] [Bug] Fix Spark MERGE callers to pass snapshot UUID for ABA detection (#9352) - Update PaimonSparkWriter.rowIdCheckConflict to accept UUID parameter - Fix Spark MERGE (common + 4.0) to pass readSnapshot.uuid() - Fix ConflictDetectionTest compilation error: hasMessageContaining on OptionalAssert.get() chain Signed-off-by: zhang-arvin --- .../commit/ConflictDetectionTest.java | 20 +++++++++---------- .../MergeIntoPaimonDataEvolutionTable.scala | 2 +- .../MergeIntoPaimonDataEvolutionTable.scala | 2 +- .../spark/commands/PaimonSparkWriter.scala | 8 +++++++- 4 files changed, 19 insertions(+), 13 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java index 4db3994e088a..2485815bbd1d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java @@ -1662,16 +1662,16 @@ void testRowIdCheckConflictAbaDetectsRollback() { RowIdConflictChecker checker = mock(RowIdConflictChecker.class); when(checker.isEmpty()).thenReturn(false); - assertThat( - detection.checkConflicts( - latestSnapshot, - Collections.emptyList(), - Collections.emptyList(), - Collections.emptyList(), - checker, - Snapshot.CommitKind.APPEND)) - .isPresent() - .get() + Optional 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); } diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala index c6db56419be7..949302f1d32b 100644 --- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala +++ b/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, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala index f40db70fc120..50c5e1feea75 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala +++ b/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, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala index 91efc3d541b1..87e8b8c03042 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala +++ b/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 = { From 72e2a45c54987eb51587db7370721c2d46c2ba09 Mon Sep 17 00:00:00 2001 From: zhang-arvin Date: Sun, 30 Aug 2026 09:50:25 +0800 Subject: [PATCH 3/5] [Bug] Fix NPE when baseSnapshotUuid is null in ABA detection (#9352) Signed-off-by: zhang-arvin --- .../operation/commit/DataEvolutionConflictDetection.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index 5899f714c2d4..045fc36ec23f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -406,6 +406,9 @@ private Optional checkForRowIdFromSnapshot( 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; } From 5bc4079466c167ab5ec8139e1757672bccdb3a74 Mon Sep 17 00:00:00 2001 From: zhang-arvin Date: Thu, 3 Sep 2026 12:55:11 +0800 Subject: [PATCH 4/5] [Bug] Fix lineage validation gaps in DataEvolutionConflictDetection (#9363) Address JingsongLi's review feedback: 1. Move rollback/ABA lineage validation BEFORE empty RowIdConflictChecker check so that DV-only and index-only commits are also protected. 2. Replace dead null-check with try/catch for snapshotManager.snapshot() which throws RuntimeException when the snapshot file is missing. 3. Invalidate snapshot cache before reading base snapshot to avoid stale entries after rollback and ID reuse. All 52 ConflictDetectionTest tests pass. --- .../DataEvolutionConflictDetection.java | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index 045fc36ec23f..7108855d2c4d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -367,12 +367,12 @@ private Optional checkForRowIdFromSnapshot( List deltaEntries, List 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. @@ -386,8 +386,14 @@ private Optional checkForRowIdFromSnapshot( // 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. - Snapshot baseSnapshot = snapshotManager.snapshot(rowIdCheckFromSnapshot); - if (baseSnapshot == null) { + // 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)); @@ -398,6 +404,10 @@ private Optional checkForRowIdFromSnapshot( ErrorMessages.DATA_EVOLUTION_SNAPSHOT_LINEAGE_CONFLICT_MESSAGE)); } + if (conflictChecker == null || conflictChecker.isEmpty()) { + return Optional.empty(); + } + List changedPartitions = changedPartitions(deltaEntries, deltaIndexEntries); Long checkNextRowId = baseSnapshot.nextRowId(); checkState( From 34bd69269d34ba198712ebd42636cd3e883d35e3 Mon Sep 17 00:00:00 2001 From: zhang-arvin Date: Fri, 4 Sep 2026 22:01:57 +0800 Subject: [PATCH 5/5] [Bug] Capture snapshot UUID in single-arg setRowIdCheckFromSnapshot for ABA detection (#9363) The single-argument setRowIdCheckFromSnapshot(Long) was passing null for the base snapshot UUID, causing the ABA check in checkForRowIdFromSnapshot to never be invoked. This fix captures the UUID from the snapshot manager in the single-arg method so that even callers using the single-arg API benefit from ABA protection. --- .../commit/DataEvolutionConflictDetection.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index 7108855d2c4d..cd58095d6416 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -106,8 +106,17 @@ 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, null, DataEvolutionDmlRowIdConflictCheck.INSTANCE); + rowIdCheckFromSnapshot, uuid, DataEvolutionDmlRowIdConflictCheck.INSTANCE); } @Override