From 8b7bc7233083035782f15a45ac2f339fb4df4857 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 23 Jul 2026 15:54:10 +0800 Subject: [PATCH 1/8] [python][ray] Incrementally commit update_by_row_id file groups --- docs/docs/pypaimon/ray-data.md | 7 + .../pypaimon/ray/data_evolution_merge_join.py | 34 +++- .../pypaimon/ray/update_by_row_id.py | 118 +++++++++++- .../tests/ray_update_by_row_id_test.py | 176 ++++++++++++++++++ 4 files changed, 320 insertions(+), 15 deletions(-) diff --git a/docs/docs/pypaimon/ray-data.md b/docs/docs/pypaimon/ray-data.md index 88c0a7143ac4..2bbf2bb9e712 100644 --- a/docs/docs/pypaimon/ray-data.md +++ b/docs/docs/pypaimon/ray-data.md @@ -524,6 +524,7 @@ metrics = update_by_row_id( source=ray_dataset, # ray.data.Dataset / pa.Table / pandas, carrying _ROW_ID catalog_options={"warehouse": "/path/to/warehouse"}, update_cols=["feature"], # non-blob columns to overwrite + max_groups_per_commit=64, # optional incremental commit window ) print(metrics) # {"num_updated": 50} ``` @@ -537,6 +538,9 @@ print(metrics) # {"num_updated": 50} - `num_partitions`: parallelism for grouping the update rows by target file; defaults to `max(1, cluster_cpus * 2)`. - `ray_remote_args`: Ray remote options applied to the update tasks. +- `max_groups_per_commit`: optional positive number of completed target file groups + per commit. By default, all groups are committed together after every worker + finishes. For example, `64` commits after each 64 `_FIRST_ROW_ID` file groups. **Returns:** `{"num_updated": }`. @@ -548,6 +552,9 @@ print(metrics) # {"num_updated": 50} - Partition columns cannot be updated (in-place rewrite can't move a row across partitions). - Deletion-vectors-enabled tables are not supported yet: a DV-deleted row still lives in its data file, so it can't be told apart from a live row without reading the target. +- Incremental commit mode overlaps later worker writes with earlier commits, but is not + atomic across the whole operation. If a later group fails, earlier committed windows + remain visible. ## Read By Row Id diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index 359af8e6efa2..0638dbca395a 100644 --- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py +++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py @@ -16,7 +16,7 @@ # limitations under the License. ################################################################################ -from typing import Any, Dict, List, Optional, Sequence, Tuple +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple import pyarrow as pa @@ -445,7 +445,15 @@ def distributed_update_apply( ray_remote_args: Optional[Dict[str, Any]] = None, base_snapshot_id: Optional[int] = None, collect_row_ids: bool = False, + on_group_result: Optional[Callable[[list, int, list], None]] = None, ) -> Tuple[list, int, list]: + """Apply updates grouped by target file and collect their commit messages. + + When ``on_group_result`` is set, it is called on the driver once for every + completed ``_FIRST_ROW_ID`` group. Commit messages are delivered to the + callback instead of being retained in the returned list, which lets callers + commit completed file groups incrementally while the remaining groups run. + """ import numpy as np import pickle import uuid @@ -599,13 +607,25 @@ def _apply_group(group: pa.Table) -> pa.Table: num_updated = 0 action_row_ids = [] for batch in msgs_ds.iter_batches(batch_format="pyarrow"): - for blob in batch.column("msgs_blob").to_pylist(): - all_msgs.extend(pickle.loads(blob)) - for n in batch.column("n_updated").to_pylist(): + message_blobs = batch.column("msgs_blob").to_pylist() + updated_counts = batch.column("n_updated").to_pylist() + row_id_blobs = ( + batch.column("row_ids_blob").to_pylist() + if collect_row_ids else [None] * len(message_blobs) + ) + for blob, n, row_ids_blob in zip( + message_blobs, updated_counts, row_id_blobs): + group_msgs = pickle.loads(blob) + group_row_ids = ( + pickle.loads(row_ids_blob) if collect_row_ids else [] + ) + if on_group_result is None: + all_msgs.extend(group_msgs) + else: + on_group_result(group_msgs, n, group_row_ids) num_updated += n - if collect_row_ids: - for blob in batch.column("row_ids_blob").to_pylist(): - action_row_ids.extend(pickle.loads(blob)) + if collect_row_ids: + action_row_ids.extend(group_row_ids) return all_msgs, num_updated, action_row_ids diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index 41b968c4fdc8..bbeba039bd47 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -56,6 +56,7 @@ def update_by_row_id( update_cols: List[str], num_partitions: Optional[int] = None, ray_remote_args: Optional[Dict[str, Any]] = None, + max_groups_per_commit: Optional[int] = None, ) -> Dict[str, int]: """Update ``update_cols`` of a data-evolution table by ``_ROW_ID``. @@ -65,6 +66,11 @@ def update_by_row_id( the target is never fully read and there is no join against it. Requires ``ray >= 2.50`` and a target with ``data-evolution.enabled`` + ``row-tracking.enabled``. + By default all file groups are committed atomically after every worker + finishes. Set ``max_groups_per_commit`` to incrementally commit each completed + window of target file groups. Incremental mode can leave earlier windows + committed if a later window fails. + Returns ``{"num_updated": }``. """ from pypaimon.catalog.catalog_factory import CatalogFactory @@ -74,6 +80,11 @@ def update_by_row_id( _require_ray_join() if not update_cols: raise ValueError("update_cols must be non-empty.") + if max_groups_per_commit is not None: + if (isinstance(max_groups_per_commit, bool) + or not isinstance(max_groups_per_commit, int) + or max_groups_per_commit <= 0): + raise ValueError("max_groups_per_commit must be a positive integer.") update_cols = list(dict.fromkeys(update_cols)) # de-dup, keep order num_partitions = _resolve_num_partitions(num_partitions) @@ -139,22 +150,113 @@ def _project_cast(batch: pa.Table) -> pa.Table: raise ValueError( f"target '{target}' has no rows; every _ROW_ID in the source is foreign.") return {"num_updated": 0} + incremental_committer = ( + _IncrementalUpdateCommitter(table, max_groups_per_commit) + if max_groups_per_commit is not None else None + ) try: + apply_kwargs = { + "num_partitions": num_partitions, + "ray_remote_args": ray_remote_args, + "base_snapshot_id": base.id, + } + if incremental_committer is not None: + apply_kwargs["on_group_result"] = incremental_committer.add_group msgs, num_updated, _ = distributed_update_apply( - update_ds, table, update_cols, - num_partitions=num_partitions, - ray_remote_args=ray_remote_args, - base_snapshot_id=base.id, + update_ds, table, update_cols, **apply_kwargs ) + if incremental_committer is None: + if msgs: + _commit_update_messages(table, msgs) + else: + incremental_committer.finish() except Exception as e: + if incremental_committer is not None: + incremental_committer.abort_pending() _reraise_inner(e) - raise # _reraise_inner always raises; keeps msgs/num_updated defined for linters - - if msgs: - _commit_update_messages(table, msgs) + raise # _reraise_inner always raises; keeps num_updated defined for linters + finally: + if incremental_committer is not None: + incremental_committer.close() return {"num_updated": num_updated} +class _IncrementalUpdateCommitter: + + def __init__(self, table, max_groups_per_commit: int): + self._table = table + self._max_groups_per_commit = max_groups_per_commit + self._pending_messages: list = [] + self._pending_groups = 0 + self._table_commit = None + self._next_commit_identifier = 1 + self._pending_commit_started = False + + def add_group(self, commit_messages, _num_updated, _row_ids) -> None: + self._pending_messages.extend(commit_messages) + self._pending_groups += 1 + if self._pending_groups >= self._max_groups_per_commit: + self._commit_pending() + + def finish(self) -> None: + self._commit_pending() + + def _commit_pending(self) -> None: + if self._pending_groups == 0: + return + if not self._pending_messages: + self._pending_groups = 0 + return + + if self._table_commit is None: + self._table_commit = ( + self._table.new_stream_write_builder().new_commit() + ) + + self._pending_commit_started = True + group_count = self._pending_groups + self._table_commit.commit( + self._pending_messages, self._next_commit_identifier + ) + self._pending_messages = [] + self._pending_groups = 0 + self._pending_commit_started = False + self._next_commit_identifier += 1 + logger.info( + "Incrementally committed %d update_by_row_id file groups.", + group_count, + ) + + def abort_pending(self) -> None: + if not self._pending_messages or self._pending_commit_started: + return + + if self._table_commit is None: + _abort_pending_update_messages(self._table, self._pending_messages) + return + + try: + self._table_commit.abort(self._pending_messages) + except Exception as abort_error: + logger.warning( + "Failed to abort pending incremental update_by_row_id messages: %s", + abort_error, + exc_info=abort_error, + ) + + def close(self) -> None: + if self._table_commit is None: + return + try: + self._table_commit.close() + except Exception as close_error: + logger.warning( + "Failed to close incremental update_by_row_id commit: %s", + close_error, + exc_info=close_error, + ) + + def _commit_update_messages(table, commit_messages) -> None: pending_msgs: list = list(commit_messages) commit_started = False diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py index 613fa50d0319..3a24840f5f0a 100644 --- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py @@ -130,6 +130,182 @@ def test_updates_correct_row_across_files(self): self.assertEqual(got[21], 999) self.assertTrue(all(v == 0 for k, v in got.items() if k != 21)) + def test_incrementally_commits_file_group_windows(self): + target = self._create() + chunks = [ + [group_start, group_start + 1] + for group_start in range(10, 90, 10) + ] + for chunk in chunks: + self._write(target, pa.Table.from_pydict( + {"id": chunk, "name": ["x"] * len(chunk), "age": [0] * len(chunk)}, + schema=self.pa_schema, + )) + + table = self.catalog.get_table(target) + base_snapshot_id = table.snapshot_manager().get_latest_snapshot().id + rid = self._rowid_by_id(target) + updated_ids = [chunk[0] for chunk in chunks] + src = pa.table( + { + "_ROW_ID": [rid[row_id] for row_id in updated_ids], + "age": updated_ids, + }, + schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]), + ) + + stats = update_by_row_id( + target, + ray.data.from_arrow(src).repartition(8), + self.catalog_options, + update_cols=["age"], + num_partitions=4, + max_groups_per_commit=3, + ) + + self.assertEqual({"num_updated": 8}, stats) + latest = table.snapshot_manager().get_latest_snapshot() + self.assertEqual(base_snapshot_id + 3, latest.id) + incremental_snapshots = [ + table.snapshot_manager().get_snapshot_by_id(base_snapshot_id + offset) + for offset in range(1, 4) + ] + self.assertEqual(1, len({ + snapshot.commit_user for snapshot in incremental_snapshots + })) + self.assertEqual( + [1, 2, 3], + [snapshot.commit_identifier for snapshot in incremental_snapshots], + ) + + back = self._read(target).sort_by("id").to_pydict() + got = dict(zip(back["id"], back["age"])) + self.assertEqual( + {row_id: row_id for row_id in updated_ids}, + {row_id: got[row_id] for row_id in updated_ids}, + ) + for chunk in chunks: + self.assertEqual(0, got[chunk[1]]) + + def test_incremental_committer_batches_complete_groups(self): + import importlib + m = importlib.import_module("pypaimon.ray.update_by_row_id") + recorder = {"commits": [], "aborts": [], "close_calls": 0} + + class FakeCommit: + def commit(self, msgs, commit_identifier): + recorder["commits"].append((list(msgs), commit_identifier)) + + def abort(self, msgs): + recorder["aborts"].append(list(msgs)) + + def close(self): + recorder["close_calls"] += 1 + + class FakeBuilder: + def new_commit(self): + return FakeCommit() + + class FakeTable: + def new_stream_write_builder(self): + return FakeBuilder() + + committer = m._IncrementalUpdateCommitter(FakeTable(), 2) + committer.add_group(["group-1"], 1, []) + self.assertEqual([], recorder["commits"]) + committer.add_group(["group-2"], 1, []) + committer.add_group(["group-3"], 1, []) + committer.finish() + committer.close() + + self.assertEqual([ + (["group-1", "group-2"], 1), + (["group-3"], 2), + ], recorder["commits"]) + self.assertEqual([], recorder["aborts"]) + self.assertEqual(1, recorder["close_calls"]) + + def test_incremental_committer_aborts_only_uncommitted_groups(self): + import importlib + m = importlib.import_module("pypaimon.ray.update_by_row_id") + recorder = {"commits": [], "aborts": []} + + class FakeCommit: + def commit(self, msgs, commit_identifier): + recorder["commits"].append((list(msgs), commit_identifier)) + + def abort(self, msgs): + recorder["aborts"].append(list(msgs)) + + def close(self): + pass + + class FakeBuilder: + def new_commit(self): + return FakeCommit() + + class FakeTable: + def new_stream_write_builder(self): + return FakeBuilder() + + committer = m._IncrementalUpdateCommitter(FakeTable(), 2) + committer.add_group(["committed-1"], 1, []) + committer.add_group(["committed-2"], 1, []) + committer.add_group(["pending"], 1, []) + committer.abort_pending() + committer.close() + + self.assertEqual([ + (["committed-1", "committed-2"], 1), + ], recorder["commits"]) + self.assertEqual([["pending"]], recorder["aborts"]) + + def test_incremental_committer_does_not_abort_uncertain_commit(self): + import importlib + m = importlib.import_module("pypaimon.ray.update_by_row_id") + recorder = {"abort_calls": 0} + + class FakeCommit: + def commit(self, msgs, commit_identifier): + raise RuntimeError("commit outcome is uncertain") + + def abort(self, msgs): + recorder["abort_calls"] += 1 + + def close(self): + pass + + class FakeBuilder: + def new_commit(self): + return FakeCommit() + + class FakeTable: + def new_stream_write_builder(self): + return FakeBuilder() + + committer = m._IncrementalUpdateCommitter(FakeTable(), 1) + with self.assertRaisesRegex(RuntimeError, "outcome is uncertain"): + committer.add_group(["possibly-committed"], 1, []) + committer.abort_pending() + committer.close() + + self.assertEqual(0, recorder["abort_calls"]) + + def test_rejects_invalid_max_groups_per_commit(self): + target = self._create() + src = pa.table({"_ROW_ID": pa.array([], pa.int64()), + "age": pa.array([], pa.int32())}) + for value in (0, -1, True, 1.5): + with self.assertRaisesRegex( + ValueError, "max_groups_per_commit must be a positive integer"): + update_by_row_id( + target, + src, + self.catalog_options, + update_cols=["age"], + max_groups_per_commit=value, + ) + def test_pins_base_snapshot_for_conflict_detection(self): # The update pins its base snapshot and threads it to distributed_update_apply, # which uses it for commit-time conflict detection against concurrent writers. From b2e7c6da7ed94b647ea983c956d59daf64f4fdc3 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 23 Jul 2026 18:41:46 +0800 Subject: [PATCH 2/8] [python][ray] Fix incremental update_by_row_id commits --- .../pypaimon/ray/data_evolution_merge_join.py | 7 +- .../pypaimon/ray/update_by_row_id.py | 49 +++-- .../tests/ray_update_by_row_id_test.py | 197 ++++++++++++++++-- .../tests/write/conflict_detection_test.py | 89 +++++++- .../write/commit/conflict_detection.py | 50 ++++- .../pypaimon/write/file_store_commit.py | 4 +- paimon-python/pypaimon/write/table_commit.py | 6 + 7 files changed, 363 insertions(+), 39 deletions(-) diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index 0638dbca395a..3eaf16581fcc 100644 --- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py +++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py @@ -606,7 +606,12 @@ def _apply_group(group: pa.Table) -> pa.Table: all_msgs: list = [] num_updated = 0 action_row_ids = [] - for batch in msgs_ds.iter_batches(batch_format="pyarrow"): + iter_batches_kwargs = {"batch_format": "pyarrow"} + if on_group_result is not None: + # Yield ready groups without waiting for batch fill or prefetch. + iter_batches_kwargs["batch_size"] = 1 + iter_batches_kwargs["prefetch_batches"] = 0 + for batch in msgs_ds.iter_batches(**iter_batches_kwargs): message_blobs = batch.column("msgs_blob").to_pylist() updated_counts = batch.column("n_updated").to_pylist() row_id_blobs = ( diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index bbeba039bd47..a2543299de6d 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -190,16 +190,21 @@ def __init__(self, table, max_groups_per_commit: int): self._pending_groups = 0 self._table_commit = None self._next_commit_identifier = 1 - self._pending_commit_started = False + self._deferred_error = None + self._uncertain_messages: list = [] def add_group(self, commit_messages, _num_updated, _row_ids) -> None: self._pending_messages.extend(commit_messages) self._pending_groups += 1 - if self._pending_groups >= self._max_groups_per_commit: + if (self._deferred_error is None + and self._pending_groups >= self._max_groups_per_commit): self._commit_pending() def finish(self) -> None: - self._commit_pending() + if self._deferred_error is None: + self._commit_pending() + if self._deferred_error is not None: + raise self._deferred_error def _commit_pending(self) -> None: if self._pending_groups == 0: @@ -209,18 +214,33 @@ def _commit_pending(self) -> None: return if self._table_commit is None: - self._table_commit = ( - self._table.new_stream_write_builder().new_commit() - ) + try: + self._table_commit = ( + self._table.new_stream_write_builder().new_commit() + ) + except Exception as error: + # No commit started; all messages remain safe to abort. + self._deferred_error = error + return - self._pending_commit_started = True group_count = self._pending_groups - self._table_commit.commit( - self._pending_messages, self._next_commit_identifier - ) + commit_identifier = self._next_commit_identifier + try: + self._table_commit.commit( + self._pending_messages, commit_identifier + ) + except Exception as error: + # Defer the error; this window's commit outcome is uncertain. + self._uncertain_messages.extend(self._pending_messages) + self._pending_messages = [] + self._pending_groups = 0 + self._deferred_error = error + return + + # Later windows are disjoint by _FIRST_ROW_ID. + self._table_commit.ignore_row_id_conflict_for_commit(commit_identifier) self._pending_messages = [] self._pending_groups = 0 - self._pending_commit_started = False self._next_commit_identifier += 1 logger.info( "Incrementally committed %d update_by_row_id file groups.", @@ -228,11 +248,13 @@ def _commit_pending(self) -> None: ) def abort_pending(self) -> None: - if not self._pending_messages or self._pending_commit_started: + if not self._pending_messages: return if self._table_commit is None: _abort_pending_update_messages(self._table, self._pending_messages) + self._pending_messages = [] + self._pending_groups = 0 return try: @@ -243,6 +265,9 @@ def abort_pending(self) -> None: abort_error, exc_info=abort_error, ) + finally: + self._pending_messages = [] + self._pending_groups = 0 def close(self) -> None: if self._table_commit is None: diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py index 3a24840f5f0a..661db5e0b46e 100644 --- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py @@ -18,6 +18,7 @@ import os import shutil import tempfile +import threading import types import unittest import uuid @@ -154,16 +155,56 @@ def test_incrementally_commits_file_group_windows(self): schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]), ) - stats = update_by_row_id( - target, - ray.data.from_arrow(src).repartition(8), - self.catalog_options, - update_cols=["age"], - num_partitions=4, - max_groups_per_commit=3, - ) + original_iter_batches = ray.data.Dataset.iter_batches + iter_batch_options = [] + first_window_committed = threading.Event() + resume_iteration = threading.Event() + observed = {} + + def pausing_iter_batches(dataset, *args, **kwargs): + iter_batch_options.append(( + kwargs.get("batch_size"), + kwargs.get("prefetch_batches"), + )) + for batch_number, batch in enumerate( + original_iter_batches(dataset, *args, **kwargs), 1): + yield batch + if batch_number == 3: + first_window_committed.set() + if not resume_iteration.wait(30): + raise TimeoutError("timed out waiting to resume iteration") + + def observe_first_window(): + if not first_window_committed.wait(30): + observed["error"] = "first commit was not observed" + else: + observed["snapshot_id"] = ( + table.snapshot_manager().get_latest_snapshot().id + ) + resume_iteration.set() + + observer = threading.Thread(target=observe_first_window) + observer.start() + + try: + with mock.patch.object( + ray.data.Dataset, "iter_batches", pausing_iter_batches): + stats = update_by_row_id( + target, + ray.data.from_arrow(src).repartition(8), + self.catalog_options, + update_cols=["age"], + num_partitions=4, + max_groups_per_commit=3, + ) + finally: + resume_iteration.set() + observer.join(30) self.assertEqual({"num_updated": 8}, stats) + self.assertEqual([(1, 0)], iter_batch_options) + self.assertNotIn("error", observed) + self.assertEqual(base_snapshot_id + 1, observed["snapshot_id"]) latest = table.snapshot_manager().get_latest_snapshot() self.assertEqual(base_snapshot_id + 3, latest.id) incremental_snapshots = [ @@ -187,10 +228,76 @@ def test_incrementally_commits_file_group_windows(self): for chunk in chunks: self.assertEqual(0, got[chunk[1]]) + def test_incremental_commit_failure_drains_later_groups(self): + import importlib + from pypaimon.write.table_commit import StreamTableCommit + + m = importlib.import_module("pypaimon.ray.update_by_row_id") + target = self._create() + chunks = [[start, start + 1] for start in range(10, 50, 10)] + for chunk in chunks: + self._write(target, pa.Table.from_pydict( + {"id": chunk, "name": ["x", "x"], "age": [0, 0]}, + schema=self.pa_schema, + )) + + rid = self._rowid_by_id(target) + src = pa.table( + { + "_ROW_ID": [rid[chunk[0]] for chunk in chunks], + "age": [chunk[0] for chunk in chunks], + }, + schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]), + ) + seen_groups = [] + aborted = [] + commit_calls = [] + original_add_group = m._IncrementalUpdateCommitter.add_group + original_abort = StreamTableCommit.abort + + def record_group(committer, messages, num_updated, row_ids): + seen_groups.append(list(messages)) + return original_add_group( + committer, messages, num_updated, row_ids) + + def fail_commit(_commit, _messages, commit_identifier): + commit_calls.append(commit_identifier) + raise RuntimeError("commit failed") + + def record_abort(commit, messages): + aborted.append(list(messages)) + return original_abort(commit, messages) + + with mock.patch.object( + m._IncrementalUpdateCommitter, "add_group", record_group), \ + mock.patch.object(StreamTableCommit, "commit", fail_commit), \ + mock.patch.object(StreamTableCommit, "abort", record_abort): + with self.assertRaisesRegex(RuntimeError, "commit failed"): + update_by_row_id( + target, + ray.data.from_arrow(src).repartition(4), + self.catalog_options, + update_cols=["age"], + num_partitions=4, + max_groups_per_commit=2, + ) + + self.assertEqual(4, len(seen_groups)) + self.assertEqual([1], commit_calls) + self.assertEqual( + [[message for group in seen_groups[2:] for message in group]], + aborted, + ) + def test_incremental_committer_batches_complete_groups(self): import importlib m = importlib.import_module("pypaimon.ray.update_by_row_id") - recorder = {"commits": [], "aborts": [], "close_calls": 0} + recorder = { + "commits": [], + "ignored": [], + "aborts": [], + "close_calls": 0, + } class FakeCommit: def commit(self, msgs, commit_identifier): @@ -199,6 +306,9 @@ def commit(self, msgs, commit_identifier): def abort(self, msgs): recorder["aborts"].append(list(msgs)) + def ignore_row_id_conflict_for_commit(self, commit_identifier): + recorder["ignored"].append(commit_identifier) + def close(self): recorder["close_calls"] += 1 @@ -222,13 +332,14 @@ def new_stream_write_builder(self): (["group-1", "group-2"], 1), (["group-3"], 2), ], recorder["commits"]) + self.assertEqual([1, 2], recorder["ignored"]) self.assertEqual([], recorder["aborts"]) self.assertEqual(1, recorder["close_calls"]) def test_incremental_committer_aborts_only_uncommitted_groups(self): import importlib m = importlib.import_module("pypaimon.ray.update_by_row_id") - recorder = {"commits": [], "aborts": []} + recorder = {"commits": [], "ignored": [], "aborts": []} class FakeCommit: def commit(self, msgs, commit_identifier): @@ -237,6 +348,9 @@ def commit(self, msgs, commit_identifier): def abort(self, msgs): recorder["aborts"].append(list(msgs)) + def ignore_row_id_conflict_for_commit(self, commit_identifier): + recorder["ignored"].append(commit_identifier) + def close(self): pass @@ -258,19 +372,24 @@ def new_stream_write_builder(self): self.assertEqual([ (["committed-1", "committed-2"], 1), ], recorder["commits"]) + self.assertEqual([1], recorder["ignored"]) self.assertEqual([["pending"]], recorder["aborts"]) - def test_incremental_committer_does_not_abort_uncertain_commit(self): + def test_incremental_committer_drains_after_uncertain_commit(self): import importlib m = importlib.import_module("pypaimon.ray.update_by_row_id") - recorder = {"abort_calls": 0} + recorder = {"commit_calls": 0, "ignored": [], "aborts": []} class FakeCommit: def commit(self, msgs, commit_identifier): + recorder["commit_calls"] += 1 raise RuntimeError("commit outcome is uncertain") def abort(self, msgs): - recorder["abort_calls"] += 1 + recorder["aborts"].append(list(msgs)) + + def ignore_row_id_conflict_for_commit(self, commit_identifier): + recorder["ignored"].append(commit_identifier) def close(self): pass @@ -283,13 +402,59 @@ class FakeTable: def new_stream_write_builder(self): return FakeBuilder() - committer = m._IncrementalUpdateCommitter(FakeTable(), 1) + committer = m._IncrementalUpdateCommitter(FakeTable(), 2) + committer.add_group(["uncertain-1"], 1, []) + committer.add_group(["uncertain-2"], 1, []) + committer.add_group(["pending-1"], 1, []) + committer.add_group(["pending-2"], 1, []) with self.assertRaisesRegex(RuntimeError, "outcome is uncertain"): - committer.add_group(["possibly-committed"], 1, []) + committer.finish() committer.abort_pending() committer.close() - self.assertEqual(0, recorder["abort_calls"]) + self.assertEqual(1, recorder["commit_calls"]) + self.assertEqual([], recorder["ignored"]) + self.assertEqual([["pending-1", "pending-2"]], recorder["aborts"]) + + def test_incremental_committer_aborts_all_after_new_commit_failure(self): + import importlib + m = importlib.import_module("pypaimon.ray.update_by_row_id") + recorder = {"aborts": [], "close_calls": 0} + + class AbortCommit: + def abort(self, msgs): + recorder["aborts"].append(list(msgs)) + + def close(self): + recorder["close_calls"] += 1 + + class FailingBuilder: + def new_commit(self): + raise RuntimeError("new_commit failed") + + class AbortBuilder: + def new_commit(self): + return AbortCommit() + + class FakeTable: + def new_stream_write_builder(self): + return FailingBuilder() + + def new_batch_write_builder(self): + return AbortBuilder() + + committer = m._IncrementalUpdateCommitter(FakeTable(), 2) + for group in ("group-1", "group-2", "group-3"): + committer.add_group([group], 1, []) + with self.assertRaisesRegex(RuntimeError, "new_commit failed"): + committer.finish() + committer.abort_pending() + + self.assertEqual( + [["group-1", "group-2", "group-3"]], + recorder["aborts"], + ) + self.assertEqual(1, recorder["close_calls"]) def test_rejects_invalid_max_groups_per_commit(self): target = self._create() diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py b/paimon-python/pypaimon/tests/write/conflict_detection_test.py index 62889370f56d..b45d9b50bfe6 100644 --- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py +++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py @@ -314,18 +314,23 @@ def test_delete_entry_missing_from_base_conflicts(self): class _FakeSnapshot: - def __init__(self, snapshot_id, commit_kind, next_row_id=None): + def __init__(self, snapshot_id, commit_kind, next_row_id=None, + commit_user=None, commit_identifier=None): self.id = snapshot_id self.commit_kind = commit_kind self.next_row_id = next_row_id + self.commit_user = commit_user + self.commit_identifier = commit_identifier class _FakeSnapshotManager: def __init__(self, snapshots): self._by_id = {s.id: s for s in snapshots} + self.requests = [] def get_snapshot_by_id(self, snapshot_id): + self.requests.append(snapshot_id) return self._by_id.get(snapshot_id) @@ -334,11 +339,16 @@ class _FakeCommitScanner: def __init__(self, entries_by_snapshot_id, raw_entries_by_snapshot_id=None): self._by_id = entries_by_snapshot_id self._raw_by_id = raw_entries_by_snapshot_id or {} + self.entry_calls = [] + self.raw_entry_calls = [] def read_incremental_entries_from_changed_partitions(self, snapshot, _): - return self._by_id.get(snapshot.id, []) + self.entry_calls.append(snapshot.id) + return self._by_id.get( + snapshot.id, self._raw_by_id.get(snapshot.id, [])) def read_incremental_raw_entries_from_changed_partitions(self, snapshot, _): + self.raw_entry_calls.append(snapshot.id) return self._raw_by_id.get(snapshot.id, self._by_id.get(snapshot.id, [])) @@ -417,6 +427,81 @@ def test_compact_no_conflict_when_no_matching_delete(self): self.assertIsNone( detection.check_row_id_from_snapshot(compact_snap, delta)) + def test_own_history_is_discovered_once_and_not_scanned(self): + base = _FakeSnapshot(1, "APPEND", next_row_id=1000) + own = [ + _FakeSnapshot( + snapshot_id, + "APPEND", + next_row_id=1000, + commit_user="incremental", + commit_identifier=snapshot_id - 1, + ) + for snapshot_id in range(2, 9) + ] + manager = _FakeSnapshotManager([base] + own) + scanner = _FakeCommitScanner({}) + detection = ConflictDetection( + data_evolution_enabled=True, + snapshot_manager=manager, + manifest_list_manager=None, + table=_FakeTable(_FakeSchemaManager([_DEFAULT_SCHEMA])), + commit_scanner=scanner, + ) + detection.set_row_id_check_from_snapshot(1) + for snapshot in own: + detection.ignore_row_id_commit( + snapshot.commit_user, snapshot.commit_identifier) + + delta = self._blob_delta() + for latest in own: + self.assertIsNone( + detection.check_row_id_from_snapshot(latest, delta)) + + self.assertEqual([], scanner.entry_calls) + for snapshot_id in range(2, 9): + self.assertEqual(1, manager.requests.count(snapshot_id)) + + def test_cached_external_history_checks_later_windows(self): + base = _FakeSnapshot(1, "APPEND", next_row_id=1000) + external = _FakeSnapshot( + 2, "APPEND", next_row_id=1000, + commit_user="external", commit_identifier=1) + own = _FakeSnapshot( + 3, "APPEND", next_row_id=1000, + commit_user="incremental", commit_identifier=1) + external_entry = _make_entry( + "external.parquet", + first_row_id=100, + row_count=50, + write_cols=["col_a"], + ) + manager = _FakeSnapshotManager([base, external, own]) + scanner = _FakeCommitScanner({2: [external_entry]}) + detection = ConflictDetection( + data_evolution_enabled=True, + snapshot_manager=manager, + manifest_list_manager=None, + table=_FakeTable(_FakeSchemaManager([_DEFAULT_SCHEMA])), + commit_scanner=scanner, + ) + detection.set_row_id_check_from_snapshot(1) + detection.ignore_row_id_commit("incremental", 1) + + self.assertIsNone( + detection.check_row_id_from_snapshot(own, self._blob_delta())) + later_delta = [_make_entry( + "later.blob", + first_row_id=100, + row_count=50, + write_cols=["col_a"], + )] + result = detection.check_row_id_from_snapshot(own, later_delta) + + self.assertIsNotNone(result) + self.assertIn("updating the same file", str(result)) + self.assertEqual([2, 2], scanner.entry_calls) + class TestRowIdColumnConflictChecker(unittest.TestCase): diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py b/paimon-python/pypaimon/write/commit/conflict_detection.py index 30615ec893a5..d061851c2c04 100644 --- a/paimon-python/pypaimon/write/commit/conflict_detection.py +++ b/paimon-python/pypaimon/write/commit/conflict_detection.py @@ -159,6 +159,10 @@ def __init__(self, data_evolution_enabled, snapshot_manager, self.manifest_list_manager = manifest_list_manager self.table = table self._row_id_check_from_snapshot = None + self._row_id_ignored_commits = set() + self._row_id_history_base_snapshot = None + self._row_id_history_cursor = None + self._row_id_external_snapshots = [] self.commit_scanner = commit_scanner def should_be_overwrite_commit(self, append_file_entries=None, append_index_files=None): @@ -173,6 +177,44 @@ def should_be_overwrite_commit(self, append_file_entries=None, append_index_file def has_row_id_check_from_snapshot(self): return self._row_id_check_from_snapshot is not None + def set_row_id_check_from_snapshot(self, snapshot_id): + if self._row_id_check_from_snapshot == snapshot_id: + return + self._row_id_check_from_snapshot = snapshot_id + self._row_id_history_base_snapshot = None + self._row_id_history_cursor = None + self._row_id_external_snapshots = [] + + def ignore_row_id_commit(self, commit_user, commit_identifier): + """Exclude one known-disjoint commit from later checks.""" + self._row_id_ignored_commits.add((commit_user, commit_identifier)) + + def _row_id_history_snapshots(self, latest_snapshot): + """Cache external history while skipping registered disjoint windows.""" + base_snapshot = self._row_id_check_from_snapshot + if (self._row_id_history_base_snapshot != base_snapshot + or self._row_id_history_cursor is None + or latest_snapshot.id < self._row_id_history_cursor): + self._row_id_history_base_snapshot = base_snapshot + self._row_id_history_cursor = base_snapshot + self._row_id_external_snapshots = [] + + for snapshot_id in range( + self._row_id_history_cursor + 1, + latest_snapshot.id + 1): + snapshot = self.snapshot_manager.get_snapshot_by_id(snapshot_id) + if snapshot is None: + continue + identity = ( + getattr(snapshot, "commit_user", None), + getattr(snapshot, "commit_identifier", None), + ) + if identity not in self._row_id_ignored_commits: + self._row_id_external_snapshots.append(snapshot) + + self._row_id_history_cursor = latest_snapshot.id + return self._row_id_external_snapshots + @staticmethod def has_global_index_additions(index_entries=None): return bool(ConflictDetection.global_index_file_additions(index_entries)) @@ -576,13 +618,7 @@ def check_row_id_from_snapshot(self, latest_snapshot, commit_entries): delta_signatures.append( (DataFileMeta.is_blob_file(f.file_name), r.from_, r.to)) - for snapshot_id in range( - self._row_id_check_from_snapshot + 1, - latest_snapshot.id + 1): - snapshot = self.snapshot_manager.get_snapshot_by_id(snapshot_id) - if snapshot is None: - continue - + for snapshot in self._row_id_history_snapshots(latest_snapshot): if snapshot.commit_kind == "COMPACT": err = self._compact_conflicts_with_delta( snapshot, delta_signatures, column_checker, commit_entries) diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 1147d7385866..0105cfbcdf5c 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -132,7 +132,9 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int): valid_snapshots = [msg.check_from_snapshot for msg in commit_messages if msg.check_from_snapshot != -1] if valid_snapshots: - self.conflict_detection._row_id_check_from_snapshot = min(valid_snapshots) + self.conflict_detection.set_row_id_check_from_snapshot( + min(valid_snapshots) + ) logger.info( "Ready to commit to table %s, number of commit messages: %d", diff --git a/paimon-python/pypaimon/write/table_commit.py b/paimon-python/pypaimon/write/table_commit.py index 4a758cb17662..e1a77421c667 100644 --- a/paimon-python/pypaimon/write/table_commit.py +++ b/paimon-python/pypaimon/write/table_commit.py @@ -91,6 +91,12 @@ def _commit(self, commit_messages: List[CommitMessage], commit_identifier: int = def abort(self, commit_messages: List[CommitMessage]): self.file_store_commit.abort(commit_messages) + def ignore_row_id_conflict_for_commit(self, commit_identifier: int) -> None: + """Exclude a known-disjoint commit from later row-id checks.""" + self.file_store_commit.conflict_detection.ignore_row_id_commit( + self.commit_user, commit_identifier + ) + def close(self): self.file_store_commit.close() From 9182b646e1fceb3fca3f3e76f1edc1be560ea75e Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 23 Jul 2026 21:17:27 +0800 Subject: [PATCH 3/8] [python][ray] Avoid double unwrapping commit errors --- .../pypaimon/ray/update_by_row_id.py | 7 +++--- .../tests/ray_update_by_row_id_test.py | 23 +++++++++++++++++++ 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index a2543299de6d..899d9b67e17b 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -165,10 +165,7 @@ def _project_cast(batch: pa.Table) -> pa.Table: msgs, num_updated, _ = distributed_update_apply( update_ds, table, update_cols, **apply_kwargs ) - if incremental_committer is None: - if msgs: - _commit_update_messages(table, msgs) - else: + if incremental_committer is not None: incremental_committer.finish() except Exception as e: if incremental_committer is not None: @@ -178,6 +175,8 @@ def _project_cast(batch: pa.Table) -> pa.Table: finally: if incremental_committer is not None: incremental_committer.close() + if incremental_committer is None and msgs: + _commit_update_messages(table, msgs) return {"num_updated": num_updated} diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py index 661db5e0b46e..6b859245a9e6 100644 --- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py @@ -524,6 +524,29 @@ def test_commit_failure_does_not_abort_after_commit_started(self): self.assertEqual(recorder["abort_calls"], 0) self.assertEqual(recorder["close_calls"], 1) + def test_chained_commit_failure_is_unwrapped_once(self): + import importlib + + m = importlib.import_module("pypaimon.ray.update_by_row_id") + retry_error = ValueError("retry failed") + try: + raise RuntimeError("commit failed") from retry_error + except RuntimeError as commit_error: + chained_error = commit_error + + reraise_calls = [] + + def reraise_once(error): + reraise_calls.append(error) + raise ValueError("retry failed") from None + + with mock.patch.object(m, "_reraise_inner", side_effect=reraise_once): + with self.assertRaisesRegex(ValueError, "retry failed"): + self._run_with_fake_commit(commit_error=chained_error) + + self.assertEqual(1, len(reraise_calls)) + self.assertIs(chained_error, reraise_calls[0]) + def test_close_failure_after_success_warns_and_returns_stats(self): close_error = RuntimeError("close failed") From 503828486ef3eb3c10eb24345b134c4a9dbb942f Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 23 Jul 2026 22:07:30 +0800 Subject: [PATCH 4/8] [python][ray] Fix rollback and commit failure handling --- .../pypaimon/ray/update_by_row_id.py | 20 +- .../pypaimon/tests/file_store_commit_test.py | 103 +++++++- .../tests/ray_update_by_row_id_test.py | 230 +++++++++++++++++- .../tests/write/commit_callback_test.py | 26 ++ .../tests/write/conflict_detection_test.py | 93 ++++++- .../write/commit/conflict_detection.py | 46 +++- .../pypaimon/write/file_store_commit.py | 194 ++++++++++----- 7 files changed, 632 insertions(+), 80 deletions(-) diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index 899d9b67e17b..9c59cc023aaf 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -37,6 +37,7 @@ from pypaimon.ray.data_evolution_merge_join import distributed_update_apply from pypaimon.ray.data_evolution_merge_transform import build_update_schema from pypaimon.schema.data_types import is_blob_file_field +from pypaimon.write.file_store_commit import CommitOutcomeUnknownError __all__ = ["update_by_row_id"] @@ -229,18 +230,25 @@ def _commit_pending(self) -> None: self._pending_messages, commit_identifier ) except Exception as error: - # Defer the error; this window's commit outcome is uncertain. - self._uncertain_messages.extend(self._pending_messages) - self._pending_messages = [] - self._pending_groups = 0 + if isinstance(error, CommitOutcomeUnknownError): + self._uncertain_messages.extend(self._pending_messages) + self._pending_messages = [] + self._pending_groups = 0 self._deferred_error = error return - # Later windows are disjoint by _FIRST_ROW_ID. - self._table_commit.ignore_row_id_conflict_for_commit(commit_identifier) + committed_messages = self._pending_messages self._pending_messages = [] self._pending_groups = 0 self._next_commit_identifier += 1 + try: + # Later windows are disjoint by _FIRST_ROW_ID. + self._table_commit.ignore_row_id_conflict_for_commit( + commit_identifier) + except Exception as error: + self._uncertain_messages.extend(committed_messages) + self._deferred_error = error + return logger.info( "Incrementally committed %d update_by_row_id file groups.", group_count, diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index 98c8867fe646..b00e66a768c7 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -24,7 +24,11 @@ from pypaimon.snapshot.snapshot_commit import PartitionStatistics from pypaimon.table.row.generic_row import GenericRow from pypaimon.write.commit_message import CommitMessage -from pypaimon.write.file_store_commit import FileStoreCommit +from pypaimon.write.file_store_commit import ( + CommitOutcomeUnknownError, + FileStoreCommit, + RetryResult, +) @patch('pypaimon.write.file_store_commit.ManifestFileManager') @@ -448,6 +452,103 @@ def test_append_commit_inherits_index_manifest( snapshot_commit.commit.call_args[0][0].index_manifest ) + def test_atomic_exception_has_unknown_outcome( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 0 + file_store_commit.commit_timeout = 1000 + file_store_commit.snapshot_manager.get_latest_snapshot.return_value = None + atomic_error = RuntimeError("atomic commit failed") + file_store_commit._try_commit_once = Mock(return_value=RetryResult( + None, atomic_error, outcome_unknown=True)) + file_store_commit._is_commit_persisted = Mock(return_value=False) + + with self.assertRaises(CommitOutcomeUnknownError) as raised: + file_store_commit._try_commit( + "APPEND", 1, lambda _snapshot: [Mock()]) + + self.assertIs(atomic_error, raised.exception.__cause__) + + def test_atomic_commit_exception_marks_retry_unknown( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + self.mock_table.identifier = 'default.test_table' + self.mock_table.table_schema = Mock(id=7) + self.mock_table.options.row_tracking_enabled.return_value = False + atomic_error = RuntimeError("atomic commit failed") + snapshot_commit = MagicMock() + snapshot_commit.__enter__.return_value = snapshot_commit + snapshot_commit.__exit__.return_value = False + snapshot_commit.commit.side_effect = atomic_error + file_store_commit.snapshot_commit = snapshot_commit + file_store_commit._write_manifest_files = Mock(return_value=[Mock()]) + file_store_commit._generate_partition_statistics = Mock(return_value=[]) + file_store_commit.manifest_list_manager.read_all.return_value = [] + commit_entry = Mock(kind=0) + commit_entry.file = Mock(row_count=1) + + result = file_store_commit._try_commit_once( + retry_result=None, + commit_kind="APPEND", + commit_entries=[commit_entry], + changelog_entries=[], + commit_identifier=1, + latest_snapshot=None, + ) + + self.assertFalse(result.is_success()) + self.assertTrue(result.outcome_unknown) + self.assertIs(atomic_error, result.exception) + + def test_false_commit_has_deterministic_outcome( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 0 + file_store_commit.commit_timeout = 1000 + file_store_commit.snapshot_manager.get_latest_snapshot.return_value = None + file_store_commit._try_commit_once = Mock( + return_value=RetryResult(None)) + + with self.assertRaises(RuntimeError) as raised: + file_store_commit._try_commit( + "APPEND", 1, lambda _snapshot: [Mock()]) + + self.assertNotIsInstance( + raised.exception, CommitOutcomeUnknownError) + + def test_unknown_outcome_survives_later_failure( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 2 + file_store_commit.commit_timeout = 1000 + file_store_commit.snapshot_manager.get_latest_snapshot.return_value = None + file_store_commit._commit_retry_wait = Mock() + atomic_error = RuntimeError("atomic commit failed") + file_store_commit._try_commit_once = Mock(side_effect=[ + RetryResult(None, atomic_error, outcome_unknown=True), + RuntimeError("conflict"), + ]) + file_store_commit._is_commit_persisted = Mock(return_value=False) + + with self.assertRaises(CommitOutcomeUnknownError) as raised: + file_store_commit._try_commit( + "APPEND", 1, lambda _snapshot: [Mock()]) + + self.assertIs(atomic_error, raised.exception.__cause__) + + def test_persisted_identity_resolves_unknown_outcome( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 0 + file_store_commit.commit_timeout = 1000 + file_store_commit.snapshot_manager.get_latest_snapshot.return_value = None + file_store_commit._try_commit_once = Mock(return_value=RetryResult( + None, RuntimeError("response lost"), outcome_unknown=True)) + file_store_commit._is_commit_persisted = Mock(return_value=True) + + file_store_commit._try_commit( + "APPEND", 1, lambda _snapshot: [Mock()]) + def test_null_partition_value( self, mock_manifest_list_manager, mock_manifest_file_manager): from pypaimon.data.timestamp import Timestamp diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py index 6b859245a9e6..b6d99becd2ba 100644 --- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py @@ -230,6 +230,7 @@ def observe_first_window(): def test_incremental_commit_failure_drains_later_groups(self): import importlib + from pypaimon.write.file_store_commit import CommitOutcomeUnknownError from pypaimon.write.table_commit import StreamTableCommit m = importlib.import_module("pypaimon.ray.update_by_row_id") @@ -262,7 +263,7 @@ def record_group(committer, messages, num_updated, row_ids): def fail_commit(_commit, _messages, commit_identifier): commit_calls.append(commit_identifier) - raise RuntimeError("commit failed") + raise CommitOutcomeUnknownError("commit failed") def record_abort(commit, messages): aborted.append(list(messages)) @@ -289,6 +290,148 @@ def record_abort(commit, messages): aborted, ) + def test_incremental_conflict_aborts_output_files(self): + from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER + from pypaimon.write.table_commit import StreamTableCommit + from pypaimon.write.table_update_by_row_id import TableUpdateByRowId + + target = self._create() + self._write(target, pa.Table.from_pydict( + {"id": [1], "name": ["a"], "age": [0]}, + schema=self.pa_schema, + )) + table = self.catalog.get_table(target) + row_id = self._rowid_by_id(target)[1] + update_schema = pa.schema([ + ("_ROW_ID", pa.int64()), + ("age", pa.int32()), + ]) + source = pa.table( + {"_ROW_ID": [row_id], "age": [300]}, + schema=update_schema, + ) + output_paths = [] + original_commit = StreamTableCommit.commit + + def commit_after_concurrent_update( + stream_commit, messages, commit_identifier): + output_paths.extend( + file.file_path + for message in messages + for file in message.new_files + ) + updater = TableUpdateByRowId( + table, "concurrent", BATCH_COMMIT_IDENTIFIER) + concurrent_messages = updater.update_columns( + pa.table( + {"_ROW_ID": [row_id], "age": [999]}, + schema=update_schema, + ), + ["age"], + ) + concurrent_commit = table.new_batch_write_builder().new_commit() + try: + concurrent_commit.commit(concurrent_messages) + finally: + concurrent_commit.close() + return original_commit( + stream_commit, messages, commit_identifier) + + with mock.patch.object( + StreamTableCommit, "commit", commit_after_concurrent_update): + with self.assertRaisesRegex( + RuntimeError, "updating the same file"): + update_by_row_id( + target, + ray.data.from_arrow(source), + self.catalog_options, + update_cols=["age"], + max_groups_per_commit=1, + ) + + self.assertTrue(output_paths) + self.assertTrue(all( + not table.file_io.exists(path) for path in output_paths + )) + self.assertEqual([999], self._read(target)["age"].to_pylist()) + + def test_rollback_reused_snapshot_id_detects_conflict(self): + from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER + from pypaimon.write.table_commit import StreamTableCommit + from pypaimon.write.table_update_by_row_id import TableUpdateByRowId + + target = self._create() + for row_id in range(1, 4): + self._write(target, pa.Table.from_pydict( + {"id": [row_id], "name": ["a"], "age": [0]}, + schema=self.pa_schema, + )) + table = self.catalog.get_table(target) + base_snapshot_id = table.snapshot_manager().get_latest_snapshot().id + row_ids = self._rowid_by_id(target) + update_schema = pa.schema([ + ("_ROW_ID", pa.int64()), + ("age", pa.int32()), + ]) + source = pa.table( + { + "_ROW_ID": [row_ids[row_id] for row_id in range(1, 4)], + "age": [100, 200, 300], + }, + schema=update_schema, + ) + original_commit = StreamTableCommit.commit + commit_calls = [] + + def commit_then_rebuild_history( + stream_commit, messages, commit_identifier): + commit_calls.append(commit_identifier) + result = original_commit( + stream_commit, messages, commit_identifier) + if commit_identifier == 2: + table.rollback_to(base_snapshot_id) + updater = TableUpdateByRowId( + table, "concurrent", BATCH_COMMIT_IDENTIFIER) + concurrent_messages = updater.update_columns( + pa.table( + { + "_ROW_ID": [ + row_ids[row_id] for row_id in range(1, 4) + ], + "age": [999, 999, 999], + }, + schema=update_schema, + ), + ["age"], + ) + concurrent_commit = ( + table.new_batch_write_builder().new_commit() + ) + try: + concurrent_commit.commit(concurrent_messages) + finally: + concurrent_commit.close() + return result + + with mock.patch.object( + StreamTableCommit, "commit", commit_then_rebuild_history): + with self.assertRaisesRegex( + RuntimeError, "updating the same file"): + update_by_row_id( + target, + ray.data.from_arrow(source).repartition(3), + self.catalog_options, + update_cols=["age"], + num_partitions=3, + max_groups_per_commit=1, + ) + + self.assertEqual([1, 2, 3], commit_calls) + self.assertEqual( + [999, 999, 999], + self._read(target).sort_by("id")["age"].to_pylist(), + ) + def test_incremental_committer_batches_complete_groups(self): import importlib m = importlib.import_module("pypaimon.ray.update_by_row_id") @@ -377,13 +520,16 @@ def new_stream_write_builder(self): def test_incremental_committer_drains_after_uncertain_commit(self): import importlib + from pypaimon.write.file_store_commit import CommitOutcomeUnknownError + m = importlib.import_module("pypaimon.ray.update_by_row_id") recorder = {"commit_calls": 0, "ignored": [], "aborts": []} class FakeCommit: def commit(self, msgs, commit_identifier): recorder["commit_calls"] += 1 - raise RuntimeError("commit outcome is uncertain") + raise CommitOutcomeUnknownError( + "commit outcome is uncertain") def abort(self, msgs): recorder["aborts"].append(list(msgs)) @@ -416,6 +562,86 @@ def new_stream_write_builder(self): self.assertEqual([], recorder["ignored"]) self.assertEqual([["pending-1", "pending-2"]], recorder["aborts"]) + def test_incremental_committer_aborts_deterministic_failure(self): + import importlib + + m = importlib.import_module("pypaimon.ray.update_by_row_id") + recorder = {"commit_calls": 0, "aborts": []} + + class FakeCommit: + def commit(self, msgs, commit_identifier): + recorder["commit_calls"] += 1 + raise RuntimeError("conflict") + + def abort(self, msgs): + recorder["aborts"].append(list(msgs)) + + def close(self): + pass + + class FakeBuilder: + def new_commit(self): + return FakeCommit() + + class FakeTable: + def new_stream_write_builder(self): + return FakeBuilder() + + committer = m._IncrementalUpdateCommitter(FakeTable(), 2) + for group in ("failed-1", "failed-2", "pending-1", "pending-2"): + committer.add_group([group], 1, []) + with self.assertRaisesRegex(RuntimeError, "conflict"): + committer.finish() + committer.abort_pending() + committer.close() + + self.assertEqual(1, recorder["commit_calls"]) + self.assertEqual([[ + "failed-1", "failed-2", "pending-1", "pending-2" + ]], recorder["aborts"]) + + def test_incremental_committer_protects_committed_window(self): + import importlib + + m = importlib.import_module("pypaimon.ray.update_by_row_id") + recorder = {"commits": [], "aborts": []} + + class FakeCommit: + def commit(self, msgs, commit_identifier): + recorder["commits"].append(list(msgs)) + + def abort(self, msgs): + recorder["aborts"].append(list(msgs)) + + def ignore_row_id_conflict_for_commit(self, commit_identifier): + raise RuntimeError("ignore failed") + + def close(self): + pass + + class FakeBuilder: + def new_commit(self): + return FakeCommit() + + class FakeTable: + def new_stream_write_builder(self): + return FakeBuilder() + + committer = m._IncrementalUpdateCommitter(FakeTable(), 2) + for group in ("committed-1", "committed-2", "pending-1", "pending-2"): + committer.add_group([group], 1, []) + with self.assertRaisesRegex(RuntimeError, "ignore failed"): + committer.finish() + committer.abort_pending() + committer.close() + + self.assertEqual([[ + "committed-1", "committed-2" + ]], recorder["commits"]) + self.assertEqual([[ + "pending-1", "pending-2" + ]], recorder["aborts"]) + def test_incremental_committer_aborts_all_after_new_commit_failure(self): import importlib m = importlib.import_module("pypaimon.ray.update_by_row_id") diff --git a/paimon-python/pypaimon/tests/write/commit_callback_test.py b/paimon-python/pypaimon/tests/write/commit_callback_test.py index 4e02bf32f67a..5013d40c8026 100644 --- a/paimon-python/pypaimon/tests/write/commit_callback_test.py +++ b/paimon-python/pypaimon/tests/write/commit_callback_test.py @@ -24,6 +24,7 @@ from pypaimon import CatalogFactory, Schema from pypaimon.write.commit_callback import CommitCallback, CommitCallbackContext +from pypaimon.write.file_store_commit import CommitOutcomeUnknownError class RecordingCallback(CommitCallback): @@ -169,6 +170,31 @@ def test_callback_not_invoked_when_no_data(self): table_write.close() table_commit.close() + def test_callback_failure_keeps_committed_snapshot(self): + class FailingCallback(CommitCallback): + def call(self, context): + raise RuntimeError("callback failed") + + table = self._create_table('test_callback_failure') + write_builder = table.new_batch_write_builder() + table_write = write_builder.new_write() + table_commit = write_builder.new_commit() + table_commit.add_commit_callback(FailingCallback()) + table_write.write_arrow(pa.Table.from_pydict({ + 'id': [1], + 'name': ['a'], + 'dt': ['p1'], + }, schema=self.pa_schema)) + + with self.assertRaisesRegex( + CommitOutcomeUnknownError, "callback failed"): + table_commit.commit(table_write.prepare_commit()) + + self.assertEqual( + 1, table.snapshot_manager().get_latest_snapshot().id) + table_write.close() + table_commit.close() + def test_stream_commit_callback_multiple_rounds(self): table = self._create_table('test_stream_callback') write_builder = table.new_stream_write_builder() diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py b/paimon-python/pypaimon/tests/write/conflict_detection_test.py index b45d9b50bfe6..e99220ec12f6 100644 --- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py +++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py @@ -315,12 +315,14 @@ def test_delete_entry_missing_from_base_conflicts(self): class _FakeSnapshot: def __init__(self, snapshot_id, commit_kind, next_row_id=None, - commit_user=None, commit_identifier=None): + commit_user=None, commit_identifier=None, + delta_manifest_list=None): self.id = snapshot_id self.commit_kind = commit_kind self.next_row_id = next_row_id self.commit_user = commit_user self.commit_identifier = commit_identifier + self.delta_manifest_list = delta_manifest_list class _FakeSnapshotManager: @@ -427,7 +429,7 @@ def test_compact_no_conflict_when_no_matching_delete(self): self.assertIsNone( detection.check_row_id_from_snapshot(compact_snap, delta)) - def test_own_history_is_discovered_once_and_not_scanned(self): + def test_own_history_is_not_scanned(self): base = _FakeSnapshot(1, "APPEND", next_row_id=1000) own = [ _FakeSnapshot( @@ -459,8 +461,9 @@ def test_own_history_is_discovered_once_and_not_scanned(self): detection.check_row_id_from_snapshot(latest, delta)) self.assertEqual([], scanner.entry_calls) - for snapshot_id in range(2, 9): - self.assertEqual(1, manager.requests.count(snapshot_id)) + for snapshot_id in range(2, 8): + self.assertEqual(2, manager.requests.count(snapshot_id)) + self.assertEqual(1, manager.requests.count(8)) def test_cached_external_history_checks_later_windows(self): base = _FakeSnapshot(1, "APPEND", next_row_id=1000) @@ -502,6 +505,88 @@ def test_cached_external_history_checks_later_windows(self): self.assertIn("updating the same file", str(result)) self.assertEqual([2, 2], scanner.entry_calls) + def test_reused_snapshot_id_invalidates_cached_history(self): + base = _FakeSnapshot(1, "APPEND", next_row_id=1000) + own = _FakeSnapshot( + 2, "APPEND", next_row_id=1000, + commit_user="incremental", commit_identifier=1, + delta_manifest_list="own-manifest") + manager = _FakeSnapshotManager([base, own]) + scanner = _FakeCommitScanner({}) + detection = ConflictDetection( + data_evolution_enabled=True, + snapshot_manager=manager, + manifest_list_manager=None, + table=_FakeTable(_FakeSchemaManager([_DEFAULT_SCHEMA])), + commit_scanner=scanner, + ) + detection.set_row_id_check_from_snapshot(1) + detection.ignore_row_id_commit("incremental", 1) + + delta = self._blob_delta() + self.assertIsNone(detection.check_row_id_from_snapshot(own, delta)) + + replacement = _FakeSnapshot( + 2, "APPEND", next_row_id=1000, + commit_user="external", commit_identifier=1, + delta_manifest_list="external-manifest") + manager._by_id[2] = replacement + scanner._by_id[2] = [_make_entry( + "external.parquet", + first_row_id=0, + row_count=51, + write_cols=["col_a"], + )] + + result = detection.check_row_id_from_snapshot(replacement, delta) + + self.assertIsNotNone(result) + self.assertIn("updating the same file", str(result)) + self.assertEqual([2], scanner.entry_calls) + + def test_rebuilt_history_past_cursor_invalidates_cache(self): + base = _FakeSnapshot(1, "APPEND", next_row_id=1000) + own = _FakeSnapshot( + 2, "APPEND", next_row_id=1000, + commit_user="incremental", commit_identifier=1, + delta_manifest_list="own-manifest") + manager = _FakeSnapshotManager([base, own]) + scanner = _FakeCommitScanner({}) + detection = ConflictDetection( + data_evolution_enabled=True, + snapshot_manager=manager, + manifest_list_manager=None, + table=_FakeTable(_FakeSchemaManager([_DEFAULT_SCHEMA])), + commit_scanner=scanner, + ) + detection.set_row_id_check_from_snapshot(1) + detection.ignore_row_id_commit("incremental", 1) + + delta = self._blob_delta() + self.assertIsNone(detection.check_row_id_from_snapshot(own, delta)) + + replacement = _FakeSnapshot( + 2, "APPEND", next_row_id=1000, + commit_user="external", commit_identifier=1, + delta_manifest_list="external-manifest") + latest = _FakeSnapshot( + 3, "APPEND", next_row_id=1000, + commit_user="external", commit_identifier=2, + delta_manifest_list="latest-manifest") + manager._by_id.update({2: replacement, 3: latest}) + scanner._by_id[2] = [_make_entry( + "external.parquet", + first_row_id=0, + row_count=51, + write_cols=["col_a"], + )] + + result = detection.check_row_id_from_snapshot(latest, delta) + + self.assertIsNotNone(result) + self.assertIn("updating the same file", str(result)) + self.assertEqual([2], scanner.entry_calls) + class TestRowIdColumnConflictChecker(unittest.TestCase): diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py b/paimon-python/pypaimon/write/commit/conflict_detection.py index d061851c2c04..0c6bca856cb7 100644 --- a/paimon-python/pypaimon/write/commit/conflict_detection.py +++ b/paimon-python/pypaimon/write/commit/conflict_detection.py @@ -162,6 +162,7 @@ def __init__(self, data_evolution_enabled, snapshot_manager, self._row_id_ignored_commits = set() self._row_id_history_base_snapshot = None self._row_id_history_cursor = None + self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] self.commit_scanner = commit_scanner @@ -181,8 +182,13 @@ def set_row_id_check_from_snapshot(self, snapshot_id): if self._row_id_check_from_snapshot == snapshot_id: return self._row_id_check_from_snapshot = snapshot_id + self.reset_row_id_history() + + def reset_row_id_history(self): + """Discard cached row-id conflict history.""" self._row_id_history_base_snapshot = None self._row_id_history_cursor = None + self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] def ignore_row_id_commit(self, commit_user, commit_identifier): @@ -192,11 +198,27 @@ def ignore_row_id_commit(self, commit_user, commit_identifier): def _row_id_history_snapshots(self, latest_snapshot): """Cache external history while skipping registered disjoint windows.""" base_snapshot = self._row_id_check_from_snapshot - if (self._row_id_history_base_snapshot != base_snapshot - or self._row_id_history_cursor is None - or latest_snapshot.id < self._row_id_history_cursor): + reset_history = ( + self._row_id_history_base_snapshot != base_snapshot + or self._row_id_history_cursor is None + or latest_snapshot.id < self._row_id_history_cursor + ) + if not reset_history: + cursor_snapshot = ( + latest_snapshot + if latest_snapshot.id == self._row_id_history_cursor + else self.snapshot_manager.get_snapshot_by_id( + self._row_id_history_cursor) + ) + reset_history = ( + self._snapshot_identity(cursor_snapshot) + != self._row_id_history_cursor_identity + ) + + if reset_history: self._row_id_history_base_snapshot = base_snapshot self._row_id_history_cursor = base_snapshot + self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] for snapshot_id in range( @@ -213,8 +235,26 @@ def _row_id_history_snapshots(self, latest_snapshot): self._row_id_external_snapshots.append(snapshot) self._row_id_history_cursor = latest_snapshot.id + self._row_id_history_cursor_identity = self._snapshot_identity( + latest_snapshot) return self._row_id_external_snapshots + @staticmethod + def _snapshot_identity(snapshot): + if snapshot is None: + return None + return ( + snapshot.id, + getattr(snapshot, "commit_user", None), + getattr(snapshot, "commit_identifier", None), + getattr(snapshot, "commit_kind", None), + getattr(snapshot, "time_millis", None), + getattr(snapshot, "base_manifest_list", None), + getattr(snapshot, "delta_manifest_list", None), + getattr(snapshot, "changelog_manifest_list", None), + getattr(snapshot, "index_manifest", None), + ) + @staticmethod def has_global_index_additions(index_entries=None): return bool(ConflictDetection.global_index_file_additions(index_entries)) diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 0105cfbcdf5c..fea0ca8ec3a0 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -55,6 +55,10 @@ def is_success(self) -> bool: raise NotImplementedError +class CommitOutcomeUnknownError(RuntimeError): + """Raised when commit messages may already be in a snapshot.""" + + class SuccessResult(CommitResult): """Result indicating successful commit.""" @@ -65,9 +69,11 @@ def is_success(self) -> bool: class RetryResult(CommitResult): def __init__(self, latest_snapshot, exception: Optional[Exception] = None, - base_data_files: Optional[List[ManifestEntry]] = None): + base_data_files: Optional[List[ManifestEntry]] = None, + outcome_unknown: bool = False): self.latest_snapshot = latest_snapshot self.exception = exception + self.outcome_unknown = outcome_unknown # 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 @@ -295,73 +301,110 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, retry_count = 0 retry_result = None + outcome_unknown = False + outcome_unknown_cause = None start_time_ms = int(time.time() * 1000) while True: - latest_snapshot = self.snapshot_manager.get_latest_snapshot() - commit_entries = commit_entries_plan(latest_snapshot) - - # No entries to commit (e.g. drop_partitions with no matching data): skip commit - # to avoid creating manifest/snapshot with empty partition_stats (causes read errors). - if not commit_entries and not index_deletes and not index_adds: - break - - result = self._try_commit_once( - retry_result=retry_result, - commit_kind=commit_kind, - commit_entries=commit_entries, - changelog_entries=changelog_entries or [], - commit_identifier=commit_identifier, - latest_snapshot=latest_snapshot, - detect_conflicts=detect_conflicts, - allow_rollback=allow_rollback, - index_deletes=index_deletes, - index_adds=index_adds, - ) + try: + latest_snapshot = self.snapshot_manager.get_latest_snapshot() + commit_entries = commit_entries_plan(latest_snapshot) + + # Skip commits with no changes. + if not commit_entries and not index_deletes and not index_adds: + if outcome_unknown: + if self._is_commit_persisted( + commit_identifier, commit_kind): + return + raise CommitOutcomeUnknownError( + "Atomic commit outcome is unknown." + ) from outcome_unknown_cause + break + + result = self._try_commit_once( + retry_result=retry_result, + commit_kind=commit_kind, + commit_entries=commit_entries, + changelog_entries=changelog_entries or [], + commit_identifier=commit_identifier, + latest_snapshot=latest_snapshot, + detect_conflicts=detect_conflicts, + allow_rollback=allow_rollback, + index_deletes=index_deletes, + index_adds=index_adds, + ) - if result.is_success(): - commit_duration_ms = int(time.time() * 1000) - start_time_ms - if commit_kind == "OVERWRITE": - logger.info( - "Finished overwrite to table %s, duration %d ms", - self.table.identifier, - commit_duration_ms, + if result.is_success(): + commit_duration_ms = int(time.time() * 1000) - start_time_ms + if commit_kind == "OVERWRITE": + logger.info( + "Finished overwrite to table %s, duration %d ms", + self.table.identifier, + commit_duration_ms, + ) + else: + logger.info( + "Finished commit to table %s, duration %d ms", + self.table.identifier, + commit_duration_ms, + ) + break + + retry_result = result + outcome_unknown = outcome_unknown or result.outcome_unknown + if result.outcome_unknown and outcome_unknown_cause is None: + outcome_unknown_cause = result.exception + + elapsed_ms = int(time.time() * 1000) - start_time_ms + if (elapsed_ms > self.commit_timeout + or retry_count >= self.commit_max_retries): + if commit_kind == "OVERWRITE": + logger.info( + "Finished (Uncertain of success) overwrite to " + "table %s, duration %d ms", + self.table.identifier, + elapsed_ms, + ) + else: + logger.info( + "Finished (Uncertain of success) commit to table " + "%s, duration %d ms", + self.table.identifier, + elapsed_ms, + ) + error_msg = ( + f"Commit failed " + f"{latest_snapshot.id + 1 if latest_snapshot else 1} " + f"after {elapsed_ms} millis with {retry_count} retries, " + "there maybe exist commit conflicts between multiple jobs." ) - else: - logger.info( - "Finished commit to table %s, duration %d ms", - self.table.identifier, - commit_duration_ms, + if (outcome_unknown and self._is_commit_persisted( + commit_identifier, commit_kind)): + return + error_type = ( + CommitOutcomeUnknownError + if outcome_unknown else RuntimeError ) - break - - retry_result = result - - elapsed_ms = int(time.time() * 1000) - start_time_ms - if elapsed_ms > self.commit_timeout or retry_count >= self.commit_max_retries: - if commit_kind == "OVERWRITE": - logger.info( - "Finished (Uncertain of success) overwrite to table %s, duration %d ms", - self.table.identifier, - elapsed_ms, + error_cause = ( + outcome_unknown_cause + if outcome_unknown else retry_result.exception ) - else: - logger.info( - "Finished (Uncertain of success) commit to table %s, duration %d ms", - self.table.identifier, - elapsed_ms, + if error_cause: + raise error_type(error_msg) from error_cause + raise error_type(error_msg) + + self._commit_retry_wait(retry_count) + retry_count += 1 + except CommitOutcomeUnknownError: + raise + except Exception as error: + if outcome_unknown: + if self._is_commit_persisted( + commit_identifier, commit_kind): + return + raise CommitOutcomeUnknownError(str(error)) from ( + outcome_unknown_cause or error ) - error_msg = ( - f"Commit failed {latest_snapshot.id + 1 if latest_snapshot else 1} " - f"after {elapsed_ms} millis with {retry_count} retries, " - f"there maybe exist commit conflicts between multiple jobs." - ) - if retry_result.exception: - raise RuntimeError(error_msg) from retry_result.exception - else: - raise RuntimeError(error_msg) - - self._commit_retry_wait(retry_count) - retry_count += 1 + raise def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str, commit_entries: List[ManifestEntry], @@ -424,6 +467,7 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str if self.rollback.try_to_rollback(latest_snapshot): # Rolled back: base/snapshot no longer valid; next attempt # re-scans from scratch (matches Java RollbackRetryResult). + self.conflict_detection.reset_row_id_history() return RetryResult(None, conflict_exception) raise conflict_exception @@ -539,7 +583,12 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str 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, base_data_files=base_data_files) + return RetryResult( + latest_snapshot, + e, + base_data_files=base_data_files, + outcome_unknown=True, + ) logger.info( "Successfully commit snapshot %d to table %s by user %s " @@ -557,8 +606,13 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str commit_entries=commit_entries, identifier=commit_identifier, ) - for callback in self.commit_callbacks: - callback.call(context) + try: + for callback in self.commit_callbacks: + callback.call(context) + except Exception as error: + raise CommitOutcomeUnknownError( + "Snapshot committed, but a commit callback failed." + ) from error return SuccessResult() @@ -584,6 +638,18 @@ def _is_duplicate_commit(self, retry_result, latest_snapshot, commit_identifier, return True return False + def _is_commit_persisted(self, commit_identifier, commit_kind) -> bool: + try: + snapshots = self.snapshot_manager.list_snapshots() + return any( + snapshot.commit_user == self.commit_user + and snapshot.commit_identifier == commit_identifier + and snapshot.commit_kind == commit_kind + for snapshot in snapshots + ) + except Exception: + return False + def _create_dynamic_partition_filter(self, commit_messages: List[CommitMessage]): """Build a partition filter from the unique partitions present in commit_messages.""" predicate_builder = PredicateBuilder(self.table.partition_keys_fields) From e90de1ceb8678ccd9fab0fb3b47c5fef4ef4ec3b Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 23 Jul 2026 22:30:07 +0800 Subject: [PATCH 5/8] [python] Polish incremental commit comments --- paimon-python/pypaimon/ray/data_evolution_merge_join.py | 2 +- paimon-python/pypaimon/ray/update_by_row_id.py | 2 +- paimon-python/pypaimon/write/commit/conflict_detection.py | 6 +++--- paimon-python/pypaimon/write/file_store_commit.py | 3 +-- paimon-python/pypaimon/write/table_commit.py | 2 +- 5 files changed, 7 insertions(+), 8 deletions(-) diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index 3eaf16581fcc..c300a1cd87fe 100644 --- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py +++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py @@ -608,7 +608,7 @@ def _apply_group(group: pa.Table) -> pa.Table: action_row_ids = [] iter_batches_kwargs = {"batch_format": "pyarrow"} if on_group_result is not None: - # Yield ready groups without waiting for batch fill or prefetch. + # Yield each group immediately. iter_batches_kwargs["batch_size"] = 1 iter_batches_kwargs["prefetch_batches"] = 0 for batch in msgs_ds.iter_batches(**iter_batches_kwargs): diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index 9c59cc023aaf..1f65255e65d3 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -219,7 +219,7 @@ def _commit_pending(self) -> None: self._table.new_stream_write_builder().new_commit() ) except Exception as error: - # No commit started; all messages remain safe to abort. + # Safe to abort before commit starts. self._deferred_error = error return diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py b/paimon-python/pypaimon/write/commit/conflict_detection.py index 0c6bca856cb7..51b14a93bca5 100644 --- a/paimon-python/pypaimon/write/commit/conflict_detection.py +++ b/paimon-python/pypaimon/write/commit/conflict_detection.py @@ -185,18 +185,18 @@ def set_row_id_check_from_snapshot(self, snapshot_id): self.reset_row_id_history() def reset_row_id_history(self): - """Discard cached row-id conflict history.""" + """Clear cached row-id history.""" self._row_id_history_base_snapshot = None self._row_id_history_cursor = None self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] def ignore_row_id_commit(self, commit_user, commit_identifier): - """Exclude one known-disjoint commit from later checks.""" + """Skip a disjoint commit in later checks.""" self._row_id_ignored_commits.add((commit_user, commit_identifier)) def _row_id_history_snapshots(self, latest_snapshot): - """Cache external history while skipping registered disjoint windows.""" + """Cache history except disjoint commits.""" base_snapshot = self._row_id_check_from_snapshot reset_history = ( self._row_id_history_base_snapshot != base_snapshot diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index fea0ca8ec3a0..7956165db610 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -56,7 +56,7 @@ def is_success(self) -> bool: class CommitOutcomeUnknownError(RuntimeError): - """Raised when commit messages may already be in a snapshot.""" + """The commit may already be visible.""" class SuccessResult(CommitResult): @@ -309,7 +309,6 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, latest_snapshot = self.snapshot_manager.get_latest_snapshot() commit_entries = commit_entries_plan(latest_snapshot) - # Skip commits with no changes. if not commit_entries and not index_deletes and not index_adds: if outcome_unknown: if self._is_commit_persisted( diff --git a/paimon-python/pypaimon/write/table_commit.py b/paimon-python/pypaimon/write/table_commit.py index e1a77421c667..0a0f62492266 100644 --- a/paimon-python/pypaimon/write/table_commit.py +++ b/paimon-python/pypaimon/write/table_commit.py @@ -92,7 +92,7 @@ def abort(self, commit_messages: List[CommitMessage]): self.file_store_commit.abort(commit_messages) def ignore_row_id_conflict_for_commit(self, commit_identifier: int) -> None: - """Exclude a known-disjoint commit from later row-id checks.""" + """Skip a disjoint commit in later row-id checks.""" self.file_store_commit.conflict_detection.ignore_row_id_commit( self.commit_user, commit_identifier ) From 2da62b93a28d329d870f415ceeab6466cf5e3967 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 23 Jul 2026 23:58:32 +0800 Subject: [PATCH 6/8] [python][ray] Fix incremental commit lineage and retries --- .../pypaimon/ray/data_evolution_merge_into.py | 21 ++- .../pypaimon/ray/update_by_row_id.py | 62 ++++++- .../pypaimon/tests/file_store_commit_test.py | 35 +++- .../tests/ray_update_by_row_id_test.py | 159 +++++++++++++++--- .../rest/rest_catalog_commit_snapshot_test.py | 3 +- .../tests/write/conflict_detection_test.py | 81 +++++++++ .../pypaimon/write/commit/commit_scanner.py | 21 ++- .../write/commit/conflict_detection.py | 41 +++++ .../pypaimon/write/file_store_commit.py | 68 +++----- 9 files changed, 404 insertions(+), 87 deletions(-) diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py b/paimon-python/pypaimon/ray/data_evolution_merge_into.py index d43ac21f36bc..13fa25755f13 100644 --- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py +++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py @@ -522,14 +522,27 @@ def _require_ray_join() -> None: def _reraise_inner(err: BaseException) -> None: """Unwrap Ray's RayTaskError so callers see the worker-side exception.""" + try: + from ray.exceptions import RayTaskError + except ImportError: + raise err + + if not isinstance(err, RayTaskError): + raise err + inner = err - cause = getattr(err, "cause", None) or getattr(err, "__cause__", None) - while cause is not None: + seen = {id(err)} + cause = getattr(err, "cause", None) + while cause is not None and id(cause) not in seen: + seen.add(id(cause)) inner = cause - cause = getattr(inner, "cause", None) or getattr(inner, "__cause__", None) + cause = ( + getattr(inner, "cause", None) + if isinstance(inner, RayTaskError) else None + ) if inner is err: raise err - raise inner from err + raise inner from None def _validate_disjoint_action_row_ids(update_row_ids, delete_row_ids) -> None: diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index 1f65255e65d3..3c7de2e4f0a7 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -152,7 +152,9 @@ def _project_cast(batch: pa.Table) -> pa.Table: f"target '{target}' has no rows; every _ROW_ID in the source is foreign.") return {"num_updated": 0} incremental_committer = ( - _IncrementalUpdateCommitter(table, max_groups_per_commit) + _IncrementalUpdateCommitter( + table, max_groups_per_commit, base.id + ) if max_groups_per_commit is not None else None ) try: @@ -183,15 +185,18 @@ def _project_cast(batch: pa.Table) -> pa.Table: class _IncrementalUpdateCommitter: - def __init__(self, table, max_groups_per_commit: int): + def __init__(self, table, max_groups_per_commit: int, + base_snapshot_id: Optional[int] = None): self._table = table self._max_groups_per_commit = max_groups_per_commit + self._base_snapshot_id = base_snapshot_id self._pending_messages: list = [] self._pending_groups = 0 self._table_commit = None self._next_commit_identifier = 1 self._deferred_error = None self._uncertain_messages: list = [] + self._last_committed_snapshot = None def add_group(self, commit_messages, _num_updated, _row_ids) -> None: self._pending_messages.extend(commit_messages) @@ -203,6 +208,11 @@ def add_group(self, commit_messages, _num_updated, _row_ids) -> None: def finish(self) -> None: if self._deferred_error is None: self._commit_pending() + if self._deferred_error is None: + try: + self._validate_committed_snapshot() + except Exception as error: + self._deferred_error = error if self._deferred_error is not None: raise self._deferred_error @@ -225,6 +235,11 @@ def _commit_pending(self) -> None: group_count = self._pending_groups commit_identifier = self._next_commit_identifier + try: + self._validate_committed_snapshot() + except Exception as error: + self._deferred_error = error + return try: self._table_commit.commit( self._pending_messages, commit_identifier @@ -242,9 +257,11 @@ def _commit_pending(self) -> None: self._pending_groups = 0 self._next_commit_identifier += 1 try: + self._record_committed_snapshot(commit_identifier) # Later windows are disjoint by _FIRST_ROW_ID. self._table_commit.ignore_row_id_conflict_for_commit( commit_identifier) + self._validate_committed_snapshot() except Exception as error: self._uncertain_messages.extend(committed_messages) self._deferred_error = error @@ -254,6 +271,47 @@ def _commit_pending(self) -> None: group_count, ) + def _record_committed_snapshot(self, commit_identifier: int) -> None: + if self._base_snapshot_id is None: + return + + self._validate_committed_snapshot() + snapshot_manager = self._table.snapshot_manager() + latest = snapshot_manager.get_latest_snapshot() + start_id = ( + self._last_committed_snapshot.id + 1 + if self._last_committed_snapshot is not None + else self._base_snapshot_id + 1 + ) + if latest is not None: + for snapshot_id in range(latest.id, start_id - 1, -1): + snapshot = snapshot_manager.get_snapshot_by_id(snapshot_id) + if (snapshot is not None + and snapshot.commit_user == self._table_commit.commit_user + and snapshot.commit_identifier == commit_identifier + and snapshot.commit_kind == "APPEND"): + self._last_committed_snapshot = snapshot + return + raise RuntimeError( + "Incremental update_by_row_id snapshot is no longer in the " + "current lineage." + ) + + def _validate_committed_snapshot(self) -> None: + committed = self._last_committed_snapshot + if committed is None: + return + + snapshot_manager = self._table.snapshot_manager() + latest = snapshot_manager.get_latest_snapshot() + current = snapshot_manager.get_snapshot_by_id(committed.id) + if (latest is None or latest.id < committed.id + or current != committed): + raise RuntimeError( + "Incremental update_by_row_id snapshot is no longer in the " + "current lineage." + ) + def abort_pending(self) -> None: if not self._pending_messages: return diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index b00e66a768c7..5a2ad71c63fe 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -461,7 +461,6 @@ def test_atomic_exception_has_unknown_outcome( atomic_error = RuntimeError("atomic commit failed") file_store_commit._try_commit_once = Mock(return_value=RetryResult( None, atomic_error, outcome_unknown=True)) - file_store_commit._is_commit_persisted = Mock(return_value=False) with self.assertRaises(CommitOutcomeUnknownError) as raised: file_store_commit._try_commit( @@ -528,7 +527,6 @@ def test_unknown_outcome_survives_later_failure( RetryResult(None, atomic_error, outcome_unknown=True), RuntimeError("conflict"), ]) - file_store_commit._is_commit_persisted = Mock(return_value=False) with self.assertRaises(CommitOutcomeUnknownError) as raised: file_store_commit._try_commit( @@ -536,18 +534,41 @@ def test_unknown_outcome_survives_later_failure( self.assertIs(atomic_error, raised.exception.__cause__) - def test_persisted_identity_resolves_unknown_outcome( + def test_atomic_failure_remains_unknown( self, mock_manifest_list_manager, mock_manifest_file_manager): file_store_commit = self._create_file_store_commit() file_store_commit.commit_max_retries = 0 file_store_commit.commit_timeout = 1000 file_store_commit.snapshot_manager.get_latest_snapshot.return_value = None + atomic_error = RuntimeError("response lost") file_store_commit._try_commit_once = Mock(return_value=RetryResult( - None, RuntimeError("response lost"), outcome_unknown=True)) - file_store_commit._is_commit_persisted = Mock(return_value=True) + None, atomic_error, outcome_unknown=True)) + + with self.assertRaises(CommitOutcomeUnknownError) as raised: + file_store_commit._try_commit( + "APPEND", 1, lambda _snapshot: [Mock()]) + + self.assertIs(atomic_error, raised.exception.__cause__) - file_store_commit._try_commit( - "APPEND", 1, lambda _snapshot: [Mock()]) + def test_unknown_duplicate_remains_unknown( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + atomic_error = RuntimeError("response lost") + retry_result = RetryResult( + None, atomic_error, outcome_unknown=True) + file_store_commit._is_duplicate_commit = Mock(return_value=True) + + with self.assertRaises(CommitOutcomeUnknownError) as raised: + file_store_commit._try_commit_once( + retry_result=retry_result, + commit_kind="APPEND", + commit_entries=[Mock()], + changelog_entries=[], + commit_identifier=1, + latest_snapshot=Mock(), + ) + + self.assertIs(atomic_error, raised.exception.__cause__) def test_null_partition_value( self, mock_manifest_list_manager, mock_manifest_file_manager): diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py index b6d99becd2ba..d631271aaeb6 100644 --- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py @@ -132,6 +132,8 @@ def test_updates_correct_row_across_files(self): self.assertTrue(all(v == 0 for k, v in got.items() if k != 21)) def test_incrementally_commits_file_group_windows(self): + from pypaimon.write.commit.commit_scanner import CommitScanner + target = self._create() chunks = [ [group_start, group_start + 1] @@ -160,6 +162,19 @@ def test_incrementally_commits_file_group_windows(self): first_window_committed = threading.Event() resume_iteration = threading.Event() observed = {} + scan_counts = {"full": 0, "incremental": 0} + original_full_scan = ( + CommitScanner.read_all_entries_from_changed_partitions + ) + original_incremental_scan = CommitScanner.read_incremental_changes + + def counting_full_scan(scanner, *args, **kwargs): + scan_counts["full"] += 1 + return original_full_scan(scanner, *args, **kwargs) + + def counting_incremental_scan(scanner, *args, **kwargs): + scan_counts["incremental"] += 1 + return original_incremental_scan(scanner, *args, **kwargs) def pausing_iter_batches(dataset, *args, **kwargs): iter_batch_options.append(( @@ -189,14 +204,23 @@ def observe_first_window(): try: with mock.patch.object( ray.data.Dataset, "iter_batches", pausing_iter_batches): - stats = update_by_row_id( - target, - ray.data.from_arrow(src).repartition(8), - self.catalog_options, - update_cols=["age"], - num_partitions=4, - max_groups_per_commit=3, - ) + with mock.patch.object( + CommitScanner, + "read_all_entries_from_changed_partitions", + counting_full_scan, + ), mock.patch.object( + CommitScanner, + "read_incremental_changes", + counting_incremental_scan, + ): + stats = update_by_row_id( + target, + ray.data.from_arrow(src).repartition(8), + self.catalog_options, + update_cols=["age"], + num_partitions=4, + max_groups_per_commit=3, + ) finally: resume_iteration.set() observer.join(30) @@ -205,6 +229,7 @@ def observe_first_window(): self.assertEqual([(1, 0)], iter_batch_options) self.assertNotIn("error", observed) self.assertEqual(base_snapshot_id + 1, observed["snapshot_id"]) + self.assertEqual({"full": 1, "incremental": 2}, scan_counts) latest = table.snapshot_manager().get_latest_snapshot() self.assertEqual(base_snapshot_id + 3, latest.id) incremental_snapshots = [ @@ -416,7 +441,7 @@ def commit_then_rebuild_history( with mock.patch.object( StreamTableCommit, "commit", commit_then_rebuild_history): with self.assertRaisesRegex( - RuntimeError, "updating the same file"): + RuntimeError, "no longer in the current lineage"): update_by_row_id( target, ray.data.from_arrow(source).repartition(3), @@ -426,12 +451,87 @@ def commit_then_rebuild_history( max_groups_per_commit=1, ) - self.assertEqual([1, 2, 3], commit_calls) + self.assertEqual([1, 2], commit_calls) self.assertEqual( [999, 999, 999], self._read(target).sort_by("id")["age"].to_pylist(), ) + def test_incremental_commit_fails_after_external_rollback(self): + target = self._create() + for row_id in range(1, 3): + self._write(target, pa.Table.from_pydict( + {"id": [row_id], "name": ["a"], "age": [0]}, + schema=self.pa_schema, + )) + table = self.catalog.get_table(target) + base_snapshot_id = table.snapshot_manager().get_latest_snapshot().id + row_ids = self._rowid_by_id(target) + source = pa.table( + { + "_ROW_ID": [row_ids[1], row_ids[2]], + "age": [100, 200], + }, + schema=pa.schema([ + ("_ROW_ID", pa.int64()), + ("age", pa.int32()), + ]), + ) + + original_iter_batches = ray.data.Dataset.iter_batches + first_window_committed = threading.Event() + resume_iteration = threading.Event() + rollback_error = [] + + def pausing_iter_batches(dataset, *args, **kwargs): + for batch_number, batch in enumerate( + original_iter_batches(dataset, *args, **kwargs), 1): + yield batch + if batch_number == 1: + first_window_committed.set() + if not resume_iteration.wait(30): + raise TimeoutError("timed out waiting for rollback") + + def rollback_first_window(): + try: + if not first_window_committed.wait(30): + raise TimeoutError("first commit was not observed") + table.rollback_to(base_snapshot_id) + except Exception as error: + rollback_error.append(error) + finally: + resume_iteration.set() + + rollback_thread = threading.Thread(target=rollback_first_window) + rollback_thread.start() + try: + with mock.patch.object( + ray.data.Dataset, "iter_batches", pausing_iter_batches): + with self.assertRaisesRegex( + RuntimeError, "no longer in the current lineage"): + update_by_row_id( + target, + ray.data.from_arrow(source).repartition(2), + self.catalog_options, + update_cols=["age"], + num_partitions=2, + max_groups_per_commit=1, + ) + finally: + resume_iteration.set() + rollback_thread.join(30) + + self.assertEqual([], rollback_error) + self.assertFalse(rollback_thread.is_alive()) + self.assertEqual( + base_snapshot_id, + table.snapshot_manager().get_latest_snapshot().id, + ) + self.assertEqual( + [0, 0], + self._read(target).sort_by("id")["age"].to_pylist(), + ) + def test_incremental_committer_batches_complete_groups(self): import importlib m = importlib.import_module("pypaimon.ray.update_by_row_id") @@ -750,28 +850,35 @@ def test_commit_failure_does_not_abort_after_commit_started(self): self.assertEqual(recorder["abort_calls"], 0) self.assertEqual(recorder["close_calls"], 1) - def test_chained_commit_failure_is_unwrapped_once(self): - import importlib + def test_driver_commit_failure_is_not_unwrapped(self): + from pypaimon.write.file_store_commit import CommitOutcomeUnknownError - m = importlib.import_module("pypaimon.ray.update_by_row_id") retry_error = ValueError("retry failed") - try: - raise RuntimeError("commit failed") from retry_error - except RuntimeError as commit_error: - chained_error = commit_error + for error_type in (RuntimeError, CommitOutcomeUnknownError): + try: + raise error_type("commit failed") from retry_error + except error_type as commit_error: + chained_error = commit_error - reraise_calls = [] + with self.assertRaises(error_type) as raised: + self._run_with_fake_commit(commit_error=chained_error) - def reraise_once(error): - reraise_calls.append(error) - raise ValueError("retry failed") from None + self.assertIs(chained_error, raised.exception) + self.assertIs(retry_error, raised.exception.__cause__) - with mock.patch.object(m, "_reraise_inner", side_effect=reraise_once): - with self.assertRaisesRegex(ValueError, "retry failed"): - self._run_with_fake_commit(commit_error=chained_error) + def test_ray_task_error_is_unwrapped(self): + from ray.exceptions import RayTaskError + + from pypaimon.ray.data_evolution_merge_into import _reraise_inner + + worker_error = ValueError("worker failed") + ray_error = RayTaskError("worker", "trace", worker_error) + + with self.assertRaises(ValueError) as raised: + _reraise_inner(ray_error) - self.assertEqual(1, len(reraise_calls)) - self.assertIs(chained_error, reraise_calls[0]) + self.assertIs(worker_error, raised.exception) + self.assertIsNone(raised.exception.__cause__) def test_close_failure_after_success_warns_and_returns_stats(self): close_error = RuntimeError("close failed") diff --git a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py index 975c6f643bc2..bc28f60c9798 100644 --- a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py @@ -33,6 +33,7 @@ from pypaimon.snapshot.snapshot import Snapshot from pypaimon.snapshot.snapshot_commit import PartitionStatistics from pypaimon.tests.rest.rest_base_test import RESTBaseTest +from pypaimon.write.file_store_commit import CommitOutcomeUnknownError class TestRESTCatalogCommitSnapshot(unittest.TestCase): @@ -352,7 +353,7 @@ def commit_then_raise(sn, st): raise RuntimeError("simulated") with patch.object(tc.file_store_commit.snapshot_commit, 'commit', side_effect=commit_then_raise): - with self.assertRaises(RuntimeError): + with self.assertRaises(CommitOutcomeUnknownError): tc.commit(cm) tw.close() tc.close() diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py b/paimon-python/pypaimon/tests/write/conflict_detection_test.py index e99220ec12f6..682c9c6f30f7 100644 --- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py +++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py @@ -354,12 +354,93 @@ def read_incremental_raw_entries_from_changed_partitions(self, snapshot, _): return self._raw_by_id.get(snapshot.id, self._by_id.get(snapshot.id, [])) +class _FakeBaseEntryScanner: + + def __init__(self, full_entries, incremental_entries, signature=None): + self._full_entries = full_entries + self._incremental_entries = incremental_entries + self._signature = signature + self.full_calls = [] + self.incremental_calls = [] + + def changed_partition_signature(self, _entries, _index_entries=None): + return self._signature + + def read_all_entries_from_changed_partitions( + self, snapshot, _entries, _index_entries=None): + self.full_calls.append(snapshot.delta_manifest_list) + return list(self._full_entries[snapshot.delta_manifest_list]) + + def read_incremental_changes( + self, from_snapshot, to_snapshot, _entries, _index_entries=None): + self.incremental_calls.append((from_snapshot.id, to_snapshot.id)) + result = [] + for snapshot_id in range(from_snapshot.id + 1, to_snapshot.id + 1): + result.extend(self._incremental_entries.get(snapshot_id, [])) + return result + + class _FakeTable: def __init__(self, schema_manager): self.schema_manager = schema_manager +class TestRowIdBaseEntriesCache(unittest.TestCase): + + def _make_detection(self, snapshots, scanner): + detection = ConflictDetection( + data_evolution_enabled=True, + snapshot_manager=_FakeSnapshotManager(snapshots), + manifest_list_manager=None, + table=_FakeTable(_FakeSchemaManager([_DEFAULT_SCHEMA])), + commit_scanner=scanner, + ) + detection.set_row_id_check_from_snapshot(1) + return detection + + def test_advances_base_entries_with_snapshot_deltas(self): + base = _FakeSnapshot(1, "APPEND", delta_manifest_list="base") + latest = _FakeSnapshot(2, "APPEND", delta_manifest_list="latest") + base_entry = _make_entry("base.parquet") + added_entry = _make_entry("added.parquet") + scanner = _FakeBaseEntryScanner( + {"base": [base_entry]}, {2: [added_entry]}) + detection = self._make_detection([base, latest], scanner) + + self.assertEqual( + [base_entry], detection.read_row_id_base_entries(base, [])) + self.assertEqual( + [base_entry, added_entry], + detection.read_row_id_base_entries(latest, []), + ) + + self.assertEqual(["base"], scanner.full_calls) + self.assertEqual([(1, 2)], scanner.incremental_calls) + + def test_reused_cursor_id_rebuilds_base_entries(self): + original = _FakeSnapshot(2, "APPEND", delta_manifest_list="original") + replacement = _FakeSnapshot( + 2, "APPEND", delta_manifest_list="replacement") + original_entry = _make_entry("original.parquet") + replacement_entry = _make_entry("replacement.parquet") + scanner = _FakeBaseEntryScanner({ + "original": [original_entry], + "replacement": [replacement_entry], + }, {}) + detection = self._make_detection([original], scanner) + + self.assertEqual( + [original_entry], detection.read_row_id_base_entries(original, [])) + self.assertEqual( + [replacement_entry], + detection.read_row_id_base_entries(replacement, []), + ) + + self.assertEqual(["original", "replacement"], scanner.full_calls) + self.assertEqual([], scanner.incremental_calls) + + class TestCheckRowIdFromSnapshot(unittest.TestCase): def _make_detection(self, snapshots, raw_entries_by_snapshot_id): diff --git a/paimon-python/pypaimon/write/commit/commit_scanner.py b/paimon-python/pypaimon/write/commit/commit_scanner.py index 4612054e6c5c..fee7d2a58d26 100644 --- a/paimon-python/pypaimon/write/commit/commit_scanner.py +++ b/paimon-python/pypaimon/write/commit/commit_scanner.py @@ -150,6 +150,19 @@ def read_incremental_changes(self, snapshot, commit_entries, partition_filter)) return entries + def changed_partition_signature(self, entries, index_entries=None): + """Return the changed partitions used by conflict scans.""" + if not self.table.partition_keys: + return None + + changed_partitions = set() + for entry in entries or []: + changed_partitions.add(tuple(entry.partition.values)) + for entry in index_entries or []: + if self._index_entry_changes_partition(entry): + changed_partitions.add(tuple(entry.partition.values)) + return frozenset(changed_partitions) or None + def _build_partition_filter_from_entries(self, entries: List[ManifestEntry]): return self._build_partition_filter_from_changes(entries) @@ -169,12 +182,8 @@ def _build_partition_filter_from_changes(self, entries, index_entries=None): if not partition_keys: return None - changed_partitions = set() - for entry in entries or []: - changed_partitions.add(tuple(entry.partition.values)) - for entry in index_entries or []: - if self._index_entry_changes_partition(entry): - changed_partitions.add(tuple(entry.partition.values)) + changed_partitions = self.changed_partition_signature( + entries, index_entries) if not changed_partitions: return None diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py b/paimon-python/pypaimon/write/commit/conflict_detection.py index 51b14a93bca5..4341d48fd5f7 100644 --- a/paimon-python/pypaimon/write/commit/conflict_detection.py +++ b/paimon-python/pypaimon/write/commit/conflict_detection.py @@ -164,6 +164,7 @@ def __init__(self, data_evolution_enabled, snapshot_manager, self._row_id_history_cursor = None self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] + self._row_id_base_entries_cache = {} self.commit_scanner = commit_scanner def should_be_overwrite_commit(self, append_file_entries=None, append_index_files=None): @@ -190,11 +191,51 @@ def reset_row_id_history(self): self._row_id_history_cursor = None self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] + self._row_id_base_entries_cache = {} def ignore_row_id_commit(self, commit_user, commit_identifier): """Skip a disjoint commit in later checks.""" self._row_id_ignored_commits.add((commit_user, commit_identifier)) + def read_row_id_base_entries(self, latest_snapshot, commit_entries, + index_entries=None): + """Read or advance cached base entries for row-id checks.""" + signature = self.commit_scanner.changed_partition_signature( + commit_entries, index_entries) + key = (self._row_id_check_from_snapshot, signature) + cached = self._row_id_base_entries_cache.get(key) + incremental = None + + if cached is not None: + cursor, cursor_identity, base_entries = cached + if latest_snapshot.id >= cursor.id: + current_cursor = ( + latest_snapshot + if latest_snapshot.id == cursor.id + else self.snapshot_manager.get_snapshot_by_id(cursor.id) + ) + if self._snapshot_identity(current_cursor) == cursor_identity: + incremental = self.commit_scanner.read_incremental_changes( + cursor, latest_snapshot, commit_entries, index_entries) + + if incremental is None: + base_entries = ( + self.commit_scanner.read_all_entries_from_changed_partitions( + latest_snapshot, commit_entries, index_entries) + ) + else: + base_entries = list(base_entries) + if incremental: + base_entries.extend(incremental) + base_entries = FileEntry.merge_entries(base_entries) + + self._row_id_base_entries_cache[key] = ( + latest_snapshot, + self._snapshot_identity(latest_snapshot), + list(base_entries), + ) + return base_entries + def _row_id_history_snapshots(self, latest_snapshot): """Cache history except disjoint commits.""" base_snapshot = self._row_id_check_from_snapshot diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 7956165db610..f0b824446ccb 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -311,9 +311,6 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, if not commit_entries and not index_deletes and not index_adds: if outcome_unknown: - if self._is_commit_persisted( - commit_identifier, commit_kind): - return raise CommitOutcomeUnknownError( "Atomic commit outcome is unknown." ) from outcome_unknown_cause @@ -376,9 +373,6 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, f"after {elapsed_ms} millis with {retry_count} retries, " "there maybe exist commit conflicts between multiple jobs." ) - if (outcome_unknown and self._is_commit_persisted( - commit_identifier, commit_kind)): - return error_type = ( CommitOutcomeUnknownError if outcome_unknown else RuntimeError @@ -397,9 +391,6 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, raise except Exception as error: if outcome_unknown: - if self._is_commit_persisted( - commit_identifier, commit_kind): - return raise CommitOutcomeUnknownError(str(error)) from ( outcome_unknown_cause or error ) @@ -416,6 +407,10 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str index_adds=None) -> CommitResult: start_millis = int(time.time() * 1000) if self._is_duplicate_commit(retry_result, latest_snapshot, commit_identifier, commit_kind): + if retry_result is not None and retry_result.outcome_unknown: + raise CommitOutcomeUnknownError( + "Atomic commit outcome is unknown." + ) from retry_result.exception return SuccessResult() unique_id = uuid.uuid4() @@ -429,29 +424,32 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str new_snapshot_id = latest_snapshot.id + 1 if latest_snapshot else 1 index_entries = (index_deletes or []) + (index_adds or []) - # Base entries for conflict detection. On retry, reuse the previous - # attempt's base + read only the incremental changes (mirrors Java). + # Reuse base entries and apply only new snapshot deltas. base_data_files = None if detect_conflicts and latest_snapshot is not None: - 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, - index_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( + if self.conflict_detection.has_row_id_check_from_snapshot(): + base_data_files = self.conflict_detection.read_row_id_base_entries( latest_snapshot, commit_entries, index_entries) + else: + 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, + index_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: + base_data_files = ( + self.commit_scanner.read_all_entries_from_changed_partitions( + latest_snapshot, commit_entries, index_entries) + ) conflict_exception = self.conflict_detection.check_conflicts( latest_snapshot, @@ -637,18 +635,6 @@ def _is_duplicate_commit(self, retry_result, latest_snapshot, commit_identifier, return True return False - def _is_commit_persisted(self, commit_identifier, commit_kind) -> bool: - try: - snapshots = self.snapshot_manager.list_snapshots() - return any( - snapshot.commit_user == self.commit_user - and snapshot.commit_identifier == commit_identifier - and snapshot.commit_kind == commit_kind - for snapshot in snapshots - ) - except Exception: - return False - def _create_dynamic_partition_filter(self, commit_messages: List[CommitMessage]): """Build a partition filter from the unique partitions present in commit_messages.""" predicate_builder = PredicateBuilder(self.table.partition_keys_fields) From b5321e8152b773f2c16da79d469782a6cfbf4e1c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 14:45:52 +0800 Subject: [PATCH 7/8] [python] Resolve duplicate commits after unknown outcome --- .../pypaimon/tests/file_store_commit_test.py | 53 ++++++++++++++----- .../rest/rest_catalog_commit_snapshot_test.py | 49 +++++++++++++++++ .../pypaimon/write/file_store_commit.py | 4 -- 3 files changed, 90 insertions(+), 16 deletions(-) diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index 5a2ad71c63fe..84eb35087ced 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -28,6 +28,7 @@ CommitOutcomeUnknownError, FileStoreCommit, RetryResult, + SuccessResult, ) @@ -550,25 +551,53 @@ def test_atomic_failure_remains_unknown( self.assertIs(atomic_error, raised.exception.__cause__) - def test_unknown_duplicate_remains_unknown( + def test_unknown_duplicate_resolves_to_success( self, mock_manifest_list_manager, mock_manifest_file_manager): file_store_commit = self._create_file_store_commit() atomic_error = RuntimeError("response lost") retry_result = RetryResult( None, atomic_error, outcome_unknown=True) - file_store_commit._is_duplicate_commit = Mock(return_value=True) + latest_snapshot = Mock( + id=1, + commit_user="test_user", + commit_identifier=1, + commit_kind="APPEND", + ) + file_store_commit.snapshot_manager.get_snapshot_by_id.return_value = ( + latest_snapshot + ) - with self.assertRaises(CommitOutcomeUnknownError) as raised: - file_store_commit._try_commit_once( - retry_result=retry_result, - commit_kind="APPEND", - commit_entries=[Mock()], - changelog_entries=[], - commit_identifier=1, - latest_snapshot=Mock(), - ) + result = file_store_commit._try_commit_once( + retry_result=retry_result, + commit_kind="APPEND", + commit_entries=[Mock()], + changelog_entries=[], + commit_identifier=1, + latest_snapshot=latest_snapshot, + ) - self.assertIs(atomic_error, raised.exception.__cause__) + self.assertTrue(result.is_success()) + + def test_unknown_retry_resolves_when_duplicate_is_found( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 1 + file_store_commit.commit_timeout = 1000 + file_store_commit._commit_retry_wait = Mock() + file_store_commit.snapshot_manager.get_latest_snapshot.side_effect = [ + None, + Mock(id=1), + ] + atomic_error = RuntimeError("response lost") + file_store_commit._try_commit_once = Mock(side_effect=[ + RetryResult(None, atomic_error, outcome_unknown=True), + SuccessResult(), + ]) + + file_store_commit._try_commit( + "APPEND", 1, lambda _snapshot: [Mock()]) + + self.assertEqual(2, file_store_commit._try_commit_once.call_count) def test_null_partition_value( self, mock_manifest_list_manager, mock_manifest_file_manager): diff --git a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py index bc28f60c9798..2e14001a68f9 100644 --- a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py @@ -365,6 +365,55 @@ def commit_then_raise(sn, st): self.assertEqual(actual.column('id').to_pylist(), [1, 2, 3]) self.assertEqual(actual.column('name').to_pylist(), ['a', 'b', 'c']) + def test_lost_commit_response_resolves_duplicate_as_success(self): + pa_schema = pa.schema([('id', pa.int32()), ('name', pa.string())]) + opts = { + 'bucket': '1', + 'file.format': 'parquet', + 'commit.max-retries': '1', + 'commit.min-retry-wait': '1ms', + 'commit.max-retry-wait': '1ms', + 'commit.timeout': '10s', + } + schema = Schema.from_pyarrow_schema(pa_schema, options=opts) + table_name = 'default.test_duplicate_commit_success' + self.rest_catalog.create_table(table_name, schema, False) + table = self.rest_catalog.get_table(table_name) + + write_builder = table.new_batch_write_builder() + table_write = write_builder.new_write() + table_commit = write_builder.new_commit() + data = pa.Table.from_pydict( + {'id': [1, 2, 3], 'name': ['a', 'b', 'c']}, schema=pa_schema) + table_write.write_arrow(data) + commit_messages = table_write.prepare_commit() + + real_commit = table_commit.file_store_commit.snapshot_commit.commit + commit_calls = [] + + def commit_then_lose_response(snapshot, statistics): + commit_calls.append(snapshot.id) + real_commit(snapshot, statistics) + raise RuntimeError("simulated lost response") + + with patch.object( + table_commit.file_store_commit.snapshot_commit, + 'commit', + side_effect=commit_then_lose_response): + table_commit.commit(commit_messages) + + table_write.close() + table_commit.close() + + self.assertEqual([1], commit_calls) + self.assertEqual(1, table.snapshot_manager().get_latest_snapshot().id) + read_builder = table.new_read_builder() + actual = read_builder.new_read().to_arrow( + read_builder.new_scan().plan().splits()) + self.assertEqual(actual.num_rows, 3) + self.assertEqual(actual.column('id').to_pylist(), [1, 2, 3]) + self.assertEqual(actual.column('name').to_pylist(), ['a', 'b', 'c']) + if __name__ == '__main__': unittest.main() diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index f0b824446ccb..834e9b84b61f 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -407,10 +407,6 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str index_adds=None) -> CommitResult: start_millis = int(time.time() * 1000) if self._is_duplicate_commit(retry_result, latest_snapshot, commit_identifier, commit_kind): - if retry_result is not None and retry_result.outcome_unknown: - raise CommitOutcomeUnknownError( - "Atomic commit outcome is unknown." - ) from retry_result.exception return SuccessResult() unique_id = uuid.uuid4() From cfbea7ba0fd68af8e6a90d9257d7a52ce7e040da Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 16:03:35 +0800 Subject: [PATCH 8/8] [python][ray] Fix incremental commit edge cases --- docs/docs/pypaimon/ray-data.md | 3 +- .../pypaimon/ray/update_by_row_id.py | 37 ++++- .../pypaimon/tests/file_store_commit_test.py | 78 +++++++++++ .../tests/overwrite_commit_conflict_test.py | 6 +- .../tests/partition_predicate_test.py | 23 +++ .../tests/ray_update_by_row_id_test.py | 132 +++++++++++++++++- .../rest/rest_catalog_commit_snapshot_test.py | 110 +++++++++++++++ .../tests/write/conflict_detection_test.py | 25 ++++ .../pypaimon/write/commit/commit_scanner.py | 7 +- .../write/commit/conflict_detection.py | 2 + .../pypaimon/write/file_store_commit.py | 10 ++ 11 files changed, 421 insertions(+), 12 deletions(-) diff --git a/docs/docs/pypaimon/ray-data.md b/docs/docs/pypaimon/ray-data.md index 2bbf2bb9e712..bd51197b1a61 100644 --- a/docs/docs/pypaimon/ray-data.md +++ b/docs/docs/pypaimon/ray-data.md @@ -554,7 +554,8 @@ print(metrics) # {"num_updated": 50} in its data file, so it can't be told apart from a live row without reading the target. - Incremental commit mode overlaps later worker writes with earlier commits, but is not atomic across the whole operation. If a later group fails, earlier committed windows - remain visible. + remain visible and the table is partially updated. Inspect or reconcile that state, + then rerun the intended update as appropriate. ## Read By Row Id diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py index 3c7de2e4f0a7..7fdc330cd889 100644 --- a/paimon-python/pypaimon/ray/update_by_row_id.py +++ b/paimon-python/pypaimon/ray/update_by_row_id.py @@ -70,7 +70,10 @@ def update_by_row_id( By default all file groups are committed atomically after every worker finishes. Set ``max_groups_per_commit`` to incrementally commit each completed window of target file groups. Incremental mode can leave earlier windows - committed if a later window fails. + committed if a later window fails. The table may therefore be partially + updated; callers must reconcile the result or rerun the intended update. + Concurrent snapshot expiration can also invalidate a committed-window + lineage check; the operation then stops with earlier windows still visible. Returns ``{"num_updated": }``. """ @@ -173,8 +176,34 @@ def _project_cast(batch: pa.Table) -> pa.Table: except Exception as e: if incremental_committer is not None: incremental_committer.abort_pending() - _reraise_inner(e) - raise # _reraise_inner always raises; keeps num_updated defined for linters + try: + _reraise_inner(e) + except Exception as worker_error: + deferred_error = ( + incremental_committer._deferred_error + if incremental_committer is not None else None + ) + # A commit error can be deferred while the remaining Ray groups drain. + # Preserve that earlier failure if a later worker fails, but do not + # chain an error to itself when finish() raised the deferred error. + if (deferred_error is not None + and deferred_error is not worker_error): + if deferred_error.__cause__ is not None: + # Python has only one explicit cause. Keep the original commit + # cause and report the later update failure separately. + logger.warning( + "A later update error occurred after an earlier incremental " + "commit error; preserving the original commit error: %s", + worker_error, + exc_info=( + type(worker_error), + worker_error, + worker_error.__traceback__, + ), + ) + raise deferred_error + raise deferred_error from worker_error + raise finally: if incremental_committer is not None: incremental_committer.close() @@ -195,6 +224,8 @@ def __init__(self, table, max_groups_per_commit: int, self._table_commit = None self._next_commit_identifier = 1 self._deferred_error = None + # Never abort messages whose commit outcome is unknown: their files may + # already be referenced by a snapshot. If not, orphan cleanup reclaims them. self._uncertain_messages: list = [] self._last_committed_snapshot = None diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index 84eb35087ced..2b7cbb73c6fe 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -599,6 +599,84 @@ def test_unknown_retry_resolves_when_duplicate_is_found( self.assertEqual(2, file_store_commit._try_commit_once.call_count) + def test_unknown_retry_resolves_duplicate_before_replanning( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 1 + file_store_commit.commit_timeout = 1000 + file_store_commit._commit_retry_wait = Mock() + committed_snapshot = Mock( + id=1, + commit_user="test_user", + commit_identifier=1, + commit_kind="OVERWRITE", + ) + file_store_commit.snapshot_manager.get_latest_snapshot.side_effect = [ + None, + committed_snapshot, + ] + file_store_commit.snapshot_manager.get_snapshot_by_id.return_value = ( + committed_snapshot + ) + atomic_error = RuntimeError("response lost") + file_store_commit._try_commit_once = Mock(return_value=RetryResult( + None, atomic_error, outcome_unknown=True)) + commit_entries_plan = Mock(side_effect=[[Mock()], []]) + + file_store_commit._try_commit( + "OVERWRITE", 1, commit_entries_plan) + + self.assertEqual(1, file_store_commit._try_commit_once.call_count) + self.assertEqual(1, commit_entries_plan.call_count) + get_snapshot_by_id = ( + file_store_commit.snapshot_manager.get_snapshot_by_id + ) + get_snapshot_by_id.assert_called_once_with(1) + + def test_unknown_retry_with_empty_plan_and_no_duplicate_remains_unknown( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit.commit_max_retries = 1 + file_store_commit.commit_timeout = 1000 + file_store_commit._commit_retry_wait = Mock() + other_snapshot = Mock( + id=1, + commit_user="other_user", + commit_identifier=1, + commit_kind="OVERWRITE", + ) + file_store_commit.snapshot_manager.get_latest_snapshot.side_effect = [ + None, + other_snapshot, + ] + file_store_commit.snapshot_manager.get_snapshot_by_id.return_value = ( + other_snapshot + ) + atomic_error = RuntimeError("response lost") + file_store_commit._try_commit_once = Mock(return_value=RetryResult( + None, atomic_error, outcome_unknown=True)) + commit_entries_plan = Mock(side_effect=[[Mock()], []]) + + with self.assertRaises(CommitOutcomeUnknownError) as raised: + file_store_commit._try_commit( + "OVERWRITE", 1, commit_entries_plan) + + self.assertIs(atomic_error, raised.exception.__cause__) + self.assertEqual(1, file_store_commit._try_commit_once.call_count) + self.assertEqual(2, commit_entries_plan.call_count) + + def test_empty_plan_without_retry_remains_no_op( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit = self._create_file_store_commit() + file_store_commit._try_commit_once = Mock() + file_store_commit._is_duplicate_commit = Mock() + + file_store_commit._try_commit( + "OVERWRITE", 1, lambda _snapshot: []) + + file_store_commit._try_commit_once.assert_not_called() + file_store_commit._is_duplicate_commit.assert_not_called() + def test_null_partition_value( self, mock_manifest_list_manager, mock_manifest_file_manager): from pypaimon.data.timestamp import Timestamp diff --git a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py index 738d455b8513..bb12169f4915 100644 --- a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py +++ b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py @@ -271,15 +271,15 @@ def patched_cas(snapshot, statistics): 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): + def test_retry_base_matches_full_scan_across_overwrite_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. + # A non-APPEND snapshot lands between retries. OVERWRITE falls back to + # a full scan; COMPACT can still merge deltas. Either base must be fresh. K = 2 wb = self.table.new_batch_write_builder().overwrite({'f0': 1}) diff --git a/paimon-python/pypaimon/tests/partition_predicate_test.py b/paimon-python/pypaimon/tests/partition_predicate_test.py index c362def4aa55..bba2878dab23 100644 --- a/paimon-python/pypaimon/tests/partition_predicate_test.py +++ b/paimon-python/pypaimon/tests/partition_predicate_test.py @@ -474,3 +474,26 @@ def test_raw_entries_filter_unmatched_partition(self, mock_mfm_cls): self.assertEqual(len(result), 1) self.assertEqual(tuple(result[0].partition.values), ('p1', 'us')) + + def test_incremental_changes_falls_back_on_overwrite(self): + append = Mock(id=2, commit_kind='APPEND') + overwrite = Mock(id=3, commit_kind='OVERWRITE') + after_overwrite = Mock(id=4, commit_kind='APPEND') + snapshots = {2: append, 3: overwrite, 4: after_overwrite} + + table = _mock_table() + snapshot_manager = table.snapshot_manager.return_value + snapshot_manager.get_snapshot_by_id.side_effect = snapshots.get + scanner = CommitScanner(table, Mock()) + scanner.read_incremental_raw_entries_from_changed_partitions = Mock( + return_value=[]) + + result = scanner.read_incremental_changes( + Mock(id=1), after_overwrite, []) + + self.assertIsNone(result) + self.assertEqual( + [call.args[0] for call in snapshot_manager.get_snapshot_by_id.call_args_list], + [2, 3], + ) + scanner.read_incremental_raw_entries_from_changed_partitions.assert_called_once() diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py index d631271aaeb6..aa5aad201a8b 100644 --- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py @@ -278,6 +278,8 @@ def test_incremental_commit_failure_drains_later_groups(self): seen_groups = [] aborted = [] commit_calls = [] + retry_error = ValueError("commit retry failed") + commit_error = CommitOutcomeUnknownError("commit failed") original_add_group = m._IncrementalUpdateCommitter.add_group original_abort = StreamTableCommit.abort @@ -288,7 +290,7 @@ def record_group(committer, messages, num_updated, row_ids): def fail_commit(_commit, _messages, commit_identifier): commit_calls.append(commit_identifier) - raise CommitOutcomeUnknownError("commit failed") + raise commit_error from retry_error def record_abort(commit, messages): aborted.append(list(messages)) @@ -298,7 +300,7 @@ def record_abort(commit, messages): m._IncrementalUpdateCommitter, "add_group", record_group), \ mock.patch.object(StreamTableCommit, "commit", fail_commit), \ mock.patch.object(StreamTableCommit, "abort", record_abort): - with self.assertRaisesRegex(RuntimeError, "commit failed"): + with self.assertRaises(CommitOutcomeUnknownError) as raised: update_by_row_id( target, ray.data.from_arrow(src).repartition(4), @@ -308,6 +310,8 @@ def record_abort(commit, messages): max_groups_per_commit=2, ) + self.assertIs(commit_error, raised.exception) + self.assertIs(retry_error, raised.exception.__cause__) self.assertEqual(4, len(seen_groups)) self.assertEqual([1], commit_calls) self.assertEqual( @@ -315,6 +319,130 @@ def record_abort(commit, messages): aborted, ) + def test_incremental_commit_error_precedes_later_worker_failure(self): + import importlib + + from ray.exceptions import RayTaskError + + from pypaimon.write.file_store_commit import CommitOutcomeUnknownError + from pypaimon.write.table_commit import StreamTableCommit + + m = importlib.import_module("pypaimon.ray.update_by_row_id") + target = self._create() + self._write(target, pa.Table.from_pydict( + {"id": [1], "name": ["x"], "age": [0]}, + schema=self.pa_schema, + )) + src = pa.table( + {"_ROW_ID": [0], "age": [1]}, + schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]), + ) + commit_error = CommitOutcomeUnknownError("commit outcome is uncertain") + worker_error = ValueError("worker failed") + aborted = [] + close_calls = [] + + def fail_commit(_commit, _messages, _commit_identifier): + raise commit_error + + def record_abort(_commit, messages): + aborted.append(list(messages)) + + def record_close(_commit): + close_calls.append(True) + + def fail_after_commit(_update_ds, _table, _update_cols, **kwargs): + on_group_result = kwargs["on_group_result"] + on_group_result(["uncertain-1"], 1, []) + on_group_result(["uncertain-2"], 1, []) + on_group_result(["pending"], 1, []) + raise RayTaskError("worker", "trace", worker_error) + + with mock.patch.object( + m, "distributed_update_apply", fail_after_commit), \ + mock.patch.object(StreamTableCommit, "commit", fail_commit), \ + mock.patch.object(StreamTableCommit, "abort", record_abort), \ + mock.patch.object(StreamTableCommit, "close", record_close): + with self.assertRaises(CommitOutcomeUnknownError) as raised: + update_by_row_id( + target, + src, + self.catalog_options, + update_cols=["age"], + max_groups_per_commit=2, + ) + + self.assertIs(commit_error, raised.exception) + self.assertIs(worker_error, raised.exception.__cause__) + self.assertEqual([["pending"]], aborted) + self.assertEqual([True], close_calls) + + def test_incremental_commit_preserves_cause_before_worker_failure(self): + import importlib + + from ray.exceptions import RayTaskError + + from pypaimon.write.file_store_commit import CommitOutcomeUnknownError + from pypaimon.write.table_commit import StreamTableCommit + + m = importlib.import_module("pypaimon.ray.update_by_row_id") + target = self._create() + self._write(target, pa.Table.from_pydict( + {"id": [1], "name": ["x"], "age": [0]}, + schema=self.pa_schema, + )) + src = pa.table( + {"_ROW_ID": [0], "age": [1]}, + schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]), + ) + commit_cause = ValueError("commit retry failed") + commit_error = CommitOutcomeUnknownError("commit outcome is uncertain") + worker_error = ValueError("worker failed") + aborted = [] + close_calls = [] + + def fail_commit(_commit, _messages, _commit_identifier): + raise commit_error from commit_cause + + def record_abort(_commit, messages): + aborted.append(list(messages)) + + def record_close(_commit): + close_calls.append(True) + + def fail_after_commit(_update_ds, _table, _update_cols, **kwargs): + on_group_result = kwargs["on_group_result"] + on_group_result(["uncertain-1"], 1, []) + on_group_result(["uncertain-2"], 1, []) + on_group_result(["pending"], 1, []) + raise RayTaskError("worker", "trace", worker_error) + + with mock.patch.object( + m, "distributed_update_apply", fail_after_commit), \ + mock.patch.object(StreamTableCommit, "commit", fail_commit), \ + mock.patch.object(StreamTableCommit, "abort", record_abort), \ + mock.patch.object(StreamTableCommit, "close", record_close), \ + self.assertLogs( + "pypaimon.ray.update_by_row_id", level="WARNING") as logs: + with self.assertRaises(CommitOutcomeUnknownError) as raised: + update_by_row_id( + target, + src, + self.catalog_options, + update_cols=["age"], + max_groups_per_commit=2, + ) + + self.assertIs(commit_error, raised.exception) + self.assertIs(commit_cause, raised.exception.__cause__) + self.assertIs(worker_error, raised.exception.__context__) + self.assertIn( + "preserving the original commit error: worker failed", + "\n".join(logs.output), + ) + self.assertEqual([["pending"]], aborted) + self.assertEqual([True], close_calls) + def test_incremental_conflict_aborts_output_files(self): from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER from pypaimon.write.table_commit import StreamTableCommit diff --git a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py index 2e14001a68f9..5fe6da3e0bd9 100644 --- a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py @@ -414,6 +414,116 @@ def commit_then_lose_response(snapshot, statistics): self.assertEqual(actual.column('id').to_pylist(), [1, 2, 3]) self.assertEqual(actual.column('name').to_pylist(), ['a', 'b', 'c']) + def test_lost_truncate_response_resolves_duplicate_as_success(self): + pa_schema = pa.schema([('id', pa.int32()), ('name', pa.string())]) + opts = { + 'bucket': '1', + 'file.format': 'parquet', + 'commit.max-retries': '1', + 'commit.min-retry-wait': '1ms', + 'commit.max-retry-wait': '1ms', + 'commit.timeout': '10s', + } + schema = Schema.from_pyarrow_schema(pa_schema, options=opts) + table_name = 'default.test_duplicate_truncate_success' + self.rest_catalog.create_table(table_name, schema, False) + table = self.rest_catalog.get_table(table_name) + + write_builder = table.new_batch_write_builder() + table_write = write_builder.new_write() + table_commit = write_builder.new_commit() + data = pa.Table.from_pydict( + {'id': [1, 2, 3], 'name': ['a', 'b', 'c']}, schema=pa_schema) + table_write.write_arrow(data) + table_commit.commit(table_write.prepare_commit()) + table_write.close() + table_commit.close() + + truncate_commit = table.new_batch_write_builder().new_commit() + real_commit = truncate_commit.file_store_commit.snapshot_commit.commit + commit_calls = [] + + def commit_then_lose_response(snapshot, statistics): + commit_calls.append(snapshot.id) + real_commit(snapshot, statistics) + raise RuntimeError("simulated lost response") + + with patch.object( + truncate_commit.file_store_commit.snapshot_commit, + 'commit', + side_effect=commit_then_lose_response): + truncate_commit.truncate_table() + truncate_commit.close() + + self.assertEqual([2], commit_calls) + latest_snapshot = table.snapshot_manager().get_latest_snapshot() + self.assertEqual(2, latest_snapshot.id) + self.assertEqual('OVERWRITE', latest_snapshot.commit_kind) + read_builder = table.new_read_builder() + actual = read_builder.new_read().to_arrow( + read_builder.new_scan().plan().splits()) + self.assertEqual(0, actual.num_rows) + + def test_lost_drop_partitions_response_resolves_duplicate_as_success(self): + pa_schema = pa.schema([ + ('id', pa.int32()), + ('name', pa.string()), + ('dt', pa.string()), + ]) + opts = { + 'bucket': '1', + 'file.format': 'parquet', + 'commit.max-retries': '1', + 'commit.min-retry-wait': '1ms', + 'commit.max-retry-wait': '1ms', + 'commit.timeout': '10s', + } + schema = Schema.from_pyarrow_schema( + pa_schema, partition_keys=['dt'], options=opts) + table_name = 'default.test_duplicate_drop_partitions_success' + self.rest_catalog.create_table(table_name, schema, False) + table = self.rest_catalog.get_table(table_name) + + write_builder = table.new_batch_write_builder() + table_write = write_builder.new_write() + table_commit = write_builder.new_commit() + data = pa.Table.from_pydict({ + 'id': [1, 2], + 'name': ['drop', 'keep'], + 'dt': ['p1', 'p2'], + }, schema=pa_schema) + table_write.write_arrow(data) + table_commit.commit(table_write.prepare_commit()) + table_write.close() + table_commit.close() + + partition_commit = table.new_batch_write_builder().new_commit() + real_commit = partition_commit.file_store_commit.snapshot_commit.commit + commit_calls = [] + + def commit_then_lose_response(snapshot, statistics): + commit_calls.append(snapshot.id) + real_commit(snapshot, statistics) + raise RuntimeError("simulated lost response") + + with patch.object( + partition_commit.file_store_commit.snapshot_commit, + 'commit', + side_effect=commit_then_lose_response): + partition_commit.truncate_partitions([{'dt': 'p1'}]) + partition_commit.close() + + self.assertEqual([2], commit_calls) + latest_snapshot = table.snapshot_manager().get_latest_snapshot() + self.assertEqual(2, latest_snapshot.id) + self.assertEqual('OVERWRITE', latest_snapshot.commit_kind) + read_builder = table.new_read_builder() + actual = read_builder.new_read().to_arrow( + read_builder.new_scan().plan().splits()) + self.assertEqual([2], actual.column('id').to_pylist()) + self.assertEqual(['keep'], actual.column('name').to_pylist()) + self.assertEqual(['p2'], actual.column('dt').to_pylist()) + if __name__ == '__main__': unittest.main() diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py b/paimon-python/pypaimon/tests/write/conflict_detection_test.py index 682c9c6f30f7..2892fd53f30a 100644 --- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py +++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py @@ -18,6 +18,7 @@ import unittest from dataclasses import dataclass from typing import List +from unittest.mock import Mock from pypaimon.index.index_file_meta import IndexFileMeta from pypaimon.manifest.index_manifest_entry import IndexManifestEntry @@ -440,6 +441,30 @@ def test_reused_cursor_id_rebuilds_base_entries(self): self.assertEqual(["original", "replacement"], scanner.full_calls) self.assertEqual([], scanner.incremental_calls) + def test_incremental_fallback_rebuilds_overwritten_base_entries(self): + base = _FakeSnapshot(1, "APPEND", delta_manifest_list="base") + overwrite = _FakeSnapshot( + 2, "OVERWRITE", delta_manifest_list="overwrite") + old_entry = _make_entry("old.parquet") + replacement_entry = _make_entry("replacement.parquet") + scanner = _FakeBaseEntryScanner({ + "base": [old_entry], + "overwrite": [replacement_entry], + }, {}) + scanner.read_incremental_changes = Mock(return_value=None) + detection = self._make_detection([base, overwrite], scanner) + + self.assertEqual( + [old_entry], detection.read_row_id_base_entries(base, [])) + self.assertEqual( + [replacement_entry], + detection.read_row_id_base_entries(overwrite, []), + ) + + self.assertEqual(["base", "overwrite"], scanner.full_calls) + scanner.read_incremental_changes.assert_called_once_with( + base, overwrite, [], None) + class TestCheckRowIdFromSnapshot(unittest.TestCase): diff --git a/paimon-python/pypaimon/write/commit/commit_scanner.py b/paimon-python/pypaimon/write/commit/commit_scanner.py index fee7d2a58d26..4e6ab6ddd9ba 100644 --- a/paimon-python/pypaimon/write/commit/commit_scanner.py +++ b/paimon-python/pypaimon/write/commit/commit_scanner.py @@ -134,8 +134,9 @@ def read_incremental_changes(self, 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). An OVERWRITE may replace the base manifest + without fully describing the replacement in its delta manifest. """ snapshot_manager = self.table.snapshot_manager() partition_filter = self._build_partition_filter_from_changes( @@ -143,7 +144,7 @@ def read_incremental_changes(self, 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: + if snapshot is None or snapshot.commit_kind == "OVERWRITE": return None entries.extend( self.read_incremental_raw_entries_from_changed_partitions( diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py b/paimon-python/pypaimon/write/commit/conflict_detection.py index 4341d48fd5f7..863472ca898b 100644 --- a/paimon-python/pypaimon/write/commit/conflict_detection.py +++ b/paimon-python/pypaimon/write/commit/conflict_detection.py @@ -262,6 +262,8 @@ def _row_id_history_snapshots(self, latest_snapshot): self._row_id_history_cursor_identity = None self._row_id_external_snapshots = [] + # Snapshot files below the cursor are immutable. A rollback replaces the + # whole tail, including the cursor, whose identity check above resets cache. for snapshot_id in range( self._row_id_history_cursor + 1, latest_snapshot.id + 1): diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 834e9b84b61f..4b8a0916e5c6 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -307,6 +307,16 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, while True: try: latest_snapshot = self.snapshot_manager.get_latest_snapshot() + # A successful attempt whose response was lost can make an + # overwrite/truncate retry plan empty. Resolve it before replanning. + if (outcome_unknown + and self._is_duplicate_commit( + retry_result, + latest_snapshot, + commit_identifier, + commit_kind)): + break + commit_entries = commit_entries_plan(latest_snapshot) if not commit_entries and not index_deletes and not index_adds: