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 3f6c3e5553b5..bd0d9df35216 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.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 @@ def check_conflicts( 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 @@ def check_conflicts( 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 "" + 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 3a5ea143ca51..dfb57badcb18 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 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 @@ def _compact_target(self, f1_val): 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 a36b80ce2ebf..95b6b17de103 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.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 @@ def test_bucket_num_mismatch_conflicts(self): 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, ""), + ]) + 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):