This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a6ffec2b1f [python][ray] Rebase self-merge updates after compaction
(#9339)
a6ffec2b1f is described below
commit a6ffec2b1f169cd0594a97a4bb80dc154731275f
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Aug 23 09:19:50 2026 +0800
[python][ray] Rebase self-merge updates after compaction (#9339)
---
.../pypaimon/ray/data_evolution_merge_into.py | 58 +-
.../pypaimon/ray/row_id_conflict_rewriter.py | 462 ++++++++++++++
.../pypaimon/tests/file_store_commit_test.py | 86 ++-
.../tests/overwrite_commit_conflict_test.py | 23 +-
.../tests/ray_data_evolution_merge_into_test.py | 707 +++++++++++++++++++++
.../tests/ray_row_id_conflict_rewriter_test.py | 289 +++++++++
.../pypaimon/tests/reader_append_only_test.py | 4 +-
.../tests/write/conflict_detection_test.py | 14 +
.../pypaimon/tests/write/table_write_test.py | 20 +-
.../pypaimon/write/commit/conflict_detection.py | 4 +-
paimon-python/pypaimon/write/file_store_commit.py | 63 +-
11 files changed, 1670 insertions(+), 60 deletions(-)
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
index 6124ebefd5..e0544f1890 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
@@ -436,6 +436,8 @@ def _execute_and_commit(
num_deleted = 0
delete_row_ids = []
num_inserted = 0
+ insert_msgs: list = []
+ self_merge_update = isinstance(update_ds, _SelfMergeUpdatePlan)
try:
if update_ds is not None:
@@ -494,20 +496,32 @@ def _execute_and_commit(
all_msgs: list = list(commit_messages)
if all_msgs:
- table_commit = None
- try:
- table_commit = table.new_batch_write_builder().new_commit()
- table_commit.commit(all_msgs)
- finally:
- if table_commit is not None:
- try:
- table_commit.close()
- except Exception as close_error:
- logger.warning(
- "Failed to close merge_into commit: %s",
- close_error,
- exc_info=close_error,
- )
+ if self_merge_update and update_msgs:
+ from pypaimon.ray.row_id_conflict_rewriter import (
+ commit_self_merge_with_compaction_retry,
+ )
+ commit_self_merge_with_compaction_retry(
+ table,
+ update_msgs,
+ delete_msgs + insert_msgs,
+ num_partitions=num_partitions,
+ ray_remote_args=ray_remote_args,
+ )
+ else:
+ table_commit = None
+ try:
+ table_commit = table.new_batch_write_builder().new_commit()
+ table_commit.commit(all_msgs)
+ finally:
+ if table_commit is not None:
+ try:
+ table_commit.close()
+ except Exception as close_error:
+ logger.warning(
+ "Failed to close merge_into commit: %s",
+ close_error,
+ exc_info=close_error,
+ )
except Exception as e:
_reraise_inner(e)
@@ -555,14 +569,22 @@ def _require_ray_join() -> None:
def _reraise_inner(err: BaseException) -> None:
- """Unwrap Ray's RayTaskError so callers see the worker-side exception."""
+ """Unwrap only RayTaskError layers and preserve ordinary exception
chains."""
+ try:
+ from ray.exceptions import RayTaskError
+ except ImportError:
+ raise err
+
inner = err
- cause = getattr(err, "cause", None) or getattr(err, "__cause__", None)
- while cause is not None:
+ while isinstance(inner, RayTaskError):
+ cause = getattr(inner, "cause", None)
+ if cause is None or cause is inner:
+ break
inner = cause
- cause = getattr(inner, "cause", None) or getattr(inner, "__cause__",
None)
if inner is err:
raise err
+ if getattr(inner, "__cause__", None) is not None:
+ raise inner
raise inner from err
diff --git a/paimon-python/pypaimon/ray/row_id_conflict_rewriter.py
b/paimon-python/pypaimon/ray/row_id_conflict_rewriter.py
new file mode 100644
index 0000000000..14cfb0ac22
--- /dev/null
+++ b/paimon-python/pypaimon/ray/row_id_conflict_rewriter.py
@@ -0,0 +1,462 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+################################################################################
+
+"""Rebase Ray self-merge updates after concurrent compaction."""
+
+import logging
+import random
+import time
+from dataclasses import dataclass, replace
+from typing import Dict, List, Optional, Sequence, Tuple
+
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.read.split import DataSplit
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.utils.range import Range
+from pypaimon.write.commit.conflict_detection import RowIdExistenceConflict
+from pypaimon.write.commit_message import CommitMessage
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class _StagedFile:
+ message_index: int
+ message: CommitMessage
+ file: DataFileMeta
+
+
+@dataclass(frozen=True)
+class _RewriteResult:
+ update_messages: List[CommitMessage]
+ rewritten_file_count: int
+
+
+def commit_self_merge_with_compaction_retry(
+ table,
+ update_messages: Sequence[CommitMessage],
+ other_messages: Sequence[CommitMessage],
+ *,
+ num_partitions: int,
+ ray_remote_args=None,
+) -> None:
+ """Commit self-merge messages, rebasing stale updates when compaction
wins."""
+ current_updates = list(update_messages)
+ other_messages = list(other_messages)
+ retry_count = 0
+ start_millis = int(time.time() * 1000)
+
+ # Ray performs the rebase below without the local driver's size limit.
+ commit_table = table.copy_without_time_travel({
+ CoreOptions.DATA_EVOLUTION_ROW_ID_CONFLICT_REWRITE_MAX_SIZE.key():
+ "0 B",
+ })
+ base_snapshot_ids = _base_snapshot_ids(current_updates)
+ latest_snapshot = table.snapshot_manager().get_latest_snapshot()
+ if (
+ len(base_snapshot_ids) == 1
+ and latest_snapshot is not None
+ and latest_snapshot.id != next(iter(base_snapshot_ids))
+ ):
+ result = _rewrite_updates(
+ table,
+ current_updates,
+ latest_snapshot,
+ num_partitions=num_partitions,
+ ray_remote_args=ray_remote_args,
+ )
+ if result is not None:
+ current_updates = result.update_messages
+ logger.info(
+ "Rewrote %d stale self-merge file(s) against snapshot %d "
+ "before committing to table %s.",
+ result.rewritten_file_count,
+ latest_snapshot.id,
+ table.identifier,
+ )
+
+ # Match Spark's PaimonSparkWriter: every outer rebase attempt creates a
+ # fresh commit from one write builder, preserving the commit user.
+ write_builder = commit_table.new_batch_write_builder()
+ while True:
+ commit = None
+ conflict = None
+ try:
+ commit = write_builder.new_commit()
+ commit.commit(current_updates + other_messages)
+ except Exception as error:
+ conflict = _find_row_id_conflict(error)
+ if conflict is None:
+ raise
+ finally:
+ if commit is not None:
+ try:
+ commit.close()
+ except Exception as close_error:
+ logger.warning(
+ "Failed to close self-merge commit: %s",
+ close_error,
+ exc_info=close_error,
+ )
+
+ if conflict is None:
+ return
+
+ elapsed = int(time.time() * 1000) - start_millis
+ if (
+ elapsed > table.options.commit_timeout()
+ or retry_count >= table.options.commit_max_retries()
+ ):
+ raise conflict
+
+ latest_snapshot = table.snapshot_manager().get_latest_snapshot()
+ if latest_snapshot is None:
+ raise conflict
+
+ try:
+ result = _rewrite_updates(
+ table,
+ current_updates,
+ latest_snapshot,
+ num_partitions=num_partitions,
+ ray_remote_args=ray_remote_args,
+ )
+ except Exception as rewrite_error:
+ raise RuntimeError(
+ "{} {}".format(conflict, rewrite_error)
+ ) from conflict
+ if result is None:
+ raise conflict
+
+ current_updates = result.update_messages
+ elapsed = int(time.time() * 1000) - start_millis
+ if elapsed > table.options.commit_timeout():
+ raise conflict
+
+ logger.info(
+ "Rewrote %d stale self-merge file(s) against snapshot %d "
+ "before retrying commit to table %s.",
+ result.rewritten_file_count,
+ latest_snapshot.id,
+ table.identifier,
+ )
+ _retry_wait(table, retry_count)
+ retry_count += 1
+
+
+def _rewrite_updates(
+ table,
+ update_messages: List[CommitMessage],
+ latest_snapshot,
+ *,
+ num_partitions: int,
+ ray_remote_args=None,
+) -> Optional[_RewriteResult]:
+ if table.options.deletion_vectors_enabled(False):
+ return None
+ if latest_snapshot.next_row_id is None:
+ return None
+ if any(
+ message.deleted_files or message.changelog_files
+ for message in update_messages
+ ):
+ return None
+
+ base_snapshot_ids = _base_snapshot_ids(update_messages)
+ if len(base_snapshot_ids) != 1:
+ return None
+ base_snapshot = table.snapshot_manager().get_snapshot_by_id(
+ next(iter(base_snapshot_ids))
+ )
+ if (
+ base_snapshot is None
+ or base_snapshot.schema_id != latest_snapshot.schema_id
+ ):
+ return None
+
+ # The failed commit already checked for logical conflicts up to the
+ # snapshot it observed. Repeat that check against the snapshot selected
+ # for this rewrite to close the race between those two reads of latest.
+ _validate_no_logical_conflict(
+ table,
+ update_messages,
+ latest_snapshot,
+ min(base_snapshot_ids),
+ )
+
+ scan_table = table.copy_without_time_travel({
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): str(latest_snapshot.id),
+ })
+ scan_plan = scan_table.new_read_builder().new_scan().plan_for_write()
+ if scan_plan.snapshot_id != latest_snapshot.id:
+ return None
+ current_splits = list(scan_plan.splits())
+ current_files = [
+ (split, file)
+ for split in current_splits
+ for file in split.files
+ if _is_normal_row_id_file(file)
+ ]
+ current_exact_ranges = {
+ _range_key(tuple(split.partition.values), split.bucket, file)
+ for split, file in current_files
+ }
+
+ staged = [
+ _StagedFile(index, message, file)
+ for index, message in enumerate(update_messages)
+ for file in message.new_files
+ ]
+ if any(
+ _is_dedicated_file(item.file)
+ and item.file.first_row_id is not None
+ and item.file.first_row_id < latest_snapshot.next_row_id
+ for item in staged
+ ):
+ return None
+
+ candidates = [
+ item for item in staged
+ if _is_rewrite_candidate(
+ item,
+ current_exact_ranges,
+ latest_snapshot.next_row_id,
+ )
+ ]
+ if not candidates:
+ return None
+ if not _ranges_are_still_covered(current_files, candidates):
+ return None
+
+ candidate_ids = {id(candidate.file) for candidate in candidates}
+
+ remaining_messages = []
+ for message in update_messages:
+ kept_files = [
+ file for file in message.new_files
+ if id(file) not in candidate_ids
+ ]
+ remaining = replace(
+ message,
+ new_files=kept_files,
+ check_from_snapshot=latest_snapshot.id,
+ )
+ if not remaining.is_empty():
+ remaining_messages.append(remaining)
+
+ rewritten_messages = []
+ groups: Dict[Tuple[str, ...], List[_StagedFile]] = {}
+ for candidate in candidates:
+ groups.setdefault(tuple(candidate.file.write_cols), []).append(
+ candidate
+ )
+ for columns, files in groups.items():
+ messages, rewritten_rows, _ = _rewrite_group(
+ table,
+ list(columns),
+ files,
+ latest_snapshot.id,
+ num_partitions=num_partitions,
+ ray_remote_args=ray_remote_args,
+ )
+ expected_rows = sum(item.file.row_count for item in files)
+ if rewritten_rows != expected_rows:
+ raise RuntimeError(
+ "Distributed row-id conflict rewrite read {} rows from "
+ "staged files, expected {}.".format(
+ rewritten_rows,
+ expected_rows,
+ )
+ )
+ rewritten_messages.extend(messages)
+
+ return _RewriteResult(
+ update_messages=remaining_messages + rewritten_messages,
+ rewritten_file_count=len(candidates),
+ )
+
+
+def _base_snapshot_ids(update_messages: Sequence[CommitMessage]):
+ return {
+ message.check_from_snapshot
+ for message in update_messages
+ if message.check_from_snapshot is not None
+ and message.check_from_snapshot >= 0
+ }
+
+
+def _rewrite_group(
+ table,
+ columns: List[str],
+ candidates: List[_StagedFile],
+ snapshot_id: int,
+ *,
+ num_partitions: int,
+ ray_remote_args=None,
+):
+ import ray
+
+ from pypaimon.ray.data_evolution_merge_join import (
+ distributed_update_apply,
+ )
+ from pypaimon.read.datasource.ray_datasource import RayDatasource
+ from pypaimon.read.datasource.split_provider import (
+ PreResolvedSplitProvider,
+ )
+
+ read_type = [table.field_dict[name] for name in columns]
+ read_type.append(SpecialFields.ROW_ID)
+ splits = [
+ DataSplit(
+ files=[candidate.file],
+ partition=GenericRow(
+ list(candidate.message.partition),
+ table.partition_keys_fields,
+ ),
+ bucket=candidate.message.bucket,
+ raw_convertible=True,
+ )
+ for candidate in candidates
+ ]
+ provider = PreResolvedSplitProvider(
+ table,
+ splits,
+ read_type,
+ )
+ parallelism = max(1, min(num_partitions, len(splits)))
+ staged_updates = ray.data.read_datasource(
+ RayDatasource(provider),
+ ray_remote_args=ray_remote_args,
+ concurrency=parallelism,
+ override_num_blocks=parallelism,
+ )
+ return distributed_update_apply(
+ staged_updates,
+ table,
+ columns,
+ num_partitions=num_partitions,
+ ray_remote_args=ray_remote_args,
+ base_snapshot_id=snapshot_id,
+ )
+
+
+def _validate_no_logical_conflict(
+ table,
+ update_messages: List[CommitMessage],
+ latest_snapshot,
+ base_snapshot_id: int,
+) -> None:
+ """Reject non-compaction changes before advancing the rewrite baseline."""
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ file_store_commit = commit.file_store_commit
+ file_store_commit.conflict_detection._row_id_check_from_snapshot = (
+ base_snapshot_id
+ )
+ entries = file_store_commit._collect_manifest_entries(update_messages)
+ conflict =
file_store_commit.conflict_detection.check_row_id_from_snapshot(
+ latest_snapshot,
+ entries,
+ check_compaction=False,
+ )
+ if conflict is not None:
+ raise conflict
+ finally:
+ commit.close()
+
+
+def _find_row_id_conflict(error) -> Optional[RowIdExistenceConflict]:
+ seen = set()
+ current = error
+ while current is not None and id(current) not in seen:
+ seen.add(id(current))
+ if isinstance(current, RowIdExistenceConflict):
+ return current
+ current = (
+ getattr(current, "cause", None)
+ or getattr(current, "__cause__", None)
+ )
+ return None
+
+
+def _is_rewrite_candidate(
+ item: _StagedFile,
+ current_exact_ranges,
+ next_row_id: int,
+) -> bool:
+ file = item.file
+ return (
+ _is_normal_row_id_file(file)
+ and file.first_row_id < next_row_id
+ and bool(file.write_cols)
+ and not any(
+ SpecialFields.is_system_field(name)
+ for name in file.write_cols
+ )
+ and _range_key(item.message.partition, item.message.bucket, file)
+ not in current_exact_ranges
+ )
+
+
+def _ranges_are_still_covered(current_files, candidates) -> bool:
+ current_ranges = {}
+ for split, file in current_files:
+ key = (tuple(split.partition.values), split.bucket)
+ current_ranges.setdefault(key, []).append(file.row_id_range())
+ current_ranges = {
+ key: Range.sort_and_merge_overlap(ranges, True, True)
+ for key, ranges in current_ranges.items()
+ }
+ for candidate in candidates:
+ key = (tuple(candidate.message.partition), candidate.message.bucket)
+ if candidate.file.row_id_range().exclude(
+ current_ranges.get(key, [])):
+ return False
+ return True
+
+
+def _range_key(partition, bucket, file):
+ return (
+ tuple(partition),
+ bucket,
+ file.first_row_id,
+ file.row_count,
+ )
+
+
+def _is_normal_row_id_file(file) -> bool:
+ return file.first_row_id is not None and not _is_dedicated_file(file)
+
+
+def _is_dedicated_file(file) -> bool:
+ return (
+ DataFileMeta.is_blob_file(file.file_name)
+ or DataFileMeta.is_vector_file(file.file_name)
+ )
+
+
+def _retry_wait(table, retry_count: int) -> None:
+ wait_millis = min(
+ table.options.commit_min_retry_wait() * (2 ** retry_count),
+ table.options.commit_max_retry_wait(),
+ )
+ jitter = random.randint(0, max(1, int(wait_millis * 0.2)))
+ time.sleep((wait_millis + jitter) / 1000.0)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 3576d659cb..df1992cc71 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -32,9 +32,10 @@ from pypaimon.table.row.generic_row import GenericRow,
GenericRowSerializer
from pypaimon.write.commit.row_id_conflict_rewriter import RowIdRewriteResult
from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.file_store_commit import (
+ CommitFailRetryResult,
FileStoreCommit,
ManifestMergeResult,
- RetryResult,
+ RollbackRetryResult,
RewriteResult,
_abort_commit_messages,
_try_replace_manifest_files,
@@ -355,6 +356,77 @@ class TestFileStoreCommit(unittest.TestCase):
for manifest in result.merge_after_manifests],
)
+ def test_conflict_rollback_retry_skips_history_and_rescans_base(
+ self, mock_manifest_list_manager, mock_manifest_file_manager):
+ file_store_commit = self._create_file_store_commit()
+ conflict = RuntimeError("conflicting compaction")
+ read_all = Mock(return_value=[])
+
file_store_commit.commit_scanner.read_all_entries_from_changed_partitions =
read_all
+ file_store_commit.conflict_detection.check_conflicts = Mock(
+ return_value=conflict
+ )
+ file_store_commit.rollback = Mock()
+ file_store_commit.rollback.try_to_rollback.return_value = True
+
+ result = file_store_commit._try_commit_once(
+ retry_result=None,
+ commit_kind="APPEND",
+ commit_entries=[Mock()],
+ changelog_entries=[],
+ commit_identifier=11,
+ latest_snapshot=Mock(id=3),
+ detect_conflicts=True,
+ allow_rollback=True,
+ )
+
+ self.assertIsInstance(result, RollbackRetryResult)
+ self.assertIs(result.exception, conflict)
+
+ file_store_commit.snapshot_manager.get_snapshot_by_id.side_effect = (
+ AssertionError("rollback retry must not scan snapshot history")
+ )
+ file_store_commit.commit_scanner.read_incremental_changes = Mock(
+ side_effect=AssertionError(
+ "rollback retry must not reuse the previous conflict base"
+ )
+ )
+ read_all.reset_mock()
+ file_store_commit.conflict_detection.check_conflicts.return_value = (
+ RuntimeError("conflict after rollback retry")
+ )
+
+ with self.assertRaisesRegex(
+ RuntimeError, "conflict after rollback retry"):
+ file_store_commit._try_commit_once(
+ retry_result=result,
+ commit_kind="APPEND",
+ commit_entries=[Mock()],
+ changelog_entries=[],
+ commit_identifier=11,
+ latest_snapshot=Mock(id=2),
+ detect_conflicts=True,
+ )
+
+
file_store_commit.snapshot_manager.get_snapshot_by_id.assert_not_called()
+
file_store_commit.commit_scanner.read_incremental_changes.assert_not_called()
+ read_all.assert_called_once()
+
+ def test_commit_fail_retry_missing_snapshot_still_fails_closed(
+ self, mock_manifest_list_manager, mock_manifest_file_manager):
+ file_store_commit = self._create_file_store_commit()
+ file_store_commit.snapshot_manager.get_snapshot_by_id.return_value =
None
+
+ with self.assertRaisesRegex(
+ RuntimeError, "snapshot 1 cannot be found"):
+ file_store_commit._is_duplicate_commit(
+ CommitFailRetryResult(None),
+ Mock(id=3),
+ 11,
+ "APPEND",
+ )
+
+
file_store_commit.snapshot_manager.get_snapshot_by_id.assert_called_once_with(1)
+
def _run_manifest_commit_attempt(self, commit_side_effect=None,
commit_result=None, retry_result=None,
existing_manifests=None,
@@ -429,7 +501,7 @@ class TestFileStoreCommit(unittest.TestCase):
file_store_commit, result = self._run_manifest_commit_attempt(
commit_result=False)
- self.assertIsInstance(result, RetryResult)
+ self.assertIsInstance(result, CommitFailRetryResult)
self.assertIsNone(result.exception)
self.assertEqual(
['before'],
@@ -449,7 +521,7 @@ class TestFileStoreCommit(unittest.TestCase):
file_store_commit, result = self._run_manifest_commit_attempt(
commit_side_effect=failure)
- self.assertIsInstance(result, RetryResult)
+ self.assertIsInstance(result, CommitFailRetryResult)
self.assertIs(failure, result.exception)
self.assertTrue(result.commit_result_may_be_uncertain)
self.assertIsNone(result.manifest_merge_result)
@@ -463,7 +535,7 @@ class TestFileStoreCommit(unittest.TestCase):
self._manifest_meta('before-b'),
]
previous_after = [self._manifest_meta('merged')]
- retry_result = RetryResult(
+ retry_result = CommitFailRetryResult(
Mock(id=3),
manifest_merge_result=ManifestMergeResult(
previous_before, previous_after),
@@ -481,7 +553,7 @@ class TestFileStoreCommit(unittest.TestCase):
existing_manifests=current,
)
- self.assertIsInstance(result, RetryResult)
+ self.assertIsInstance(result, CommitFailRetryResult)
self.assertEqual(
['prefix', 'before-a', 'before-b', 'suffix'],
[manifest.file_name for manifest
@@ -507,7 +579,7 @@ class TestFileStoreCommit(unittest.TestCase):
self._manifest_meta('before-a'),
self._manifest_meta('before-b'),
]
- retry_result = RetryResult(
+ retry_result = CommitFailRetryResult(
Mock(id=3),
manifest_merge_result=ManifestMergeResult(
previous_before, [self._manifest_meta('merged')]),
@@ -524,7 +596,7 @@ class TestFileStoreCommit(unittest.TestCase):
existing_manifests=current,
)
- self.assertIsInstance(result, RetryResult)
+ self.assertIsInstance(result, CommitFailRetryResult)
self.assertIsNone(result.manifest_merge_result)
file_store_commit.manifest_file_merger.merge.assert_not_called()
base_manifests = (
diff --git a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
index e18439ea84..da20c300bc 100644
--- a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
+++ b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
@@ -209,9 +209,9 @@ class OverwriteCommitConflictTest(unittest.TestCase):
self.assertEqual(sorted(actual[actual['f0'] == 1]['f1'].tolist()),
['new'])
self.assertEqual(sorted(actual[actual['f0'] == 2]['f1'].tolist()),
['c'])
- def test_falls_back_to_full_scan_when_intermediate_snapshot_missing(self):
- # A missing intermediate snapshot -> read_incremental_changes returns
None
- # and the retry falls back to a full scan.
+ def test_fails_when_intermediate_snapshot_missing(self):
+ # Match Java SnapshotManager.snapshot: duplicate detection cannot skip
+ # an unavailable snapshot in the retry history.
K = 1
missing_id = self.table.snapshot_manager().get_latest_snapshot().id + 1
@@ -239,8 +239,8 @@ class OverwriteCommitConflictTest(unittest.TestCase):
fsc.commit_scanner.read_all_entries_from_changed_partitions = spy_full
fsc.commit_scanner.read_incremental_changes = spy_incr
- # Only the scanner's lookups see missing_id as absent; the commit's own
- # manager (bound earlier) is untouched.
+ # Both duplicate detection and incremental conflict scanning observe
+ # the expired snapshot.
real_mgr = fsc.commit_scanner.table.snapshot_manager()
class _Wrap:
@@ -250,7 +250,9 @@ class OverwriteCommitConflictTest(unittest.TestCase):
def get_snapshot_by_id(self, i):
return None if i == missing_id else
real_mgr.get_snapshot_by_id(i)
- fsc.commit_scanner.table.snapshot_manager = lambda: _Wrap()
+ wrapped_mgr = _Wrap()
+ fsc.snapshot_manager = wrapped_mgr
+ fsc.commit_scanner.table.snapshot_manager = lambda: wrapped_mgr
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
@@ -264,12 +266,15 @@ class OverwriteCommitConflictTest(unittest.TestCase):
fsc.snapshot_commit.commit = patched_cas
- c.commit(messages)
+ with self.assertRaisesRegex(
+ RuntimeError,
+ "snapshot {} cannot be found".format(missing_id)):
+ c.commit(messages)
c.close()
self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
- self.assertIn(None, incr_results) # incremental bailed on
missing
- self.assertEqual(full_scans['n'], 2) # first attempt + fallback
+ self.assertEqual([], incr_results)
+ self.assertEqual(full_scans['n'], 1)
def test_incremental_merge_across_non_append_snapshot(self):
self._assert_merge_equals_full_scan(self._overwrite_target)
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index 8e48773d4b..d75dbf4a39 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -114,6 +114,45 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
snap = table.snapshot_manager().get_latest_snapshot()
return snap.id if snap is not None else None
+ def _compact_all_data_files(self, table):
+ """Replace all current data files with one COMPACT output file."""
+ from pypaimon.table.special_fields import SpecialFields
+
+ read_builder = table.new_read_builder().with_projection(
+ list(table.field_names) + [SpecialFields.ROW_ID.name]
+ )
+ plan = read_builder.new_scan().plan_for_write()
+ old_files = [
+ file for split in plan.splits() for file in split.files
+ ]
+ current = read_builder.new_read().to_arrow(plan.splits()).sort_by(
+ [(SpecialFields.ROW_ID.name, 'ascending')]
+ ).select(list(table.field_names))
+
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(current)
+ messages = writer.prepare_commit()
+ self.assertEqual(len(messages), 1)
+ self.assertEqual(len(messages[0].new_files), 1)
+ messages[0].new_files = [
+ messages[0].new_files[0].assign_first_row_id(0)
+ ]
+ messages[0].deleted_files.extend(old_files)
+
+ commit = write_builder.new_commit()
+ file_store_commit = commit.file_store_commit
+ original_try_commit = file_store_commit._try_commit
+ file_store_commit._try_commit = (
+ lambda commit_kind, *args, **kwargs:
+ original_try_commit('COMPACT', *args, **kwargs)
+ )
+ try:
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
def _merge_and_capture_self_merge_plan(self, **kwargs):
from pypaimon.ray.data_evolution_merge_join import (
build_self_merge_update_plan as real_build_plan,
@@ -2089,6 +2128,674 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertEqual(result['num_matched'], 4)
self.assertEqual(self._read_sorted(target)['age'], [99, 99, 99, 99])
+ def test_self_merge_rebases_staged_updates_after_compaction(self):
+ from pypaimon.data.generic_variant import GenericVariant
+ from pypaimon.data.variant_path import variant_get, variant_replace
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+
+ options = dict(self.de_options)
+ options.update({
+ 'commit.max-retries': '0',
+ 'data-evolution.row-id-conflict-rewrite.max-size': '1 B',
+ })
+ variant_type = pa.struct([
+ pa.field('value', pa.binary(), nullable=False),
+ pa.field('metadata', pa.binary(), nullable=False),
+ ])
+ schema = pa.schema([
+ ('id', pa.int32()),
+ ('payload', variant_type),
+ ('topic_schema', pa.string()),
+ ])
+ target = 'default.tbl_{}'.format(uuid.uuid4().hex[:8])
+ self.catalog.create_table(
+ target,
+ Schema.from_pyarrow_schema(schema, options=options),
+ False,
+ )
+
+ def payload(values):
+ return GenericVariant.to_arrow_array([
+ GenericVariant.from_python({
+ 'angular_velocity': {'y': value, 'z': value + 1.0},
+ 'linear_acceleration': {
+ 'y': value + 2.0,
+ 'z': value + 3.0,
+ },
+ })
+ for value in values
+ ])
+
+ self._write(
+ target,
+ pa.table({
+ 'id': pa.array([1, 2], type=pa.int32()),
+ 'payload': payload([1.0, 10.0]),
+ 'topic_schema': ['old', 'old'],
+ }, schema=schema),
+ )
+ self._write(
+ target,
+ pa.table({
+ 'id': pa.array([3, 4], type=pa.int32()),
+ 'payload': payload([20.0, 30.0]),
+ 'topic_schema': ['old', 'old'],
+ }, schema=schema),
+ )
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+ stale_paths = []
+
+ def stage_then_compact(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ stale_paths.extend(
+ file.file_path
+ for message in result[0]
+ for file in message.new_files
+ )
+ self._compact_all_data_files(table)
+ return result
+
+ path_types = {
+ '$.angular_velocity.y': pa.float64(),
+ '$.angular_velocity.z': pa.float64(),
+ '$.linear_acceleration.y': pa.float64(),
+ '$.linear_acceleration.z': pa.float64(),
+ }
+
+ def negate_imu_yz(rows):
+ values = variant_get(rows['payload'], path_types)
+ return variant_replace(
+ rows['payload'],
+ {
+ path: pc.negate(value)
+ for path, value in values.items()
+ },
+ strict=True,
+ )
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_compact,
+ ), patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
+ ):
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ read_columns=['payload'],
+ when_matched=[WhenMatched.update({
+ 'payload': negate_imu_yz,
+ 'topic_schema': lit('imu-yz-negated-v1'),
+ })],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 4)
+ output = self._read_sorted(target)
+ decoded = [
+ GenericVariant.from_arrow_struct(value).to_python()
+ for value in output['payload']
+ ]
+ self.assertEqual(
+ [row['angular_velocity']['y'] for row in decoded],
+ [-1.0, -10.0, -20.0, -30.0],
+ )
+ self.assertEqual(
+ output['topic_schema'],
+ ['imu-yz-negated-v1'] * 4,
+ )
+ self.assertTrue(stale_paths)
+ # Match Spark: replaced staging files are left for orphan cleanup.
+ self.assertTrue(all(os.path.exists(path) for path in stale_paths))
+
+ def test_self_merge_compaction_retry_checks_core_rollback(self):
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+ from pypaimon.ray import row_id_conflict_rewriter as rewriter_module
+ from pypaimon.write.file_store_commit import FileStoreCommit
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ self._write(target, self._source(ids=(3, 4)))
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+ real_rewrite = rewriter_module._rewrite_updates
+ real_commit_init = FileStoreCommit.__init__
+ rollbacks = []
+
+ def stage_then_compact(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ self._compact_all_data_files(table)
+ return result
+
+ rewrite_calls = [0]
+
+ def miss_precommit_race(*args, **kwargs):
+ rewrite_calls[0] += 1
+ if rewrite_calls[0] == 1:
+ return None
+ return real_rewrite(*args, **kwargs)
+
+ def init_with_rollback(commit, *args, **kwargs):
+ real_commit_init(commit, *args, **kwargs)
+ commit.rollback = Mock()
+ commit.rollback.try_to_rollback.return_value = False
+ rollbacks.append(commit.rollback)
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_compact,
+ ), patch.object(
+ rewriter_module,
+ '_rewrite_updates',
+ side_effect=miss_precommit_race,
+ ), patch.object(
+ FileStoreCommit,
+ '__init__',
+ new=init_with_rollback,
+ ), patch.object(
+ FileStoreCommit,
+ '_commit_retry_wait',
+ ) as commit_retry_wait, patch.object(
+ rewriter_module,
+ '_retry_wait',
+ ) as ray_retry_wait:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update({'age': lit(99)})],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 4)
+ self.assertEqual(self._read_sorted(target)['age'], [99, 99, 99, 99])
+ commit_retry_wait.assert_not_called()
+ self.assertTrue(any(
+ rollback.try_to_rollback.called for rollback in rollbacks
+ ))
+ ray_retry_wait.assert_called_once()
+ self.assertEqual(ray_retry_wait.call_args[0][1], 0)
+ self.assertEqual(rewrite_calls[0], 2)
+
+ def test_self_merge_rebases_again_after_second_compaction(self):
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+ from pypaimon.ray import row_id_conflict_rewriter as rewriter_module
+ from pypaimon.write.write_builder import BatchWriteBuilder
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ self._write(target, self._source(ids=(3, 4)))
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+ real_rewrite = rewriter_module._rewrite_updates
+ real_new_commit = BatchWriteBuilder.new_commit
+ coordinator_commit_users = []
+
+ def capture_new_commit(write_builder):
+ commit = real_new_commit(write_builder)
+ if (
+ write_builder.table.options
+ .data_evolution_row_id_conflict_rewrite_max_size() == 0
+ ):
+ coordinator_commit_users.append(write_builder.commit_user)
+ return commit
+
+ def stage_then_compact(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ self._compact_all_data_files(table)
+ return result
+
+ rewrite_snapshot_ids = []
+
+ def miss_precommit_race(
+ rewrite_table, messages, latest_snapshot, **kwargs):
+ rewrite_snapshot_ids.append(latest_snapshot.id)
+ if len(rewrite_snapshot_ids) == 1:
+ return None
+ return real_rewrite(
+ rewrite_table,
+ messages,
+ latest_snapshot,
+ **kwargs,
+ )
+
+ retry_counts = []
+
+ def compact_before_first_retry(_table, retry_count):
+ retry_counts.append(retry_count)
+ if retry_count == 0:
+ # Change the current row-id boundary before the second
+ # compaction so the next commit observes the same
+ # RowIdExistenceConflict used by Spark's retry loop.
+ self._write(target, self._source(ids=(5, 6)))
+ self._compact_all_data_files(table)
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_compact,
+ ), patch.object(
+ rewriter_module,
+ '_rewrite_updates',
+ side_effect=miss_precommit_race,
+ ), patch.object(
+ rewriter_module,
+ '_retry_wait',
+ side_effect=compact_before_first_retry,
+ ), patch.object(
+ BatchWriteBuilder,
+ 'new_commit',
+ new=capture_new_commit,
+ ):
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update({'age': lit(99)})],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 4)
+ output = self._read_sorted(target)
+ self.assertEqual(output['id'], [1, 2, 3, 4, 5, 6])
+ self.assertEqual(output['age'], [99, 99, 99, 99, 10, 10])
+ self.assertEqual(retry_counts, [0, 1])
+ self.assertEqual(len(rewrite_snapshot_ids), 3)
+ self.assertEqual(rewrite_snapshot_ids[0], rewrite_snapshot_ids[1])
+ self.assertGreater(rewrite_snapshot_ids[2], rewrite_snapshot_ids[1])
+ self.assertEqual(len(coordinator_commit_users), 3)
+ self.assertEqual(len(set(coordinator_commit_users)), 1)
+
+ def test_self_merge_uncertain_commit_then_compaction_is_duplicate(self):
+ from pypaimon.ray import row_id_conflict_rewriter as rewriter_module
+ from pypaimon.write.write_builder import BatchWriteBuilder
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ self._write(target, self._source(ids=(3, 4)))
+ table = self.catalog.get_table(target)
+ base_snapshot_id = self._snapshot_id(target)
+ real_new_commit = BatchWriteBuilder.new_commit
+ coordinator_commit_users = []
+ duplicate_results = []
+ atomic_attempts = []
+ injected = [False]
+
+ def inject_uncertain_commit(write_builder):
+ commit = real_new_commit(write_builder)
+ if (
+ injected[0]
+ or write_builder.table.options
+ .data_evolution_row_id_conflict_rewrite_max_size() != 0
+ ):
+ return commit
+
+ injected[0] = True
+ coordinator_commit_users.append(write_builder.commit_user)
+ file_store_commit = commit.file_store_commit
+ real_atomic_commit = file_store_commit.snapshot_commit.commit
+ real_duplicate_check = file_store_commit._is_duplicate_commit
+
+ def track_duplicate(*args, **kwargs):
+ result = real_duplicate_check(*args, **kwargs)
+ duplicate_results.append(result)
+ return result
+
+ def commit_then_compact_and_timeout(
+ base_uuid, snapshot, statistics):
+ atomic_attempts.append(snapshot.id)
+ self.assertTrue(real_atomic_commit(
+ base_uuid, snapshot, statistics,
+ ))
+ self._compact_all_data_files(table)
+ raise TimeoutError('lost snapshot commit response')
+
+ file_store_commit._is_duplicate_commit = track_duplicate
+ file_store_commit.snapshot_commit.commit = (
+ commit_then_compact_and_timeout
+ )
+ file_store_commit._commit_retry_wait = Mock()
+ return commit
+
+ with patch.object(
+ BatchWriteBuilder,
+ 'new_commit',
+ new=inject_uncertain_commit,
+ ), patch.object(
+ rewriter_module,
+ '_rewrite_updates',
+ wraps=rewriter_module._rewrite_updates,
+ ) as rewrite_updates:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update({'age': lit(99)})],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertTrue(injected[0])
+ self.assertEqual(result['num_matched'], 4)
+ self.assertEqual(self._read_sorted(target)['age'], [99, 99, 99, 99])
+ self.assertEqual(atomic_attempts, [base_snapshot_id + 1])
+ self.assertEqual(duplicate_results, [False, True])
+ self.assertEqual(len(coordinator_commit_users), 1)
+ self.assertEqual(
+ table.snapshot_manager().get_snapshot_by_id(
+ base_snapshot_id + 1
+ ).commit_user,
+ coordinator_commit_users[0],
+ )
+ self.assertEqual(
+ table.snapshot_manager().get_snapshot_by_id(
+ base_snapshot_id + 2
+ ).commit_kind,
+ 'COMPACT',
+ )
+ self.assertEqual(self._snapshot_id(target), base_snapshot_id + 2)
+ rewrite_updates.assert_not_called()
+
+ def test_self_merge_rejects_concurrent_overwrite(self):
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+ staging_paths = []
+
+ def stage_then_overwrite(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ staging_paths.extend(
+ file.external_path or file.file_path
+ for message in result[0]
+ for file in message.new_files
+ )
+ replacement = pa.Table.from_pydict(
+ {
+ 'id': pa.array([30, 40], type=pa.int32()),
+ 'name': ['replacement', 'replacement'],
+ 'age': pa.array([30, 40], type=pa.int32()),
+ },
+ schema=self.pa_schema,
+ )
+ write_builder = table.new_batch_write_builder().overwrite({})
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ try:
+ writer.write_arrow(replacement)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return result
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_overwrite,
+ ), self.assertRaises(Exception):
+ merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update({'age': lit(99)})],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ output = self._read_sorted(target)
+ self.assertEqual(output['id'], [30, 40])
+ self.assertEqual(output['age'], [30, 40])
+ self.assertTrue(staging_paths)
+ # Match Spark: failed staging files are left for orphan cleanup.
+ self.assertTrue(
+ all(os.path.exists(path) for path in staging_paths)
+ )
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_update_and_delete(self):
+ options = dict(self.de_options)
+ options['deletion-vectors.enabled'] = 'true'
+ target = self._create_table(options=options)
+ self._write(target, self._source(ids=(1, 2, 3, 4)))
+
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[
+ WhenMatched.update(
+ {'age': lit(99)}, condition='t.id <= 2',
+ ),
+ WhenMatched.delete(condition='t.id = 3'),
+ ],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 3)
+ output = self._read_sorted(target)
+ self.assertEqual(output['id'], [1, 2, 4])
+ self.assertEqual(output['age'], [99, 99, 10])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_update_delete_does_not_rebase_after_compaction(self):
+ from pypaimon.ray import row_id_conflict_rewriter as rewriter_module
+
+ options = dict(self.de_options)
+ options['deletion-vectors.enabled'] = 'true'
+ target = self._create_table(options=options)
+ self._write(target, self._source(ids=(1, 2, 3, 4)))
+ table = self.catalog.get_table(target)
+ real_commit = (
+ rewriter_module.commit_self_merge_with_compaction_retry
+ )
+ real_rewrite = rewriter_module._rewrite_updates
+ rewrite_results = []
+
+ def compact_then_commit(*args, **kwargs):
+ self._compact_all_data_files(table)
+ return real_commit(*args, **kwargs)
+
+ def assert_rewrite_disabled(*args, **kwargs):
+ result = real_rewrite(*args, **kwargs)
+ rewrite_results.append(result)
+ return result
+
+ with patch.object(
+ rewriter_module,
+ 'commit_self_merge_with_compaction_retry',
+ side_effect=compact_then_commit,
+ ), patch.object(
+ rewriter_module,
+ '_rewrite_updates',
+ side_effect=assert_rewrite_disabled,
+ ), self.assertRaises(Exception):
+ merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[
+ WhenMatched.update(
+ {'age': lit(99)}, condition='t.id <= 2',
+ ),
+ WhenMatched.delete(condition='t.id = 3'),
+ ],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertTrue(rewrite_results)
+ self.assertTrue(all(result is None for result in rewrite_results))
+ output = self._read_sorted(target)
+ self.assertEqual(output['id'], [1, 2, 3, 4])
+ self.assertEqual(output['age'], [10, 10, 10, 10])
+
+ def test_self_merge_compaction_rebase_keeps_logical_conflicts(self):
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ self._write(target, self._source(ids=(3, 4)))
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+
+ def stage_then_update_and_compact(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ write_builder = table.new_batch_write_builder()
+ update = write_builder.new_update()
+ predicate = update.new_predicate_builder().equal('id', 2)
+ messages = update.update_by_predicate(
+ predicate,
+ {'age': 100},
+ )
+ write_builder.new_commit().commit(messages)
+ self._compact_all_data_files(table)
+ return result
+
+ def increment_age(rows):
+ return pc.add(rows['age'], 1)
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_update_and_compact,
+ ), self.assertRaises(Exception) as ctx:
+ merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ read_columns=['age'],
+ when_matched=[WhenMatched.update({
+ 'age': increment_age,
+ })],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertIn("multiple 'MERGE INTO'", str(ctx.exception))
+ self.assertEqual(self._read_sorted(target)['age'][1], 100)
+
+ def test_self_merge_missing_logical_snapshot_fails_closed(self):
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+ from pypaimon.snapshot.snapshot_manager import SnapshotManager
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ self._write(target, self._source(ids=(3, 4)))
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+ real_get_snapshot = SnapshotManager.get_snapshot_by_id
+ hidden_snapshot = {'id': None}
+
+ def get_snapshot_except_hidden(manager, snapshot_id):
+ if snapshot_id == hidden_snapshot['id']:
+ return None
+ return real_get_snapshot(manager, snapshot_id)
+
+ def stage_then_update_and_compact(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ write_builder = table.new_batch_write_builder()
+ update = write_builder.new_update()
+ predicate = update.new_predicate_builder().equal('id', 2)
+ messages = update.update_by_predicate(predicate, {'age': 100})
+ write_builder.new_commit().commit(messages)
+ logical_snapshot_id = self._snapshot_id(target)
+ self._compact_all_data_files(table)
+ hidden_snapshot['id'] = logical_snapshot_id
+ return result
+
+ def increment_age(rows):
+ return pc.add(rows['age'], 1)
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_update_and_compact,
+ ), patch.object(
+ SnapshotManager,
+ 'get_snapshot_by_id',
+ new=get_snapshot_except_hidden,
+ ), self.assertRaisesRegex(
+ RuntimeError,
+ "snapshot .* cannot be found",
+ ):
+ merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ read_columns=['age'],
+ when_matched=[WhenMatched.update({
+ 'age': increment_age,
+ })],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(self._read_sorted(target)['age'][1], 100)
+
+ def test_self_merge_compaction_rebase_preserves_other_column_update(self):
+ from pypaimon.ray import data_evolution_merge_into as merge_module
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2)))
+ self._write(target, self._source(ids=(3, 4)))
+ table = self.catalog.get_table(target)
+ real_apply = merge_module.distributed_self_merge_update_apply
+
+ def stage_then_update_and_compact(*args, **kwargs):
+ result = real_apply(*args, **kwargs)
+ self._write(target, self._source(ids=(5,)))
+ write_builder = table.new_batch_write_builder()
+ update = write_builder.new_update()
+ predicate = update.new_predicate_builder().equal('id', 2)
+ messages = update.update_by_predicate(
+ predicate,
+ {'name': 'concurrent'},
+ )
+ write_builder.new_commit().commit(messages)
+ self._compact_all_data_files(table)
+ return result
+
+ def increment_age(rows):
+ return pc.add(rows['age'], 1)
+
+ with patch.object(
+ merge_module,
+ 'distributed_self_merge_update_apply',
+ side_effect=stage_then_update_and_compact,
+ ), patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
+ ):
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ read_columns=['age'],
+ when_matched=[WhenMatched.update({
+ 'age': increment_age,
+ })],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 4)
+ output = self._read_sorted(target)
+ self.assertEqual(output['age'], [11, 11, 11, 11, 10])
+ self.assertEqual(
+ output['name'],
+ ['x', 'concurrent', 'x', 'x', 'x'],
+ )
+
@unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
def test_self_merge_filters_file_group_in_batches(self):
from pypaimon.ray import data_evolution_merge_into as merge_module
diff --git a/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
b/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
new file mode 100644
index 0000000000..6b2ec0d82b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
@@ -0,0 +1,289 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import importlib.util
+import unittest
+from unittest.mock import Mock, patch
+
+from pypaimon.ray.data_evolution_merge_into import _reraise_inner
+from pypaimon.ray.row_id_conflict_rewriter import (
+ commit_self_merge_with_compaction_retry,
+)
+from pypaimon.write.commit.conflict_detection import RowIdExistenceConflict
+
+
+class RayRowIdConflictRewriterTest(unittest.TestCase):
+
+ @staticmethod
+ def _message(snapshot_id, name):
+ return Mock(check_from_snapshot=snapshot_id, name=name)
+
+ @staticmethod
+ def _conflict(name):
+ entry = Mock(bucket=0)
+ entry.file = Mock(
+ file_name=name,
+ first_row_id=0,
+ row_count=1,
+ )
+ return RowIdExistenceConflict(entry)
+
+ @staticmethod
+ def _table(commits, snapshots, max_retries=3):
+ builder = Mock()
+ builder.new_commit.side_effect = commits
+ commit_table = Mock()
+ commit_table.new_batch_write_builder.return_value = builder
+ table = Mock()
+ table.copy_without_time_travel.return_value = commit_table
+ table.snapshot_manager.return_value.get_latest_snapshot.side_effect = (
+ snapshots
+ )
+ table.options.commit_timeout.return_value = 60_000
+ table.options.commit_max_retries.return_value = max_retries
+ return table
+
+ def test_nested_row_id_conflict_is_rebased(self):
+ entry = Mock(bucket=0)
+ entry.file = Mock(
+ file_name='data-file.parquet',
+ first_row_id=0,
+ row_count=1,
+ )
+ conflict = RowIdExistenceConflict(entry)
+ commit_error = RuntimeError('commit failed')
+ commit_error.__cause__ = conflict
+
+ commit = Mock()
+ commit.commit.side_effect = [commit_error, None]
+ table = self._table(
+ [commit, commit],
+ [Mock(id=1), Mock(id=2)],
+ )
+ rewrite_result = Mock(
+ update_messages=[],
+ rewritten_file_count=1,
+ )
+
+ with patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._rewrite_updates',
+ return_value=rewrite_result,
+ ) as rewrite_updates, patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
+ ):
+ commit_self_merge_with_compaction_retry(
+ table,
+ [],
+ [],
+ num_partitions=1,
+ )
+
+ rewrite_updates.assert_called_once()
+ self.assertEqual(2, commit.commit.call_count)
+ self.assertEqual(2, commit.close.call_count)
+
+ def test_commit_preserves_file_store_rollback(self):
+ rollback = Mock()
+ commit = Mock()
+ commit.file_store_commit.rollback = rollback
+ table = self._table([commit], [Mock(id=1)])
+
+ commit_self_merge_with_compaction_retry(
+ table,
+ [],
+ [],
+ num_partitions=1,
+ )
+
+ self.assertIs(commit.file_store_commit.rollback, rollback)
+
+ def test_rewrite_error_keeps_row_id_conflict_as_cause(self):
+ conflict = self._conflict('compact')
+ rewrite_error = ValueError('rewrite failed')
+ update = self._message(1, 'update')
+ commit = Mock()
+ commit.commit.side_effect = conflict
+ table = self._table(
+ [commit],
+ [Mock(id=1), Mock(id=2)],
+ )
+
+ with patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._rewrite_updates',
+ side_effect=rewrite_error,
+ ), self.assertRaises(RuntimeError) as context:
+ commit_self_merge_with_compaction_retry(
+ table,
+ [update],
+ [],
+ num_partitions=1,
+ )
+
+ self.assertIs(conflict, context.exception.__cause__)
+
+ def test_public_error_does_not_unwrap_regular_cause(self):
+ timeout = TimeoutError('commit timeout')
+ outer = RuntimeError('commit result is uncertain')
+ outer.__cause__ = timeout
+
+ with self.assertRaises(RuntimeError) as context:
+ _reraise_inner(outer)
+
+ self.assertIs(outer, context.exception)
+
+ @unittest.skipUnless(
+ importlib.util.find_spec('ray') is not None,
+ 'Ray is not installed.',
+ )
+ def test_public_error_unwraps_ray_task_error(self):
+ from ray.exceptions import RayTaskError
+
+ cause = ValueError('worker failure')
+ ray_error = RayTaskError(
+ 'worker',
+ 'Traceback (most recent call last):\nValueError: worker failure',
+ cause,
+ proctitle='ray::worker',
+ pid=1,
+ ip='127.0.0.1',
+ ).as_instanceof_cause()
+
+ with self.assertRaises(ValueError) as context:
+ _reraise_inner(ray_error)
+
+ self.assertIs(cause, context.exception)
+
+ def test_final_error_after_multiple_rebases_does_not_abort_messages(self):
+ generation_0 = self._message(1, 'generation-0')
+ generation_1 = self._message(2, 'generation-1')
+ generation_2 = self._message(3, 'generation-2')
+ other = self._message(-1, 'insert')
+ snapshot_1 = Mock(id=1, uuid='uuid-1')
+ snapshot_2 = Mock(id=2, uuid='uuid-2')
+ snapshot_3 = Mock(id=3, uuid='uuid-3')
+ commits = [Mock(), Mock(), Mock()]
+ commits[0].commit.side_effect = self._conflict('compact-1')
+ commits[1].commit.side_effect = self._conflict('compact-2')
+ terminal = RuntimeError('final attempt failed')
+ commits[2].commit.side_effect = terminal
+ table = self._table(
+ commits,
+ [snapshot_1, snapshot_2, snapshot_3],
+ )
+
+ with patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._rewrite_updates',
+ side_effect=[
+ Mock(
+ update_messages=[generation_1],
+ rewritten_file_count=1,
+ ),
+ Mock(
+ update_messages=[generation_2],
+ rewritten_file_count=1,
+ ),
+ ],
+ ), patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
+ ), patch(
+ 'pypaimon.write.file_store_commit._abort_commit_messages',
+ ) as abort_messages, self.assertRaises(RuntimeError) as context:
+ commit_self_merge_with_compaction_retry(
+ table,
+ [generation_0],
+ [other],
+ num_partitions=1,
+ )
+
+ self.assertIs(terminal, context.exception)
+ abort_messages.assert_not_called()
+
+ def test_exhausted_deterministic_conflict_does_not_abort_messages(self):
+ generation_0 = self._message(1, 'generation-0')
+ generation_1 = self._message(2, 'generation-1')
+ other = self._message(-1, 'insert')
+ snapshot_1 = Mock(id=1, uuid='uuid-1')
+ snapshot_2 = Mock(id=2, uuid='uuid-2')
+ commits = [Mock(), Mock()]
+ commits[0].commit.side_effect = self._conflict('compact-1')
+ terminal = self._conflict('compact-2')
+ commits[1].commit.side_effect = terminal
+ table = self._table(
+ commits,
+ [snapshot_1, snapshot_2],
+ max_retries=1,
+ )
+
+ with patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._rewrite_updates',
+ return_value=Mock(
+ update_messages=[generation_1],
+ rewritten_file_count=1,
+ ),
+ ), patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
+ ), patch(
+ 'pypaimon.write.file_store_commit._abort_commit_messages',
+ ) as abort_messages, self.assertRaises(
+ RowIdExistenceConflict,
+ ) as context:
+ commit_self_merge_with_compaction_retry(
+ table,
+ [generation_0],
+ [other],
+ num_partitions=1,
+ )
+
+ self.assertIs(terminal, context.exception)
+ abort_messages.assert_not_called()
+
+ def test_unknown_final_error_does_not_abort_messages(self):
+ generation_0 = self._message(1, 'generation-0')
+ generation_1 = self._message(2, 'generation-1')
+ other = self._message(-1, 'insert')
+ snapshot_1 = Mock(id=1, uuid='uuid-1')
+ snapshot_2 = Mock(id=2, uuid='uuid-2')
+ commits = [Mock(), Mock()]
+ commits[0].commit.side_effect = self._conflict('compact')
+ unknown = RuntimeError('callback failed after commit')
+ commits[1].commit.side_effect = unknown
+ table = self._table(commits, [snapshot_1, snapshot_2])
+
+ with patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._rewrite_updates',
+ return_value=Mock(
+ update_messages=[generation_1],
+ rewritten_file_count=1,
+ ),
+ ), patch(
+ 'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
+ ), patch(
+ 'pypaimon.write.file_store_commit._abort_commit_messages',
+ ) as abort_messages, self.assertRaises(RuntimeError) as context:
+ commit_self_merge_with_compaction_retry(
+ table,
+ [generation_0],
+ [other],
+ num_partitions=1,
+ )
+
+ self.assertIs(unknown, context.exception)
+ abort_messages.assert_not_called()
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py
b/paimon-python/pypaimon/tests/reader_append_only_test.py
index 4a7a266f70..3988359ad9 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -32,7 +32,7 @@ from pypaimon.common.options.core_options import CoreOptions
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.table.row.generic_row import GenericRow
-from pypaimon.write.file_store_commit import RetryResult
+from pypaimon.write.file_store_commit import CommitFailRetryResult
class AoReaderTest(unittest.TestCase):
@@ -669,7 +669,7 @@ class AoReaderTest(unittest.TestCase):
))
# mock retry
success = table_commit.file_store_commit._try_commit_once(
- RetryResult(None),
+ CommitFailRetryResult(None),
"APPEND",
commit_entries,
[],
diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
index 4d02a21252..36b2dfc053 100644
--- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
+++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
@@ -408,6 +408,20 @@ class TestCheckRowIdFromSnapshot(unittest.TestCase):
self.assertIsNone(
detection.check_row_id_from_snapshot(compact_snap,
self._blob_delta()))
+ def test_missing_intermediate_snapshot_fails_closed(self):
+ check_snap = _FakeSnapshot(1, "APPEND", next_row_id=200)
+ latest_snap = _FakeSnapshot(3, "COMPACT", next_row_id=200)
+ detection = self._make_detection(
+ [check_snap, latest_snap], {3: []})
+
+ with self.assertRaisesRegex(
+ RuntimeError, "snapshot 2 cannot be found"):
+ detection.check_row_id_from_snapshot(
+ latest_snap,
+ self._blob_delta(),
+ check_compaction=False,
+ )
+
def test_compact_no_conflict_when_no_matching_delete(self):
check_snap = _FakeSnapshot(1, "APPEND", next_row_id=400)
compact_snap = _FakeSnapshot(2, "COMPACT", next_row_id=400)
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 4009b25790..d7c12735e4 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -1541,9 +1541,9 @@ class TableWriteTest(unittest.TestCase):
commit_two.close()
commit_three.close()
- def test_uncertain_commit_then_cas_failure_keeps_files(self):
+ def test_uncertain_commit_with_unavailable_snapshot_fails_closed(self):
table = self._create_postpone_table(
- 'default.test_uncertain_commit_then_cas_failure',
+ 'default.test_uncertain_commit_with_unavailable_snapshot',
pa_schema=self.postpone_pa_schema,
partition_keys=['dt'],
primary_keys=['id', 'dt'],
@@ -1588,7 +1588,6 @@ class TableWriteTest(unittest.TestCase):
def hide_first_snapshot(snapshot_id):
return None if snapshot_id == 1 else
real_get_snapshot(snapshot_id)
- # Keep the retry on the CAS path after snapshot 1 becomes
unavailable.
with patch.object(
snapshot_commit,
'commit',
@@ -1601,15 +1600,20 @@ class TableWriteTest(unittest.TestCase):
file_store_commit.conflict_detection,
'check_conflicts',
return_value=None,
- ), patch.object(file_store_commit, '_commit_retry_wait'):
- with self.assertRaises(RuntimeError) as context:
+ ) as check_conflicts, patch.object(
+ file_store_commit,
+ '_commit_retry_wait',
+ ):
+ with self.assertRaisesRegex(
+ RuntimeError, 'snapshot 1 cannot be found'):
commit.commit(messages)
- self.assertIs(uncertain_error, context.exception.__cause__)
- self.assertEqual(2, attempts)
+ self.assertEqual(1, attempts)
+ check_conflicts.assert_called_once()
self.assertTrue(all(table.file_io.exists(path) for path in
data_paths))
self.assertEqual(
- [1, 2], self._read_sorted(table, 'id').column('id').to_pylist()
+ [1, 2], self._read_sorted(
+ table, 'id').column('id').to_pylist()
)
finally:
write.close()
diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py
b/paimon-python/pypaimon/write/commit/conflict_detection.py
index 4e6026f7b6..871013cf75 100644
--- a/paimon-python/pypaimon/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/write/commit/conflict_detection.py
@@ -696,7 +696,9 @@ class ConflictDetection:
latest_snapshot.id + 1):
snapshot = self.snapshot_manager.get_snapshot_by_id(snapshot_id)
if snapshot is None:
- continue
+ raise RuntimeError(
+ "Row-id conflict check cannot continue because snapshot "
+ "{} cannot be found.".format(snapshot_id))
if snapshot.commit_kind == "COMPACT":
if check_compaction:
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 7bb6db4dad..5f8f054aa8 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -150,7 +150,7 @@ class ManifestMergeResult:
def _try_reuse_manifest_merge_result(retry_result, current_manifests):
- if (retry_result is None
+ if (not isinstance(retry_result, CommitFailRetryResult)
or retry_result.commit_result_may_be_uncertain
or retry_result.manifest_merge_result is None):
return None
@@ -180,6 +180,17 @@ class RetryResult(CommitResult):
return False
+class CommitFailRetryResult(RetryResult):
+ """Retry after an atomic snapshot commit failed, matching Java."""
+
+
+class RollbackRetryResult(RetryResult):
+ """Retry after a conflicting compaction was rolled back, matching Java."""
+
+ def __init__(self, exception: Optional[Exception] = None):
+ super().__init__(None, exception)
+
+
class RewriteResult(CommitResult):
def __init__(self, rewrite: RowIdRewriteResult):
@@ -489,8 +500,8 @@ class FileStoreCommit:
else 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).
+ # No entries to commit (e.g. drop_partitions with no matching
+ # data): skip an empty snapshot.
if not commit_entries and not index_deletes and not index_adds:
break
@@ -587,7 +598,8 @@ class FileStoreCommit:
hash_index_base_snapshot=None,
commit_result_may_be_uncertain: bool = False) ->
CommitResult:
start_millis = int(time.time() * 1000)
- if self._is_duplicate_commit(retry_result, latest_snapshot,
commit_identifier, commit_kind):
+ if self._is_duplicate_commit(
+ retry_result, latest_snapshot, commit_identifier, commit_kind):
return SuccessResult()
latest_snapshot_id = latest_snapshot.id if latest_snapshot else 0
@@ -618,17 +630,22 @@ class FileStoreCommit:
base_data_files = None
if detect_conflicts:
incremental = None
+ commit_fail_retry = (
+ retry_result
+ if isinstance(retry_result, CommitFailRetryResult)
+ else None
+ )
if (latest_snapshot is not None
- and retry_result is not None
- and retry_result.latest_snapshot is not None
- and retry_result.base_data_files is not None):
+ and commit_fail_retry is not None
+ and commit_fail_retry.latest_snapshot is not None
+ and commit_fail_retry.base_data_files is not None):
incremental = self.commit_scanner.read_incremental_changes(
- retry_result.latest_snapshot,
+ commit_fail_retry.latest_snapshot,
latest_snapshot,
commit_entries,
index_entries)
if incremental is not None:
- base_data_files = list(retry_result.base_data_files)
+ base_data_files = list(commit_fail_retry.base_data_files)
if incremental:
base_data_files.extend(incremental)
base_data_files = FileEntry.merge_entries(base_data_files)
@@ -665,7 +682,7 @@ class FileStoreCommit:
if self.rollback.try_to_rollback(latest_snapshot):
# Rolled back: base/snapshot no longer valid; next
attempt
# re-scans from scratch (matches Java
RollbackRetryResult).
- return RetryResult(None, conflict_exception)
+ return RollbackRetryResult(conflict_exception)
raise conflict_exception
# Apply row tracking logic after conflict detection (matches Java
ordering)
@@ -818,7 +835,7 @@ class FileStoreCommit:
merge_after_manifests,
)
)
- return RetryResult(
+ return CommitFailRetryResult(
latest_snapshot,
None,
base_data_files=base_data_files,
@@ -827,7 +844,7 @@ class FileStoreCommit:
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(
+ return CommitFailRetryResult(
latest_snapshot,
e,
base_data_files=base_data_files,
@@ -905,15 +922,31 @@ class FileStoreCommit:
return self.manifest_file_manager.rolling_write(
commit_entries, self.manifest_target_size, base_name)
- def _is_duplicate_commit(self, retry_result, latest_snapshot,
commit_identifier, commit_kind) -> bool:
- if retry_result is not None and latest_snapshot is not None:
+ def _is_duplicate_commit(
+ self,
+ retry_result,
+ latest_snapshot,
+ commit_identifier,
+ commit_kind) -> bool:
+ if (isinstance(retry_result, CommitFailRetryResult)
+ and latest_snapshot is not None):
start_check_snapshot_id = 1 # Snapshot.FIRST_SNAPSHOT_ID
if retry_result.latest_snapshot is not None:
start_check_snapshot_id = retry_result.latest_snapshot.id + 1
for snapshot_id in range(start_check_snapshot_id,
latest_snapshot.id + 1):
snapshot =
self.snapshot_manager.get_snapshot_by_id(snapshot_id)
- if (snapshot and snapshot.commit_user == self.commit_user and
+ if snapshot is None:
+ raise RuntimeError(
+ "Cannot determine whether commit {} by user {} "
+ "succeeded because snapshot {} cannot be found."
+ .format(
+ commit_identifier,
+ self.commit_user,
+ snapshot_id,
+ )
+ )
+ if (snapshot.commit_user == self.commit_user and
snapshot.commit_identifier == commit_identifier and
snapshot.commit_kind == commit_kind):
logger.info(
