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 fcf00c792b15..a679b9d0914d 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 @@ -488,25 +488,94 @@ private Optional checkRowIdRangeConflicts( RangeHelper rangeHelper = new RangeHelper<>(SimpleFileEntry::nonNullRowIdRange); - List> merged = rangeHelper.mergeOverlappingRanges(entries); - for (List group : merged) { - List dataFiles = new ArrayList<>(); - for (SimpleFileEntry f : group) { - if (!dedicatedStorageFile(f.fileName())) { - dataFiles.add(f); - } - } - if (!rangeHelper.areAllRangesSame(dataFiles)) { + List dataFiles = + entries.stream() + .filter(file -> !dedicatedStorageFile(file.fileName())) + .collect(Collectors.toList()); + + Optional exception = + checkDataFileRowIdRangeConflicts(rangeHelper, dataFiles); + if (exception.isPresent()) { + return exception; + } + + List 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 checkDataFileRowIdRangeConflicts( + RangeHelper rangeHelper, List dataFiles) { + for (List 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 checkDedicatedFileRowIdRangeConflicts( + List dataFiles, List 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 intersectingRanges = + dataFileRowRangeIndex.intersectedRanges(dedicatedRange.from, dedicatedRange.to); + List 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 files, boolean mergeAdjacent) { + return RowRangeIndex.create( + files.stream().map(SimpleFileEntry::nonNullRowIdRange).collect(Collectors.toList()), + mergeAdjacent); + } + private Optional checkForRowIdFromSnapshot( Snapshot latestSnapshot, List deltaEntries, @@ -640,15 +709,14 @@ Optional checkRowIdExistence( return Optional.empty(); } - List existingRanges = + List 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(); 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 0695406bde6c..5be15f0dcc1e 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 @@ -585,6 +585,64 @@ void testCheckRowIdExistenceSkipsWhenNextRowIdNull() { assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries, null)).isEmpty(); } + @Test + void testCheckRowIdRangeConflictsReportsDedicatedFileSpanningDataFiles() { + ConflictDetection detection = createConflictDetection(); + + Optional 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 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 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); diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py b/paimon-python/pypaimon/tests/write/conflict_detection_test.py index 6d06762b3fa5..62889370f56d 100644 --- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py +++ b/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): diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py b/paimon-python/pypaimon/write/commit/conflict_detection.py index 14b6c8a0e2d6..30615ec893a5 100644 --- a/paimon-python/pypaimon/write/commit/conflict_detection.py +++ b/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