fix

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -275,18 +275,20 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
         if (snapshotCommit == null) {
             snapshotCommit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
         }
-        ConflictDetection conflictDetection =
-                new ConflictDetection(
-                        tableName,
-                        commitUser,
-                        partitionType,
-                        pathFactory(),
-                        newKeyComparator(),
-                        bucketMode(),
-                        options.deletionVectorsEnabled(),
-                        options.dataEvolutionEnabled(),
-                        newIndexFileHandler(),
-                        snapshotManager);
+        ConflictDetection.Factory conflictDetectFactory =
+                scanner ->
+                        new ConflictDetection(
+                                tableName,
+                                commitUser,
+                                partitionType,
+                                pathFactory(),
+                                newKeyComparator(),
+                                bucketMode(),
+                                options.deletionVectorsEnabled(),
+                                options.dataEvolutionEnabled(),
+                                newIndexFileHandler(),
+                                snapshotManager,
+                                scanner);
         StrictModeChecker strictModeChecker =
                 StrictModeChecker.create(
                         snapshotManager,
@@ -329,7 +331,7 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
                 options.commitMaxRetryWait(),
                 options.rowTrackingEnabled(),
                 options.commitDiscardDuplicateFiles(),
-                conflictDetection,
+                conflictDetectFactory,
                 strictModeChecker,
                 rollback);
     }

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

@@ -197,7 +197,7 @@ public FileStoreCommitImpl(
             long commitMaxRetryWait,
             boolean rowTrackingEnabled,
             boolean discardDuplicateFiles,
-            ConflictDetection conflictDetection,
+            ConflictDetection.Factory conflictDetectFactory,
             @Nullable StrictModeChecker strictModeChecker,
             @Nullable CommitRollback rollback) {
         this.snapshotCommit = snapshotCommit;
@@ -239,8 +239,7 @@ public FileStoreCommitImpl(
         this.rowTrackingEnabled = rowTrackingEnabled;
         this.discardDuplicateFiles = discardDuplicateFiles;
         this.strictModeChecker = strictModeChecker;
-        conflictDetection.setCommitScanner(scanner);
-        this.conflictDetection = conflictDetection;
+        this.conflictDetection = conflictDetectFactory.create(scanner);
         this.commitCleaner = new CommitCleaner(manifestList, manifestFile, indexManifestFile);
     }
 

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

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.index.DeletionVectorMeta;
@@ -82,8 +83,8 @@ public class ConflictDetection {
     private final boolean dataEvolutionEnabled;
     private final IndexFileHandler indexFileHandler;
     private final SnapshotManager snapshotManager;
+    private final CommitScanner commitScanner;
 
-    private @Nullable CommitScanner scanner;
     private @Nullable PartitionExpire partitionExpire;
     private @Nullable Long rowIdCheckFromSnapshot = null;
 
@@ -97,7 +98,8 @@ public ConflictDetection(
             boolean deletionVectorsEnabled,
             boolean dataEvolutionEnabled,
             IndexFileHandler indexFileHandler,
-            SnapshotManager snapshotManager) {
+            SnapshotManager snapshotManager,
+            CommitScanner commitScanner) {
         this.tableName = tableName;
         this.commitUser = commitUser;
         this.partitionType = partitionType;
@@ -108,16 +110,13 @@ public ConflictDetection(
         this.dataEvolutionEnabled = dataEvolutionEnabled;
         this.indexFileHandler = indexFileHandler;
         this.snapshotManager = snapshotManager;
+        this.commitScanner = commitScanner;
     }
 
     public void setRowIdCheckFromSnapshot(@Nullable Long rowIdCheckFromSnapshot) {
         this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
     }
 
-    public void setCommitScanner(CommitScanner scanner) {
-        this.scanner = scanner;
-    }
-
     @Nullable
     public Comparator<InternalRow> keyComparator() {
         return keyComparator;
@@ -378,7 +377,8 @@ private Optional<RuntimeException> checkConflictForPartitionExpire(
         return Optional.empty();
     }
 
-    private Optional<RuntimeException> checkForRowIdFromSnapshot(
+    @VisibleForTesting
+    Optional<RuntimeException> checkForRowIdFromSnapshot(
             Snapshot latestSnapshot,
             List<SimpleFileEntry> deltaEntries,
             List<IndexManifestEntry> deltaIndexEntries) {
@@ -401,14 +401,18 @@ private Optional<RuntimeException> checkForRowIdFromSnapshot(
         }
 
         // check history row id ranges
-        long checkNextRowId = snapshotManager.snapshot(rowIdCheckFromSnapshot).nextRowId();
+        Long checkNextRowId = snapshotManager.snapshot(rowIdCheckFromSnapshot).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.commitKind() == CommitKind.COMPACT) {
                 continue;
             }
             List<ManifestEntry> changes =
-                    scanner.readIncrementalEntries(snapshot, changedPartitions);
+                    commitScanner.readIncrementalEntries(snapshot, changedPartitions);
             for (ManifestEntry entry : changes) {
                 DataFileMeta file = entry.file();
                 long firstRowId = file.nonNullFirstRowId();
@@ -429,7 +433,8 @@ private Optional<RuntimeException> checkForRowIdFromSnapshot(
         return Optional.empty();
     }
 
-    private Optional<RuntimeException> checkRowIdRangeConflicts(
+    @VisibleForTesting
+    Optional<RuntimeException> checkRowIdRangeConflicts(
             CommitKind commitKind, Collection<SimpleFileEntry> mergedEntries) {
         if (!dataEvolutionEnabled) {
             return Optional.empty();
@@ -694,4 +699,9 @@ public int hashCode() {
             return Objects.hash(partition, bucket, level);
         }
     }
+
+    /** Factory to create {@link ConflictDetection}. */
+    public interface Factory {
+        ConflictDetection create(CommitScanner scanner);
+    }
 }