[core][python] Improve data evolution row-id conflict diagnostics

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

@@ -488,25 +488,94 @@ private Optional<RuntimeException> checkRowIdRangeConflicts(
 
         RangeHelper<SimpleFileEntry> rangeHelper =
                 new RangeHelper<>(SimpleFileEntry::nonNullRowIdRange);
-        List<List<SimpleFileEntry>> merged = rangeHelper.mergeOverlappingRanges(entries);
-        for (List<SimpleFileEntry> group : merged) {
-            List<SimpleFileEntry> dataFiles = new ArrayList<>();
-            for (SimpleFileEntry f : group) {
-                if (!dedicatedStorageFile(f.fileName())) {
-                    dataFiles.add(f);
-                }
-            }
-            if (!rangeHelper.areAllRangesSame(dataFiles)) {
+        List<SimpleFileEntry> dataFiles =
+                entries.stream()
+                        .filter(file -> !dedicatedStorageFile(file.fileName()))
+                        .collect(Collectors.toList());
+
+        Optional<RuntimeException> exception =
+                checkDataFileRowIdRangeConflicts(rangeHelper, dataFiles);
+        if (exception.isPresent()) {
+            return exception;
+        }
+
+        List<SimpleFileEntry> dedicatedFiles =
+                entries.stream()
+                        .filter(file -> dedicatedStorageFile(file.fileName()))
+                        .collect(Collectors.toList());
+        exception = checkDedicatedFileRowIdRangeConflicts(dataFiles, dedicatedFiles);
+        if (exception.isPresent()) {
+            return exception;
+        }
+
+        return Optional.empty();
+    }
+
+    private Optional<RuntimeException> checkDataFileRowIdRangeConflicts(
+            RangeHelper<SimpleFileEntry> rangeHelper, List<SimpleFileEntry> dataFiles) {
+        for (List<SimpleFileEntry> dataFileGroup : rangeHelper.mergeOverlappingRanges(dataFiles)) {
+            if (!rangeHelper.areAllRangesSame(dataFileGroup)) {
                 return Optional.of(
                         new RuntimeException(
-                                "For Data Evolution table, multiple 'MERGE INTO' and 'COMPACT' operations "
+                                "For Data Evolution table, multiple 'MERGE INTO' and 'COMPACT' "
+                                        + "operations "
                                         + "have encountered conflicts, data files: "
-                                        + dataFiles));
+                                        + dataFileGroup));
             }
         }
+
+        return Optional.empty();
+    }
+
+    private Optional<RuntimeException> checkDedicatedFileRowIdRangeConflicts(
+            List<SimpleFileEntry> dataFiles, List<SimpleFileEntry> dedicatedFiles) {
+        if (dedicatedFiles.isEmpty()) {
+            return Optional.empty();
+        }
+
+        RowRangeIndex dataFileRowRangeIndex = rowRangeIndex(dataFiles, false);
+
+        for (SimpleFileEntry dedicatedFile : dedicatedFiles) {
+            Range dedicatedRange = dedicatedFile.nonNullRowIdRange();
+            if (dataFileRowRangeIndex.contains(dedicatedRange)) {
+                continue;
+            }
+
+            List<Range> intersectingRanges =
+                    dataFileRowRangeIndex.intersectedRanges(dedicatedRange.from, dedicatedRange.to);
+            List<SimpleFileEntry> intersectingDataFiles =
+                    dataFiles.stream()
+                            .filter(
+                                    dataFile ->
+                                            dataFile.nonNullRowIdRange()
+                                                    .hasIntersection(dedicatedRange))
+                            .collect(Collectors.toList());
+            String conflictReason =
+                    intersectingRanges.size() > 1
+                            ? "spans multiple data file ranges"
+                            : "is not covered by one data file range";
+            return Optional.of(
+                    new RuntimeException(
+                            String.format(
+                                    "For Data Evolution table, multiple 'MERGE INTO' and 'COMPACT' "
+                                            + "operations have encountered conflicts, dedicated "
+                                            + "file %s %s %s: %s",
+                                    dedicatedFile,
+                                    dedicatedRange,
+                                    conflictReason,
+                                    intersectingDataFiles)));
+        }
+
         return Optional.empty();
     }
 
+    private static RowRangeIndex rowRangeIndex(
+            Collection<SimpleFileEntry> files, boolean mergeAdjacent) {
+        return RowRangeIndex.create(
+                files.stream().map(SimpleFileEntry::nonNullRowIdRange).collect(Collectors.toList()),
+                mergeAdjacent);
+    }
+
     private Optional<RuntimeException> checkForRowIdFromSnapshot(
             Snapshot latestSnapshot,
             List<SimpleFileEntry> deltaEntries,
@@ -640,15 +709,14 @@ Optional<RuntimeException> checkRowIdExistence(
             return Optional.empty();
         }
 
-        List<Range> existingRanges =
+        List<SimpleFileEntry> existingDataFiles =
                 baseEntries.stream()
                         .filter(
                                 base ->
                                         base.firstRowId() != null
                                                 && !dedicatedStorageFile(base.fileName()))
-                        .map(SimpleFileEntry::nonNullRowIdRange)
                         .collect(Collectors.toList());
-        RowRangeIndex existingIndex = RowRangeIndex.create(existingRanges, false);
+        RowRangeIndex existingIndex = rowRangeIndex(existingDataFiles, false);
 
         for (SimpleFileEntry entry : filesToCheck) {
             Range rowRange = entry.nonNullRowIdRange();

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

@@ -585,6 +585,64 @@ void testCheckRowIdExistenceSkipsWhenNextRowIdNull() {
         assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries, null)).isEmpty();
     }
 
+    @Test
+    void testCheckRowIdRangeConflictsReportsDedicatedFileSpanningDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        Optional<RuntimeException> exception =
+                detection.checkConflicts(
+                        snapshot(1),
+                        Arrays.asList(
+                                createFileEntryWithRowId("f1", ADD, 0L, 2L),
+                                createFileEntryWithRowId("f2", ADD, 2L, 2L)),
+                        Collections.singletonList(createFileEntryWithRowId("p1.blob", ADD, 0L, 4L)),
+                        Collections.emptyList(),
+                        null,
+                        Snapshot.CommitKind.COMPACT);
+
+        assertThat(exception).isPresent();
+        assertThat(exception.get())
+                .hasMessageContaining("dedicated file")
+                .hasMessageContaining("p1.blob")
+                .hasMessageContaining("spans multiple data file ranges")
+                .hasMessageContaining("f1")
+                .hasMessageContaining("f2");
+    }
+
+    @Test
+    void testCheckRowIdRangeConflictsAllowsAdjacentDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        Optional<RuntimeException> exception =
+                detection.checkConflicts(
+                        snapshot(1),
+                        Arrays.asList(
+                                createFileEntryWithRowId("f1", ADD, 0L, 2L),
+                                createFileEntryWithRowId("f2", ADD, 2L, 2L)),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        null,
+                        Snapshot.CommitKind.COMPACT);
+
+        assertThat(exception).isEmpty();
+    }
+
+    @Test
+    void testCheckRowIdRangeConflictsAllowsDedicatedFileCoveredByOneDataFile() {
+        ConflictDetection detection = createConflictDetection();
+
+        Optional<RuntimeException> exception =
+                detection.checkConflicts(
+                        snapshot(1),
+                        Collections.singletonList(createFileEntryWithRowId("f1", ADD, 0L, 4L)),
+                        Collections.singletonList(createFileEntryWithRowId("p1.blob", ADD, 1L, 2L)),
+                        Collections.emptyList(),
+                        null,
+                        Snapshot.CommitKind.COMPACT);
+
+        assertThat(exception).isEmpty();
+    }
+
     private SimpleFileEntry createFileEntryWithRowId(
             String fileName, FileKind kind, long firstRowId, long rowCount) {
         return createFileEntryWithRowId(fileName, kind, EMPTY_ROW, 0, firstRowId, rowCount);

paimon-python/pypaimon/tests/write/conflict_detection_test.py

@@ -137,6 +137,13 @@ def test_no_conflict_when_blob_file_range_is_covered(self):
         self.assertIsNone(
             detection.check_row_id_existence(base, delta, next_row_id=200))
 
+    def test_no_conflict_when_vector_file_range_is_covered(self):
+        detection = self._make_detection()
+        base = [_make_entry("f1", kind=0, first_row_id=0, row_count=100)]
+        delta = [_make_entry("p1.vector.0", kind=0, first_row_id=20, row_count=10)]
+        self.assertIsNone(
+            detection.check_row_id_existence(base, delta, next_row_id=200))
+
     def test_conflict_when_blob_file_range_is_not_covered(self):
         detection = self._make_detection()
         base = [_make_entry("f1", kind=0, first_row_id=0, row_count=100)]
@@ -208,6 +215,57 @@ def test_skip_when_next_row_id_is_none(self):
             detection.check_row_id_existence(base, delta, next_row_id=None))
 
 
+class TestCheckRowIdRangeConflicts(unittest.TestCase):
+
+    def _make_detection(self):
+        return ConflictDetection(
+            data_evolution_enabled=True,
+            snapshot_manager=None,
+            manifest_list_manager=None,
+            table=None,
+            commit_scanner=None,
+        )
+
+    def test_reports_dedicated_file_spanning_data_files(self):
+        detection = self._make_detection()
+        entries = [
+            _make_entry("f1", kind=0, first_row_id=0, row_count=2),
+            _make_entry("f2", kind=0, first_row_id=2, row_count=2),
+            _make_entry("p1.blob", kind=0, first_row_id=0, row_count=4),
+        ]
+
+        result = detection.check_row_id_range_conflicts("COMPACT", entries)
+
+        self.assertIsNotNone(result)
+        self.assertIn("dedicated file", str(result))
+        self.assertIn("p1.blob", str(result))
+        self.assertIn("spans multiple data file ranges", str(result))
+        self.assertIn("f1", str(result))
+        self.assertIn("f2", str(result))
+
+    def test_allows_adjacent_data_files(self):
+        detection = self._make_detection()
+        entries = [
+            _make_entry("f1", kind=0, first_row_id=0, row_count=2),
+            _make_entry("f2", kind=0, first_row_id=2, row_count=2),
+        ]
+
+        result = detection.check_row_id_range_conflicts("COMPACT", entries)
+
+        self.assertIsNone(result)
+
+    def test_allows_dedicated_file_covered_by_one_data_file(self):
+        detection = self._make_detection()
+        entries = [
+            _make_entry("f1", kind=0, first_row_id=0, row_count=4),
+            _make_entry("p1.blob", kind=0, first_row_id=1, row_count=2),
+        ]
+
+        result = detection.check_row_id_range_conflicts("COMPACT", entries)
+
+        self.assertIsNone(result)
+
+
 class TestOverwriteConflictDetection(unittest.TestCase):
 
     def _make_detection(self):

paimon-python/pypaimon/write/commit/conflict_detection.py

@@ -406,7 +406,7 @@ def check_row_id_existence(self, base_entries, delta_entries, next_row_id=None):
                 existing_index.add((
                     base.partition, base.bucket,
                     base.file.first_row_id, base.file.row_count))
-                if not DataFileMeta.is_blob_file(base.file.file_name):
+                if not self._is_dedicated_file(base.file.file_name):
                     existing_ranges.setdefault((base.partition, base.bucket), []).append(
                         base.file.row_id_range())
 
@@ -416,7 +416,7 @@ def check_row_id_existence(self, base_entries, delta_entries, next_row_id=None):
         }
 
         for entry in files_to_check:
-            if DataFileMeta.is_blob_file(entry.file.file_name):
+            if self._is_dedicated_file(entry.file.file_name):
                 base_ranges = existing_ranges.get((entry.partition, entry.bucket), [])
                 if not entry.file.row_id_range().exclude(base_ranges):
                     continue
@@ -452,29 +452,101 @@ def check_row_id_range_conflicts(self, commit_kind, commit_entries):
             return None
 
         range_helper = RangeHelper(lambda entry: entry.file.row_id_range())
-        merged_groups = range_helper.merge_overlapping_ranges(entries_with_row_id)
+        data_files = [
+            entry for entry in entries_with_row_id
+            if not self._is_dedicated_file(entry.file.file_name)
+        ]
 
-        for group in merged_groups:
-            data_files = [
-                entry for entry in group
-                if not DataFileMeta.is_blob_file(entry.file.file_name)
-            ]
-            if not range_helper.are_all_ranges_same(data_files):
+        conflict = self._check_data_file_row_id_range_conflicts(
+            range_helper, data_files)
+        if conflict is not None:
+            return conflict
+
+        dedicated_files = [
+            entry for entry in entries_with_row_id
+            if self._is_dedicated_file(entry.file.file_name)
+        ]
+        conflict = self._check_dedicated_file_row_id_range_conflicts(
+            data_files, dedicated_files)
+        if conflict is not None:
+            return conflict
+
+        return None
+
+    def _check_data_file_row_id_range_conflicts(self, range_helper, data_files):
+        for data_file_group in range_helper.merge_overlapping_ranges(data_files):
+            if not range_helper.are_all_ranges_same(data_file_group):
                 file_descriptions = [
-                    "{name}(rowId={row_id}, count={count})".format(
-                        name=entry.file.file_name,
-                        row_id=entry.file.first_row_id,
-                        count=entry.file.row_count,
-                    )
-                    for entry in data_files
+                    self._file_description(entry) for entry in data_file_group
                 ]
                 return RuntimeError(
                     "For Data Evolution table, multiple 'MERGE INTO' and 'COMPACT' "
                     "operations have encountered conflicts, data files: "
                     + str(file_descriptions))
+        return None
+
+    def _check_dedicated_file_row_id_range_conflicts(
+            self, data_files, dedicated_files):
+        if not dedicated_files:
+            return None
+
+        data_ranges = self._data_file_row_ranges(data_files)
+
+        for dedicated_file in dedicated_files:
+            dedicated_range = dedicated_file.file.row_id_range()
+            if any(self._contains(row_range, dedicated_range) for row_range in data_ranges):
+                continue
+
+            intersecting_ranges = [
+                row_range for row_range in data_ranges
+                if row_range.overlaps(dedicated_range)
+            ]
+            intersecting_files = [
+                self._file_description(entry)
+                for entry in data_files
+                if entry.file.row_id_range().overlaps(dedicated_range)
+            ]
+            conflict_reason = (
+                "spans multiple data file ranges"
+                if len(intersecting_ranges) > 1
+                else "is not covered by one data file range"
+            )
+            return RuntimeError(
+                "For Data Evolution table, multiple 'MERGE INTO' and 'COMPACT' "
+                "operations have encountered conflicts, dedicated file "
+                "{file} {row_range} {reason}: {groups}".format(
+                    file=self._file_description(dedicated_file),
+                    row_range=dedicated_range,
+                    reason=conflict_reason,
+                    groups=intersecting_files))
 
         return None
 
+    @staticmethod
+    def _data_file_row_ranges(data_files):
+        return Range.sort_and_merge_overlap(
+            [entry.file.row_id_range() for entry in data_files],
+            True,
+            False,
+        )
+
+    @staticmethod
+    def _contains(container, row_range):
+        return container.from_ <= row_range.from_ and container.to >= row_range.to
+
+    @staticmethod
+    def _is_dedicated_file(file_name):
+        return (DataFileMeta.is_blob_file(file_name)
+                or DataFileMeta.is_vector_file(file_name))
+
+    @staticmethod
+    def _file_description(entry):
+        return "{name}(rowId={row_id}, count={count})".format(
+            name=entry.file.file_name,
+            row_id=entry.file.first_row_id,
+            count=entry.file.row_count,
+        )
+
     def check_row_id_from_snapshot(self, latest_snapshot, commit_entries):
         if not self.data_evolution_enabled:
             return None