This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e194867bcc [core][python] Improve data evolution row-id conflict
diagnostics (#8491)
e194867bcc is described below
commit e194867bcc0597653ffcf79979a0f90e2eb9c99b
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 7 18:19:35 2026 +0800
[core][python] Improve data evolution row-id conflict diagnostics (#8491)
This PR improves Data Evolution row-id range conflict detection so
normal data files and dedicated storage files are checked separately. It
prevents adjacent normal data files from being reported without the
blob/vector file that actually connects their row-id ranges.
---
.../paimon/operation/commit/ConflictDetection.java | 96 ++++++++++++++++---
.../operation/commit/ConflictDetectionTest.java | 58 ++++++++++++
.../tests/write/conflict_detection_test.py | 58 ++++++++++++
.../pypaimon/write/commit/conflict_detection.py | 102 ++++++++++++++++++---
4 files changed, 285 insertions(+), 29 deletions(-)
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 fcf00c792b..a679b9d091 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 @@ public class ConflictDetection {
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 @@ public class ConflictDetection {
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();
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 0695406bde..5be15f0dcc 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 @@ class ConflictDetectionTest {
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);
diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
index 6d06762b3f..62889370f5 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 @@ class TestCheckRowIdExistence(unittest.TestCase):
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 @@ class TestCheckRowIdExistence(unittest.TestCase):
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 14b6c8a0e2..30615ec893 100644
--- a/paimon-python/pypaimon/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/write/commit/conflict_detection.py
@@ -406,7 +406,7 @@ class ConflictDetection:
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 @@ class ConflictDetection:
}
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 @@ class ConflictDetection:
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
