fix comments

Author: steFaiz

paimon-common/src/main/java/org/apache/paimon/utils/RowRangeIndex.java

@@ -43,14 +43,28 @@ private RowRangeIndex(List<Range> ranges) {
     }
 
     public static RowRangeIndex create(List<Range> ranges) {
+        return create(ranges, true);
+    }
+
+    public static RowRangeIndex create(List<Range> ranges, boolean mergeAdjacent) {
         checkArgument(ranges != null, "Ranges cannot be null");
-        return new RowRangeIndex(Range.sortAndMergeOverlap(ranges, true));
+        return new RowRangeIndex(Range.sortAndMergeOverlap(ranges, mergeAdjacent));
     }
 
     public List<Range> ranges() {
         return Collections.unmodifiableList(ranges);
     }
 
+    public boolean contains(long start, long end) {
+        int candidate = lowerBound(ends, start);
+        return candidate < starts.length && starts[candidate] <= start && ends[candidate] >= end;
+    }
+
+    public boolean containsExactly(long start, long end) {
+        int candidate = lowerBound(starts, start);
+        return candidate < starts.length && starts[candidate] == start && ends[candidate] == end;
+    }
+
     public boolean intersects(long start, long end) {
         int candidate = lowerBound(ends, start);
         return candidate < starts.length && starts[candidate] <= end;

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

@@ -37,7 +37,9 @@
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
 import org.apache.paimon.utils.RangeHelper;
+import org.apache.paimon.utils.RowRangeIndex;
 import org.apache.paimon.utils.SnapshotManager;
 
 import org.slf4j.Logger;
@@ -54,12 +56,9 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.NavigableMap;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.TreeMap;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -569,20 +568,22 @@ Optional<RuntimeException> checkRowIdExistence(
             return Optional.empty();
         }
 
-        NavigableMap<Long, Long> existingRanges = new TreeMap<>();
-        for (SimpleFileEntry base : baseEntries) {
-            if (base.firstRowId() != null && !dedicatedStorageFile(base.fileName())) {
-                existingRanges.put(base.firstRowId(), base.rowCount());
-            }
-        }
+        List<Range> existingRanges =
+                baseEntries.stream()
+                        .filter(
+                                base ->
+                                        base.firstRowId() != null
+                                                && !dedicatedStorageFile(base.fileName()))
+                        .map(SimpleFileEntry::nonNullRowIdRange)
+                        .collect(Collectors.toList());
+        RowRangeIndex existingIndex = RowRangeIndex.create(existingRanges, false);
 
         for (SimpleFileEntry entry : filesToCheck) {
+            long rowRangeEnd = entry.firstRowId() + entry.rowCount() - 1;
             boolean exists =
                     dedicatedStorageFile(entry.fileName())
-                            ? rowIdRangeCovered(
-                                    existingRanges, entry.firstRowId(), entry.rowCount())
-                            : rowIdRangeExists(
-                                    existingRanges, entry.firstRowId(), entry.rowCount());
+                            ? existingIndex.contains(entry.firstRowId(), rowRangeEnd)
+                            : existingIndex.containsExactly(entry.firstRowId(), rowRangeEnd);
             if (!exists) {
                 return Optional.of(
                         new RuntimeException(
@@ -601,20 +602,6 @@ Optional<RuntimeException> checkRowIdExistence(
         return Optional.empty();
     }
 
-    private static boolean rowIdRangeCovered(
-            NavigableMap<Long, Long> ranges, long firstRowId, long rowCount) {
-        Entry<Long, Long> range = ranges.floorEntry(firstRowId);
-        return range != null
-                && range.getKey() <= firstRowId
-                && range.getKey() + range.getValue() >= firstRowId + rowCount;
-    }
-
-    private static boolean rowIdRangeExists(
-            NavigableMap<Long, Long> ranges, long firstRowId, long rowCount) {
-        Long existingRowCount = ranges.get(firstRowId);
-        return existingRowCount != null && existingRowCount == rowCount;
-    }
-
     private static boolean dedicatedStorageFile(String fileName) {
         return isBlobFile(fileName) || isVectorStoreFile(fileName);
     }

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

@@ -419,6 +419,23 @@ void testCheckRowIdExistenceBaseFileRewritten() {
         assertThat(result.get().getMessage()).contains("Row ID existence conflict");
     }
 
+    @Test
+    void testCheckRowIdExistenceNormalFileRejectsAdjacentDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
+        baseEntries.add(createFileEntryWithRowId("f2", ADD, 2L, 2L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 4L));
+
+        Optional<RuntimeException> result =
+                detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
+        assertThat(result).isPresent();
+        assertThat(result.get().getMessage()).contains("Row ID existence conflict");
+    }
+
     @Test
     void testCheckRowIdExistenceDedicatedFileCoveredByDataFiles() {
         ConflictDetection detection = createConflictDetection();
@@ -432,6 +449,23 @@ void testCheckRowIdExistenceDedicatedFileCoveredByDataFiles() {
         assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries, 4L)).isEmpty();
     }
 
+    @Test
+    void testCheckRowIdExistenceDedicatedFileRejectsAdjacentDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
+        baseEntries.add(createFileEntryWithRowId("f2", ADD, 2L, 2L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 4L));
+
+        Optional<RuntimeException> result =
+                detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
+        assertThat(result).isPresent();
+        assertThat(result.get().getMessage()).contains("Row ID existence conflict");
+    }
+
     @Test
     void testCheckRowIdExistenceDedicatedFileRejectsRangeNotCoveredByOneDataFile() {
         ConflictDetection detection = createConflictDetection();