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 3952b562af [python][ray] Abort uncommitted merge_into messages (#8466)
3952b562af is described below
commit 3952b562af5cb89f189680312704f619ba69858e
Author: QuakeWang <[email protected]>
AuthorDate: Mon Jul 6 16:29:57 2026 +0800
[python][ray] Abort uncommitted merge_into messages (#8466)
Ray `merge_into` prepares update, delete, and insert commit messages
before committing them together. If a later branch or duplicate
`_ROW_ID` validation fails before the final commit starts, files from
earlier prepared messages have already been written but are neither
committed nor aborted, leaving orphan files.
This change tracks prepared messages until the final commit starts and
aborts them on pre-commit failures. Once commit starts, it does not
abort because a commit failure may still have registered files in a
snapshot. The commit object is also closed in a `finally` block after
creation.
---
.../pypaimon/ray/data_evolution_merge_into.py | 104 +++++++++++++------
.../pypaimon/tests/table_merge_into_test.py | 112 ++++++++++++++++++++-
2 files changed, 181 insertions(+), 35 deletions(-)
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
index ec85b784a4..a49b049af6 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
@@ -18,6 +18,7 @@
"""MERGE INTO ... USING ... for Paimon data-evolution tables via Ray
Datasets."""
+import logging
from dataclasses import dataclass
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
@@ -47,6 +48,8 @@ from pypaimon.ray.data_evolution_merge_transform import (
__all__ = ["merge_into", "WhenMatched", "WhenNotMatched"]
+logger = logging.getLogger(__name__)
+
@dataclass(frozen=True)
class _PrepareCtx:
@@ -367,12 +370,19 @@ def _execute_and_commit(
ray_remote_args, concurrency,
):
collect_action_row_ids = update_ds is not None and delete_ds is not None
+ pending_msgs: list = []
+ commit_started = False
update_msgs: list = []
num_updated = 0
update_row_ids = []
- if update_ds is not None:
- try:
+ delete_msgs: list = []
+ num_deleted = 0
+ delete_row_ids = []
+ num_inserted = 0
+
+ try:
+ if update_ds is not None:
update_msgs, num_updated, update_row_ids =
distributed_update_apply(
update_ds, table, update_cols_union,
num_partitions=num_partitions,
@@ -383,14 +393,9 @@ def _execute_and_commit(
),
collect_row_ids=collect_action_row_ids,
)
- except Exception as e:
- _reraise_inner(e)
+ pending_msgs.extend(update_msgs)
- delete_msgs: list = []
- num_deleted = 0
- delete_row_ids = []
- if delete_ds is not None:
- try:
+ if delete_ds is not None:
delete_msgs, num_deleted, delete_row_ids =
distributed_delete_apply(
delete_ds, table,
num_partitions=num_partitions,
@@ -401,34 +406,45 @@ def _execute_and_commit(
),
collect_row_ids=collect_action_row_ids,
)
- except Exception as e:
- _reraise_inner(e)
+ pending_msgs.extend(delete_msgs)
- if collect_action_row_ids:
- _validate_disjoint_action_row_ids(update_row_ids, delete_row_ids)
+ if collect_action_row_ids:
+ _validate_disjoint_action_row_ids(update_row_ids, delete_row_ids)
- all_msgs: list = list(update_msgs) + list(delete_msgs)
- num_inserted = 0
- if insert_ds is not None:
- try:
+ if insert_ds is not None:
insert_msgs = distributed_write_collect_msgs(
insert_ds, table,
ray_remote_args=ray_remote_args, concurrency=concurrency,
)
- except Exception as e:
- _reraise_inner(e)
- num_inserted = sum(
- f.row_count
- for m in insert_msgs
- for f in m.new_files
- if not DataFileMeta.is_blob_file(f.file_name)
- )
- all_msgs.extend(insert_msgs)
- if all_msgs:
- wb = table.new_batch_write_builder()
- tc = wb.new_commit()
- tc.commit(all_msgs)
- tc.close()
+ pending_msgs.extend(insert_msgs)
+ num_inserted = sum(
+ f.row_count
+ for m in insert_msgs
+ for f in m.new_files
+ if not DataFileMeta.is_blob_file(f.file_name)
+ )
+
+ all_msgs: list = list(pending_msgs)
+ if all_msgs:
+ table_commit = None
+ try:
+ table_commit = table.new_batch_write_builder().new_commit()
+ commit_started = True
+ 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:
+ if not commit_started:
+ _abort_pending_merge_messages(table, pending_msgs)
+ _reraise_inner(e)
# num_matched = rows that passed a matched condition and changed
return {
@@ -438,6 +454,32 @@ def _execute_and_commit(
}
+def _abort_pending_merge_messages(table, commit_messages) -> None:
+ if not commit_messages:
+ return
+
+ table_commit = None
+ try:
+ table_commit = table.new_batch_write_builder().new_commit()
+ table_commit.abort(commit_messages)
+ except Exception as abort_error:
+ logger.warning(
+ "Failed to abort pending merge_into commit messages: %s",
+ abort_error,
+ exc_info=abort_error,
+ )
+ finally:
+ if table_commit is not None:
+ try:
+ table_commit.close()
+ except Exception as close_error:
+ logger.warning(
+ "Failed to close merge_into abort commit: %s",
+ close_error,
+ exc_info=close_error,
+ )
+
+
def _normalize_on(on: OnSpec) -> Tuple[List[str], List[str]]:
if isinstance(on, Mapping):
target_cols = list(on.keys())
diff --git a/paimon-python/pypaimon/tests/table_merge_into_test.py
b/paimon-python/pypaimon/tests/table_merge_into_test.py
index 660443862a..5ad3a405f3 100644
--- a/paimon-python/pypaimon/tests/table_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/table_merge_into_test.py
@@ -17,7 +17,7 @@
################################################################################
import unittest
-from unittest.mock import patch
+from unittest.mock import Mock, patch
import pyarrow as pa
@@ -89,21 +89,33 @@ class TableMergeIntoTest(BatchModeMixin,
DataEvolutionTestBase, unittest.TestCas
with self.assertRaisesRegex(ValueError, "multiple source rows"):
ray_merge._validate_disjoint_action_row_ids([7], [7])
+ def _mock_ray_commit_table(self):
+ table = Mock()
+ write_builder = Mock()
+ table_commit = Mock()
+ table.new_batch_write_builder.return_value = write_builder
+ write_builder.new_commit.return_value = table_commit
+ return table, table_commit
+
def test_ray_execute_validates_mixed_update_delete_duplicate_row_ids(self):
import pypaimon.ray.data_evolution_merge_into as ray_merge
+ update_msg = object()
+ delete_msg = object()
+ table, table_commit = self._mock_ray_commit_table()
+
with patch.object(
ray_merge,
"distributed_update_apply",
- return_value=([], 1, [7])
+ return_value=([update_msg], 1, [7])
), patch.object(
ray_merge,
"distributed_delete_apply",
- return_value=([], 1, [7])
+ return_value=([delete_msg], 1, [7])
):
with self.assertRaisesRegex(ValueError, "multiple source rows"):
ray_merge._execute_and_commit(
- table=object(),
+ table=table,
update_ds=object(),
delete_ds=object(),
insert_ds=None,
@@ -113,6 +125,98 @@ class TableMergeIntoTest(BatchModeMixin,
DataEvolutionTestBase, unittest.TestCas
ray_remote_args=None,
concurrency=None,
)
+ table_commit.abort.assert_called_once_with([update_msg, delete_msg])
+ table_commit.close.assert_called_once_with()
+
+ def
test_ray_execute_aborts_prepared_messages_on_later_branch_failure(self):
+ import pypaimon.ray.data_evolution_merge_into as ray_merge
+
+ update_msg = object()
+ table, table_commit = self._mock_ray_commit_table()
+
+ with patch.object(
+ ray_merge,
+ "distributed_update_apply",
+ return_value=([update_msg], 1, [])
+ ), patch.object(
+ ray_merge,
+ "distributed_delete_apply",
+ side_effect=RuntimeError("delete failed")
+ ):
+ with self.assertRaisesRegex(RuntimeError, "delete failed"):
+ ray_merge._execute_and_commit(
+ table=table,
+ update_ds=object(),
+ delete_ds=object(),
+ insert_ds=None,
+ update_cols_union=["age"],
+ base_snapshot=None,
+ num_partitions=1,
+ ray_remote_args=None,
+ concurrency=None,
+ )
+ table_commit.abort.assert_called_once_with([update_msg])
+ table_commit.close.assert_called_once_with()
+ table_commit.commit.assert_not_called()
+
+ def test_ray_execute_aborts_prepared_messages_on_insert_failure(self):
+ import pypaimon.ray.data_evolution_merge_into as ray_merge
+
+ update_msg = object()
+ table, table_commit = self._mock_ray_commit_table()
+
+ with patch.object(
+ ray_merge,
+ "distributed_update_apply",
+ return_value=([update_msg], 1, [])
+ ), patch.object(
+ ray_merge,
+ "distributed_write_collect_msgs",
+ side_effect=RuntimeError("insert failed")
+ ):
+ with self.assertRaisesRegex(RuntimeError, "insert failed"):
+ ray_merge._execute_and_commit(
+ table=table,
+ update_ds=object(),
+ delete_ds=None,
+ insert_ds=object(),
+ update_cols_union=["age"],
+ base_snapshot=None,
+ num_partitions=1,
+ ray_remote_args=None,
+ concurrency=None,
+ )
+ table_commit.abort.assert_called_once_with([update_msg])
+ table_commit.close.assert_called_once_with()
+ table_commit.commit.assert_not_called()
+
+ def test_ray_execute_does_not_abort_after_commit_starts(self):
+ import pypaimon.ray.data_evolution_merge_into as ray_merge
+
+ update_msg = object()
+ table, table_commit = self._mock_ray_commit_table()
+ table_commit.commit.side_effect = RuntimeError("commit failed")
+
+ with patch.object(
+ ray_merge,
+ "distributed_update_apply",
+ return_value=([update_msg], 1, [])
+ ):
+ with self.assertRaisesRegex(RuntimeError, "commit failed"):
+ ray_merge._execute_and_commit(
+ table=table,
+ update_ds=object(),
+ delete_ds=None,
+ insert_ds=None,
+ update_cols_union=["age"],
+ base_snapshot=None,
+ num_partitions=1,
+ ray_remote_args=None,
+ concurrency=None,
+ )
+ table_commit.commit.assert_called_once_with([update_msg])
+ table_commit.abort.assert_not_called()
+ table_commit.close.assert_called_once_with()
def test_table_merge_into_updates_and_inserts(self):
target = self._create_table()
