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 62e26b8f1a [python] Reuse base entries on commit retry instead of full
re-scan (#8358)
62e26b8f1a is described below
commit 62e26b8f1a63aeb50db3a6150ec425da94aa4d5a
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Jun 28 00:19:20 2026 +0800
[python] Reuse base entries on commit retry instead of full re-scan (#8358)
---
.../tests/overwrite_commit_conflict_test.py | 332 +++++++++++++++++++++
.../pypaimon/write/commit/commit_scanner.py | 31 +-
paimon-python/pypaimon/write/file_store_commit.py | 40 ++-
3 files changed, 390 insertions(+), 13 deletions(-)
diff --git a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
new file mode 100644
index 0000000000..738d455b85
--- /dev/null
+++ b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
@@ -0,0 +1,332 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Conflict-detection base scan is reused across commit retries.
+
+The first attempt full-scans the changed partitions; later retries reuse that
+base and read only the incremental changes since (read_incremental_changes).
+"""
+
+import os
+import shutil
+import tempfile
+import unittest
+
+import pandas as pd
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+
+
+class OverwriteCommitConflictTest(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="ow_conflict_")
+ self.warehouse = os.path.join(self.temp_dir, 'wh')
+ self.catalog = CatalogFactory.create({"warehouse": self.warehouse})
+ self.catalog.create_database("test_db", True)
+
+ pa_schema = pa.schema([('f0', pa.int32()), ('f1', pa.string())])
+ # Static overwrite, scoped to the explicit partition f0=1.
+ schema = Schema.from_pyarrow_schema(
+ pa_schema, partition_keys=['f0'],
+ options={'dynamic-partition-overwrite': 'false'})
+ self.catalog.create_table('test_db.t', schema, False)
+ self.table = self.catalog.get_table('test_db.t')
+
+ # f0=1 is the overwrite target, f0=2 untouched.
+ self._append(pd.DataFrame({'f0': [1, 1, 2], 'f1': ['a', 'b', 'c']}))
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def _append(self, df):
+ wb = self.table.new_batch_write_builder()
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(df)
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ def _overwrite_target(self, f1_val):
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': [f1_val]}))
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ def _compact_target(self, f1_val):
+ # pypaimon has no compact API; produce a COMPACT-kind snapshot by
running
+ # an overwrite of f0=1 but labelling the snapshot COMPACT.
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ tc = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': [f1_val]}))
+ cfsc = tc.file_store_commit
+ orig_try = cfsc._try_commit
+ cfsc._try_commit = lambda commit_kind, *a, **k: orig_try("COMPACT",
*a, **k)
+ tc.commit(w.prepare_commit())
+ w.close()
+ tc.close()
+
+ def test_conflict_scan_runs_once_not_once_per_retry(self):
+ K = 3
+
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ messages = w.prepare_commit()
+
+ fsc = c.file_store_commit
+ counts = {'full_scan': 0, 'incremental': 0}
+ orig_full = fsc.commit_scanner.read_all_entries_from_changed_partitions
+ orig_incr = fsc.commit_scanner.read_incremental_changes
+
+ def spy_full(*a, **k):
+ counts['full_scan'] += 1
+ return orig_full(*a, **k)
+
+ def spy_incr(*a, **k):
+ counts['incremental'] += 1
+ return orig_incr(*a, **k)
+
+ fsc.commit_scanner.read_all_entries_from_changed_partitions = spy_full
+ fsc.commit_scanner.read_incremental_changes = spy_incr
+
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ # Each conflict appends to an unrelated partition (f0=99),
advancing
+ # latest, then fails our CAS.
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ self._append(pd.DataFrame({'f0': [99], 'f1':
[f'x{cas["fails"]}']}))
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+
+ c.commit(messages)
+ c.close()
+
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+ self.assertEqual(counts['full_scan'], 1) # once, then incremental on
retry
+ self.assertEqual(counts['incremental'], K)
+
+ read_builder = self.table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(sorted(actual[actual['f0'] == 1]['f1'].tolist()),
['new'])
+ self.assertEqual(sorted(actual[actual['f0'] == 2]['f1'].tolist()),
['c'])
+ self.assertEqual(len(actual[actual['f0'] == 99]), K)
+
+ def
test_incremental_merge_when_concurrent_append_hits_target_partition(self):
+ # Concurrent appends hit the target partition, so the incremental read
is
+ # non-empty and must be merged into the reused base.
+ K = 3
+
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ messages = w.prepare_commit()
+
+ fsc = c.file_store_commit
+ full_scans = {'n': 0}
+ incr_lengths = []
+ captured = []
+ orig_full = fsc.commit_scanner.read_all_entries_from_changed_partitions
+ orig_incr = fsc.commit_scanner.read_incremental_changes
+ orig_check = fsc.conflict_detection.check_conflicts
+
+ def spy_full(*a, **k):
+ full_scans['n'] += 1
+ return orig_full(*a, **k)
+
+ def spy_incr(*a, **k):
+ r = orig_incr(*a, **k)
+ incr_lengths.append(None if r is None else len(r))
+ return r
+
+ 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)
+
+ fsc.commit_scanner.read_all_entries_from_changed_partitions = spy_full
+ fsc.commit_scanner.read_incremental_changes = spy_incr
+ fsc.conflict_detection.check_conflicts = spy_check
+
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ self._append(pd.DataFrame({'f0': [1], 'f1':
[f'y{cas["fails"]}']}))
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+
+ c.commit(messages)
+ c.close()
+
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+ self.assertEqual(full_scans['n'], 1)
+ self.assertEqual(len(incr_lengths), K)
+ self.assertTrue(all(incr_lengths),
+ f"expected non-empty incremental on every retry, got
{incr_lengths}")
+
+ # The incremental-merged base must equal a fresh full scan.
+ last_snapshot, merged_base, last_delta = captured[-1]
+ full = orig_full(last_snapshot, last_delta)
+ self.assertEqual(
+ set(e.identifier() for e in merged_base),
+ set(e.identifier() for e in full))
+
+ read_builder = self.table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(sorted(actual[actual['f0'] == 1]['f1'].tolist()),
['new'])
+ self.assertEqual(sorted(actual[actual['f0'] == 2]['f1'].tolist()),
['c'])
+
+ def test_falls_back_to_full_scan_when_intermediate_snapshot_missing(self):
+ # A missing intermediate snapshot -> read_incremental_changes returns
None
+ # and the retry falls back to a full scan.
+ K = 1
+ missing_id = self.table.snapshot_manager().get_latest_snapshot().id + 1
+
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ messages = w.prepare_commit()
+
+ fsc = c.file_store_commit
+ full_scans = {'n': 0}
+ incr_results = []
+ orig_full = fsc.commit_scanner.read_all_entries_from_changed_partitions
+ orig_incr = fsc.commit_scanner.read_incremental_changes
+
+ def spy_full(*a, **k):
+ full_scans['n'] += 1
+ return orig_full(*a, **k)
+
+ def spy_incr(*a, **k):
+ r = orig_incr(*a, **k)
+ incr_results.append(r)
+ return r
+
+ fsc.commit_scanner.read_all_entries_from_changed_partitions = spy_full
+ fsc.commit_scanner.read_incremental_changes = spy_incr
+
+ # Only the scanner's lookups see missing_id as absent; the commit's own
+ # manager (bound earlier) is untouched.
+ real_mgr = fsc.commit_scanner.table.snapshot_manager()
+
+ class _Wrap:
+ def __getattr__(self, name):
+ return getattr(real_mgr, name)
+
+ def get_snapshot_by_id(self, i):
+ return None if i == missing_id else
real_mgr.get_snapshot_by_id(i)
+
+ fsc.commit_scanner.table.snapshot_manager = lambda: _Wrap()
+
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ self._append(pd.DataFrame({'f0': [99], 'f1': ['x']}))
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+
+ c.commit(messages)
+ c.close()
+
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+ self.assertIn(None, incr_results) # incremental bailed on
missing
+ self.assertEqual(full_scans['n'], 2) # first attempt + fallback
+
+ def test_incremental_merge_across_non_append_snapshot(self):
+ self._assert_merge_equals_full_scan(self._overwrite_target)
+
+ 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.
+ K = 2
+
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ messages = w.prepare_commit()
+
+ fsc = c.file_store_commit
+ captured = []
+ orig_check = fsc.conflict_detection.check_conflicts
+ orig_full = fsc.commit_scanner.read_all_entries_from_changed_partitions
+
+ 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)
+
+ fsc.conflict_detection.check_conflicts = spy_check
+
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ concurrent_fn(f'z{cas["fails"]}')
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+
+ c.commit(messages)
+ c.close()
+
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+
+ last_snapshot, merged_base, last_delta = captured[-1]
+ full = orig_full(last_snapshot, last_delta)
+ self.assertEqual(
+ set(e.identifier() for e in merged_base),
+ set(e.identifier() for e in full))
+
+ read_builder = self.table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(sorted(actual[actual['f0'] == 1]['f1'].tolist()),
['new'])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/write/commit/commit_scanner.py
b/paimon-python/pypaimon/write/commit/commit_scanner.py
index 95702f1d02..eb39bc0ecf 100644
--- a/paimon-python/pypaimon/write/commit/commit_scanner.py
+++ b/paimon-python/pypaimon/write/commit/commit_scanner.py
@@ -97,16 +97,18 @@ class CommitScanner:
).read_manifest_entries(delta_manifests)
def read_incremental_raw_entries_from_changed_partitions(self, snapshot:
Snapshot,
- commit_entries:
List[ManifestEntry]):
- """Like ``read_incremental_entries_from_changed_partitions`` but
- preserves DELETE entries (kind=1). The regular method funnels through
- ``read_entries_parallel`` which discards standalone DELETEs.
+ commit_entries:
List[ManifestEntry],
+
partition_filter=None):
+ """Like ``read_incremental_entries_from_changed_partitions`` but
preserves
+ DELETE entries (kind=1). ``partition_filter`` may be passed to avoid
+ rebuilding it per call.
"""
delta_manifests = self.manifest_list_manager.read_delta(snapshot)
if not delta_manifests:
return []
- partition_filter =
self._build_partition_filter_from_entries(commit_entries)
+ if partition_filter is None:
+ partition_filter =
self._build_partition_filter_from_entries(commit_entries)
mfm = ManifestFileManager(self.table)
entries = []
for mf in delta_manifests:
@@ -116,6 +118,25 @@ class CommitScanner:
entries.append(entry)
return entries
+ def read_incremental_changes(self, from_snapshot: Snapshot, to_snapshot:
Snapshot,
+ commit_entries: List[ManifestEntry]) ->
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``.
+ """
+ snapshot_manager = self.table.snapshot_manager()
+ partition_filter =
self._build_partition_filter_from_entries(commit_entries)
+ entries = []
+ for snapshot_id in range(from_snapshot.id + 1, to_snapshot.id + 1):
+ snapshot = snapshot_manager.get_snapshot_by_id(snapshot_id)
+ if snapshot is None:
+ return None
+ entries.extend(
+ self.read_incremental_raw_entries_from_changed_partitions(
+ snapshot, commit_entries, partition_filter))
+ return entries
+
def _build_partition_filter_from_entries(self, entries:
List[ManifestEntry]):
"""Build a partition predicate that matches all partitions present in
the given entries.
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 3149a3678d..304792b412 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -27,6 +27,7 @@ from pypaimon.manifest.manifest_file_manager import
ManifestFileManager
from pypaimon.manifest.manifest_file_merger import ManifestFileMerger
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.file_entry import FileEntry
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
@@ -63,9 +64,13 @@ class SuccessResult(CommitResult):
class RetryResult(CommitResult):
- def __init__(self, latest_snapshot, exception: Optional[Exception] = None):
+ def __init__(self, latest_snapshot, exception: Optional[Exception] = None,
+ base_data_files: Optional[List[ManifestEntry]] = None):
self.latest_snapshot = latest_snapshot
self.exception = exception
+ # Base entries as of latest_snapshot, carried so the next attempt
reuses
+ # them and reads only the incremental changes.
+ self.base_data_files = base_data_files
def is_success(self) -> bool:
return False
@@ -374,17 +379,36 @@ class FileStoreCommit:
# process snapshot
new_snapshot_id = latest_snapshot.id + 1 if latest_snapshot else 1
- # Conflict detection: read base entries from latest snapshot, then
check conflicts
+ # Base entries for conflict detection. On retry, reuse the previous
+ # attempt's base + read only the incremental changes (mirrors Java).
+ base_data_files = None
if detect_conflicts and latest_snapshot is not None:
- base_entries =
self.commit_scanner.read_all_entries_from_changed_partitions(
- latest_snapshot, commit_entries)
+ incremental = None
+ if (retry_result is not None
+ and retry_result.latest_snapshot is not None
+ and retry_result.base_data_files is not None):
+ incremental = self.commit_scanner.read_incremental_changes(
+ retry_result.latest_snapshot, latest_snapshot,
commit_entries)
+ if incremental is not None:
+ base_data_files = list(retry_result.base_data_files)
+ if incremental:
+ base_data_files.extend(incremental)
+ base_data_files = FileEntry.merge_entries(base_data_files)
+ else:
+ # First attempt, or incremental could not be built (missing
+ # snapshot): scan the changed partitions in full.
+ base_data_files =
self.commit_scanner.read_all_entries_from_changed_partitions(
+ latest_snapshot, commit_entries)
+
conflict_exception = self.conflict_detection.check_conflicts(
- latest_snapshot, base_entries, commit_entries, commit_kind)
+ latest_snapshot, base_data_files, commit_entries, commit_kind)
if conflict_exception is not None:
if allow_rollback and self.rollback is not None:
if self.rollback.try_to_rollback(latest_snapshot):
- return RetryResult(latest_snapshot, conflict_exception)
+ # Rolled back: base/snapshot no longer valid; next
attempt
+ # re-scans from scratch (matches Java
RollbackRetryResult).
+ return RetryResult(None, conflict_exception)
raise conflict_exception
# Apply row tracking logic after conflict detection (matches Java
ordering)
@@ -492,11 +516,11 @@ class FileStoreCommit:
commit_kind,
commit_time_s,
)
- return RetryResult(latest_snapshot, None)
+ return RetryResult(latest_snapshot, None,
base_data_files=base_data_files)
except Exception as e:
# Commit exception, not sure about the situation and should not
clean up the files
logger.warning("Retry commit for exception.", exc_info=True)
- return RetryResult(latest_snapshot, e)
+ return RetryResult(latest_snapshot, e,
base_data_files=base_data_files)
logger.info(
"Successfully commit snapshot %d to table %s by user %s "
