[python] Include partition and bucket context in file deletion conflicts

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 "<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 = {}

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
 

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, "<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):