fix

Author: JingsongLi

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

@@ -149,7 +149,7 @@ public Optional<RuntimeException> checkConflicts(
             List<IndexManifestEntry> deltaIndexEntries,
             CommitKind commitKind) {
         String baseCommitUser = latestSnapshot.commitUser();
-        if (checkForDeletionVector()) {
+        if (deletionVectorsEnabled && bucketMode.equals(BucketMode.BUCKET_UNAWARE)) {
             // Enrich dvName in fileEntry to checker for base ADD dv and delta DELETE dv.
             // For example:
             // If the base file is <ADD baseFile1, ADD dv1>,
@@ -210,23 +210,12 @@ public Optional<RuntimeException> checkConflicts(
             return exception;
         }
 
-        if (rowIdCheckFromSnapshot != null || commitKind == CommitKind.COMPACT) {
-            exception = assertRowIdRangeConflicts(mergedEntries);
-            if (exception.isPresent()) {
-                return exception;
-            }
-        }
-
-        if (rowIdCheckFromSnapshot != null) {
-            exception =
-                    assertForRowIdCheckFromSnapshot(
-                            latestSnapshot, deltaEntries, deltaIndexEntries);
-            if (exception.isPresent()) {
-                return exception;
-            }
+        exception = checkRowIdRangeConflicts(commitKind, mergedEntries);
+        if (exception.isPresent()) {
+            return exception;
         }
 
-        return Optional.empty();
+        return checkForRowIdFromSnapshot(latestSnapshot, deltaEntries, deltaIndexEntries);
     }
 
     private Optional<RuntimeException> checkBucketKeepSame(
@@ -340,10 +329,6 @@ private Function<Throwable, RuntimeException> conflictException(
         };
     }
 
-    private boolean checkForDeletionVector() {
-        return deletionVectorsEnabled && bucketMode.equals(BucketMode.BUCKET_UNAWARE);
-    }
-
     private Optional<RuntimeException> checkDeleteInEntries(
             Collection<SimpleFileEntry> mergedEntries,
             Function<Throwable, RuntimeException> exceptionFunction) {
@@ -356,7 +341,7 @@ private Optional<RuntimeException> checkDeleteInEntries(
                         tableName);
             }
         } catch (Throwable e) {
-            Optional<RuntimeException> exception = assertConflictForPartitionExpire(mergedEntries);
+            Optional<RuntimeException> exception = checkConflictForPartitionExpire(mergedEntries);
             if (exception.isPresent()) {
                 return exception;
             }
@@ -365,7 +350,7 @@ private Optional<RuntimeException> checkDeleteInEntries(
         return Optional.empty();
     }
 
-    private Optional<RuntimeException> assertConflictForPartitionExpire(
+    private Optional<RuntimeException> checkConflictForPartitionExpire(
             Collection<SimpleFileEntry> mergedEntries) {
         if (partitionExpire != null && partitionExpire.isValueExpiration()) {
             Set<BinaryRow> deletedPartitions = new HashSet<>();
@@ -393,13 +378,16 @@ private Optional<RuntimeException> assertConflictForPartitionExpire(
         return Optional.empty();
     }
 
-    private Optional<RuntimeException> assertForRowIdCheckFromSnapshot(
+    private Optional<RuntimeException> checkForRowIdFromSnapshot(
             Snapshot latestSnapshot,
             List<SimpleFileEntry> deltaEntries,
             List<IndexManifestEntry> deltaIndexEntries) {
         if (!dataEvolutionEnabled) {
             return Optional.empty();
         }
+        if (rowIdCheckFromSnapshot == null) {
+            return Optional.empty();
+        }
 
         List<BinaryRow> changedPartitions = changedPartitions(deltaEntries, deltaIndexEntries);
         // collect history row id ranges
@@ -441,11 +429,14 @@ private Optional<RuntimeException> assertForRowIdCheckFromSnapshot(
         return Optional.empty();
     }
 
-    private Optional<RuntimeException> assertRowIdRangeConflicts(
-            Collection<SimpleFileEntry> mergedEntries) {
+    private Optional<RuntimeException> checkRowIdRangeConflicts(
+            CommitKind commitKind, Collection<SimpleFileEntry> mergedEntries) {
         if (!dataEvolutionEnabled) {
             return Optional.empty();
         }
+        if (rowIdCheckFromSnapshot == null && commitKind != CommitKind.COMPACT) {
+            return Optional.empty();
+        }
 
         List<SimpleFileEntry> entries =
                 mergedEntries.stream()