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 1dbe05f9d0 [python] Include partition and bucket context in file
deletion conflicts (#10810)
1dbe05f9d0 is described below
commit 1dbe05f9d0ce553f4d467409d61664f28393e379
Author: Ran Tao <[email protected]>
AuthorDate: Tue Oct 27 14:13:39 2026 +0800
[python] Include partition and bucket context in file deletion conflicts
(#10810)
---
.../python/write/commit/conflict_detection.py | 21 +++++++-
.../tests/overwrite_commit_conflict_test.py | 35 +++++++++++++
.../tests/write/conflict_detection_test.py | 60 ++++++++++++++++++++++
3 files changed, 114 insertions(+), 2 deletions(-)
diff --git
a/paimon-python/pypaimon/execution/python/write/commit/conflict_detection.py
b/paimon-python/pypaimon/execution/python/write/commit/conflict_detection.py
index 3f6c3e5553..bd0d9df352 100644
--- a/paimon-python/pypaimon/execution/python/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/execution/python/write/commit/conflict_detection.py
@@ -26,6 +26,7 @@ from pypaimon.manifest.index_manifest_file import
IndexManifestFile
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.file_entry import FileEntry
from pypaimon.table.special_fields import SpecialFields
+from pypaimon.utils.file_store_path_factory import _escape_partition_component
from pypaimon.utils.range import Range
from pypaimon.utils.range_helper import RangeHelper
from pypaimon.write.row_id_conflict import RowIdExistenceConflict
@@ -226,8 +227,9 @@ class ConflictDetection:
if entry.kind == 1:
return RuntimeError(
"File deletion conflicts detected! Give up committing. "
- "Trying to delete file {} which is not previously
added.".format(
- entry.file.file_name))
+ "Trying to delete file {} which is not previously added "
+ "(partition={}, bucket={}).".format(
+ entry.file.file_name,
self._partition_to_string(entry.partition), entry.bucket))
conflict = self.check_bucket_num_conflicts(merged_entries)
if conflict is not None:
@@ -279,6 +281,21 @@ class ConflictDetection:
return self.check_row_id_from_snapshot(latest_snapshot, delta_entries)
+ def _partition_to_string(self, partition):
+ values = tuple(partition.values)
+ if not values:
+ return "<non-partitioned>"
+ try:
+ path_factory = self.table.path_factory()
+ partition_values = path_factory._canonical_partition(values)
+ return "/".join(
+ "{}={}".format(_escape_partition_component(key),
_escape_partition_component(partition_values[i]))
+ for i, key in enumerate(path_factory.partition_keys)
+ )
+ except Exception:
+ # Formatting diagnostics must not hide the original commit
conflict.
+ return str(values)
+
@staticmethod
def check_bucket_num_conflicts(entries):
total_buckets = {}
diff --git a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
index 3a5ea143ca..dfb57badcb 100644
--- a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
+++ b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
@@ -29,8 +29,10 @@ import unittest
import pandas as pd
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
+from pypaimon.write.commit_message import CommitMessage
class OverwriteCommitConflictTest(unittest.TestCase):
@@ -87,6 +89,39 @@ class OverwriteCommitConflictTest(unittest.TestCase):
w.close()
tc.close()
+ @pytest.mark.python_commit
+ def test_repeated_file_delete_reports_partition_and_bucket(self):
+ builder = self.table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_pandas(pd.DataFrame({'f0': [3], 'f1': ['delete-me']}))
+ messages = writer.prepare_commit()
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
+ self.assertEqual(len(messages), 1)
+ original = messages[0]
+ deletion = CommitMessage(
+ original.partition, original.bucket, [],
deleted_files=original.new_files)
+ first = builder.new_commit()
+ try:
+ first.commit([deletion])
+ finally:
+ first.close()
+
+ second = builder.new_commit()
+ try:
+ with self.assertRaises(RuntimeError) as raised:
+ second.commit([deletion])
+ message = str(raised.exception)
+ self.assertIn("File deletion conflicts detected", message)
+ self.assertIn(original.new_files[0].file_name, message)
+ self.assertIn("partition=f0=3, bucket={}".format(original.bucket),
message)
+ finally:
+ second.close()
+
def test_conflict_scan_runs_once_not_once_per_retry(self):
K = 3
diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
index a36b80ce2e..95b6b17de1 100644
--- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
+++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
@@ -17,7 +17,11 @@
import unittest
from dataclasses import dataclass
+from datetime import date
from typing import List
+from unittest.mock import Mock
+
+from parameterized import parameterized
from pypaimon.index.index_file_meta import IndexFileMeta
from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
@@ -27,6 +31,7 @@ from pypaimon.manifest.schema.manifest_entry import
ManifestEntry
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.schema.table_schema import TableSchema
from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
from pypaimon.execution.python.write.commit.conflict_detection import (
ConflictDetection,
RowIdColumnConflictChecker,
@@ -329,6 +334,61 @@ class TestOverwriteConflictDetection(unittest.TestCase):
self.assertIn("Total buckets", str(result))
+class TestFileDeletionConflictDiagnostics(unittest.TestCase):
+
+ @parameterized.expand([
+ ("string", [("dt", "STRING")], ["2026-10-26"], False, "dt=2026-10-26"),
+ ("multiple", [("dt", "STRING"), ("hr", "INT")], ["2026-10-26", 12],
False,
+ "dt=2026-10-26/hr=12"),
+ ("escaped", [("a/b", "STRING"), ("hr", "STRING")], ["x/y=z", None],
False,
+ "a%2Fb=x%2Fy%3Dz/hr=MISSING"),
+ ("null", [("dt", "STRING")], [None], False, "dt=MISSING"),
+ ("empty", [("dt", "STRING")], [""], False, "dt=MISSING"),
+ ("date", [("dt", "DATE")], [date(2026, 10, 26)], False,
"dt=2026-10-26"),
+ ("legacy_date", [("dt", "DATE")], [date(1970, 1, 2)], True, "dt=1"),
+ ("non_partitioned", [], [], False, "<non-partitioned>"),
+ ])
+ def test_missing_file_reports_partition_and_bucket(
+ self, _, field_types, values, legacy, expected):
+ fields = [DataField(i, name, AtomicType(type_name))
+ for i, (name, type_name) in enumerate(field_types)]
+ table = Mock()
+ table.path_factory.return_value = FileStorePathFactory(
+ root="/table", partition_keys=[f.name for f in fields],
+ partition_types=[f.type for f in fields],
default_part_value="MISSING",
+ format_identifier="parquet", data_file_prefix="data-",
+ changelog_file_prefix="changelog-", legacy_partition_name=legacy,
+ file_suffix_include_compression=False, file_compression="zstd",
+ data_file_path_directory="data")
+
+ result = self._missing_file_conflict(table, fields, values)
+
+ self.assertIsInstance(result, RuntimeError)
+ self.assertIn("Trying to delete file missing.parquet which is not
previously added", str(result))
+ self.assertIn("partition={}, bucket=3".format(expected), str(result))
+ if not fields:
+ table.path_factory.assert_not_called()
+
+ def test_partition_formatting_failure_preserves_conflict(self):
+ table = Mock()
+ table.path_factory.return_value._canonical_partition.side_effect =
ValueError("invalid partition")
+
+ result = self._missing_file_conflict(
+ table, [DataField(0, "dt", AtomicType("STRING"))], ["2026-10-26"])
+
+ self.assertIsInstance(result, RuntimeError)
+ self.assertIn("Trying to delete file missing.parquet which is not
previously added", str(result))
+ self.assertIn("partition=('2026-10-26',), bucket=3", str(result))
+ self.assertNotIn("invalid partition", str(result))
+
+ @staticmethod
+ def _missing_file_conflict(table, fields, values):
+ entry = _make_entry("missing.parquet", kind=1, bucket=3)
+ entry.partition = GenericRow(values, fields)
+ detection = ConflictDetection(False, None, None, table, None)
+ return detection.check_conflicts(None, [], [entry], "OVERWRITE")
+
+
class _FakeSnapshot:
def __init__(self, snapshot_id, commit_kind, next_row_id=None):
