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 00dca14161 [python] Invalidate commit retry cache after overwrite
(#9688)
00dca14161 is described below
commit 00dca141616470c67f08a4a52bf20bd9f589717a
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Sep 8 16:56:57 2026 +0800
[python] Invalidate commit retry cache after overwrite (#9688)
---
.../tests/overwrite_commit_conflict_test.py | 32 ++++++++++++++++++----
.../pypaimon/write/commit/commit_scanner.py | 6 ++--
2 files changed, 31 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
index da20c300bc..919d4830a3 100644
--- a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
+++ b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
@@ -276,15 +276,17 @@ class OverwriteCommitConflictTest(unittest.TestCase):
self.assertEqual([], incr_results)
self.assertEqual(full_scans['n'], 1)
- def test_incremental_merge_across_non_append_snapshot(self):
- self._assert_merge_equals_full_scan(self._overwrite_target)
+ def test_full_scan_across_empty_delta_overwrite_snapshot(self):
+ self._assert_merge_equals_full_scan(
+ self._overwrite_target, hide_overwrite_delta=True)
def test_incremental_merge_across_compact_snapshot(self):
self._assert_merge_equals_full_scan(self._compact_target)
- def _assert_merge_equals_full_scan(self, concurrent_fn):
- # A non-APPEND snapshot (OVERWRITE/COMPACT, delta = ADD+DELETE) lands
- # between retries; the merged base must still equal a fresh full scan.
+ def _assert_merge_equals_full_scan(self, concurrent_fn,
+ hide_overwrite_delta=False):
+ # A non-APPEND snapshot lands between retries; the conflict base must
+ # still equal a fresh full scan.
K = 2
wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
@@ -298,6 +300,18 @@ class OverwriteCommitConflictTest(unittest.TestCase):
orig_check = fsc.conflict_detection.check_conflicts
orig_full = fsc.commit_scanner.read_all_entries_from_changed_partitions
+ if hide_overwrite_delta:
+ orig_read_delta =
fsc.commit_scanner.manifest_list_manager.read_delta
+
+ def read_delta(snapshot):
+ # Row-id reassignment replaces the base manifests in an
+ # OVERWRITE snapshot while deliberately writing an empty delta.
+ if snapshot.commit_kind == "OVERWRITE":
+ return []
+ return orig_read_delta(snapshot)
+
+ fsc.commit_scanner.manifest_list_manager.read_delta = read_delta
+
def spy_check(latest_snapshot, base_entries, delta_entries, *a, **k):
captured.append((latest_snapshot, list(base_entries),
list(delta_entries)))
return orig_check(latest_snapshot, base_entries, delta_entries,
*a, **k)
@@ -316,6 +330,14 @@ class OverwriteCommitConflictTest(unittest.TestCase):
fsc.snapshot_commit.commit = patched_cas
+ if hide_overwrite_delta:
+ with self.assertRaisesRegex(
+ RuntimeError, "File deletion conflicts detected"):
+ c.commit(messages)
+ c.close()
+ self.assertEqual(cas['fails'], 1)
+ return
+
c.commit(messages)
c.close()
diff --git a/paimon-python/pypaimon/write/commit/commit_scanner.py
b/paimon-python/pypaimon/write/commit/commit_scanner.py
index 4612054e6c..835df4bf61 100644
--- a/paimon-python/pypaimon/write/commit/commit_scanner.py
+++ b/paimon-python/pypaimon/write/commit/commit_scanner.py
@@ -134,8 +134,8 @@ class CommitScanner:
index_entries=None) ->
Optional[List[ManifestEntry]]:
"""Delta entries (incl. DELETEs) in ``(from_snapshot, to_snapshot]``,
changed-partition filtered, so a retry can reuse the prior base and
read
- only the changes since. Returns None on a missing snapshot (caller then
- full-scans). Mirrors Java ``CommitScanner#readIncrementalChanges``.
+ only the changes since. Returns None on a missing or OVERWRITE snapshot
+ (caller then full-scans). Mirrors Java conflict detection behavior.
"""
snapshot_manager = self.table.snapshot_manager()
partition_filter = self._build_partition_filter_from_changes(
@@ -145,6 +145,8 @@ class CommitScanner:
snapshot = snapshot_manager.get_snapshot_by_id(snapshot_id)
if snapshot is None:
return None
+ if snapshot.commit_kind == "OVERWRITE":
+ return None
entries.extend(
self.read_incremental_raw_entries_from_changed_partitions(
snapshot, commit_entries, partition_filter))
