This is an automated email from the ASF dual-hosted git repository.
XiaoHongbo-Hope 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 a70912b5fc [python] Avoid deleting files during distributed commit
failures (#9269)
a70912b5fc is described below
commit a70912b5fc5b1506e0a927aca5a19f42b2627e01
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Aug 17 20:45:13 2026 +0800
[python] Avoid deleting files during distributed commit failures (#9269)
## What changed
- Remove `CommitConflictError`; commit conflicts now propagate as
ordinary `RuntimeError` failures.
- Never automatically abort prepared files when a table commit raises.
- Stop Ray's driver-side `on_write_failed()` callback from deleting
files returned by completed write tasks.
- Remove the pending-commit-message cleanup state from the Ray sink.
- Keep worker-local cleanup before a `CommitMessage` is returned,
explicit `TableCommit.abort()`, and pre-commit local cleanup paths.
## Why
Ray task, job, or scheduler retries can replay a previously returned
`CommitMessage`. If another failure path automatically aborts that
message first, the later attempt can commit a snapshot whose manifest
references an already deleted file. Commit retries do not verify that
every referenced data file still exists.
Post-commit callbacks create the same ambiguity: an exception can
propagate after the snapshot is already committed, so exception type is
not a safe ownership signal for deleting files.
This change deliberately prefers possible orphan files over deleting
data that another attempt may commit or a snapshot may already
reference. Orphan cleanup can reclaim those files later using table
metadata.
## Validation
- `124 passed, 15 subtests passed` across Ray sink, commit callback,
dynamic bucket, table write, and file-store commit tests.
- `py_compile` passed for all changed production files.
- `git diff --check` passed.
---
paimon-python/pypaimon/daft/daft_datasink.py | 12 +---
.../pypaimon/ray/data_evolution_merge_into.py | 40 ++---------
paimon-python/pypaimon/ray/update_by_row_id.py | 42 ++---------
.../tests/daft/daft_pk_distributed_write_test.py | 8 +++
paimon-python/pypaimon/tests/ray_sink_test.py | 83 ++++++++++------------
.../pypaimon/tests/ray_update_by_row_id_test.py | 6 +-
.../pypaimon/tests/table_merge_into_test.py | 21 +++---
.../pypaimon/tests/write/commit_callback_test.py | 47 ++++++++++++
.../pypaimon/tests/write/dynamic_bucket_test.py | 11 ++-
.../pypaimon/tests/write/table_write_test.py | 16 ++---
.../pypaimon/write/commit/conflict_detection.py | 4 --
paimon-python/pypaimon/write/file_store_commit.py | 19 +----
paimon-python/pypaimon/write/ray_datasink.py | 35 +--------
paimon-python/pypaimon/write/table_commit.py | 65 +++++++----------
14 files changed, 159 insertions(+), 250 deletions(-)
diff --git a/paimon-python/pypaimon/daft/daft_datasink.py
b/paimon-python/pypaimon/daft/daft_datasink.py
index c21767b0d2..1486832a26 100644
--- a/paimon-python/pypaimon/daft/daft_datasink.py
+++ b/paimon-python/pypaimon/daft/daft_datasink.py
@@ -289,23 +289,15 @@ class PaimonDataSink(DataSink[list[Any]]):
def finalize(self, write_results: list[WriteResult[list[Any]]]) ->
MicroPartition:
all_commit_messages = [msg for wr in write_results for msg in
wr.result]
- table_commit = self._write_builder.new_commit()
non_empty_workers = sum(
any(not message.is_empty() for message in result.result)
for result in write_results
)
primary_key_error = self._primary_key_write_error(non_empty_workers)
if primary_key_error is not None:
- try:
- table_commit.abort(all_commit_messages)
- except Exception:
- logger.warning(
- "Failed to abort uncommitted direct Daft PK files.",
- exc_info=True,
- )
- finally:
- table_commit.close()
raise ValueError(primary_key_error)
+
+ table_commit = self._write_builder.new_commit()
try:
table_commit.commit(all_commit_messages)
finally:
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
index d43ac21f36..b7ba555822 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
@@ -375,8 +375,7 @@ 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
+ commit_messages: list = []
update_msgs: list = []
num_updated = 0
@@ -398,7 +397,7 @@ def _execute_and_commit(
),
collect_row_ids=collect_action_row_ids,
)
- pending_msgs.extend(update_msgs)
+ commit_messages.extend(update_msgs)
if delete_ds is not None:
delete_msgs, num_deleted, delete_row_ids =
distributed_delete_apply(
@@ -411,7 +410,7 @@ def _execute_and_commit(
),
collect_row_ids=collect_action_row_ids,
)
- pending_msgs.extend(delete_msgs)
+ commit_messages.extend(delete_msgs)
if collect_action_row_ids:
_validate_disjoint_action_row_ids(update_row_ids, delete_row_ids)
@@ -421,7 +420,7 @@ def _execute_and_commit(
insert_ds, table,
ray_remote_args=ray_remote_args, concurrency=concurrency,
)
- pending_msgs.extend(insert_msgs)
+ commit_messages.extend(insert_msgs)
num_inserted = sum(
f.row_count
for m in insert_msgs
@@ -429,12 +428,11 @@ def _execute_and_commit(
if not DataFileMeta.is_blob_file(f.file_name)
)
- all_msgs: list = list(pending_msgs)
+ all_msgs: list = list(commit_messages)
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:
@@ -447,8 +445,6 @@ def _execute_and_commit(
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
@@ -459,32 +455,6 @@ 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/ray/update_by_row_id.py
b/paimon-python/pypaimon/ray/update_by_row_id.py
index 41b968c4fd..bc1b437e8c 100644
--- a/paimon-python/pypaimon/ray/update_by_row_id.py
+++ b/paimon-python/pypaimon/ray/update_by_row_id.py
@@ -156,52 +156,20 @@ def update_by_row_id(
def _commit_update_messages(table, commit_messages) -> None:
- pending_msgs: list = list(commit_messages)
- commit_started = False
-
- try:
- table_commit = None
- try:
- table_commit = table.new_batch_write_builder().new_commit()
- commit_started = True
- table_commit.commit(pending_msgs)
- finally:
- if table_commit is not None:
- try:
- table_commit.close()
- except Exception as close_error:
- logger.warning(
- "Failed to close update_by_row_id commit: %s",
- close_error,
- exc_info=close_error,
- )
- except Exception as e:
- if not commit_started:
- _abort_pending_update_messages(table, pending_msgs)
- _reraise_inner(e)
-
-
-def _abort_pending_update_messages(table, commit_messages) -> None:
- if not commit_messages:
- return
-
+ messages = list(commit_messages)
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 update_by_row_id commit messages: %s",
- abort_error,
- exc_info=abort_error,
- )
+ table_commit.commit(messages)
+ except Exception as e:
+ _reraise_inner(e)
finally:
if table_commit is not None:
try:
table_commit.close()
except Exception as close_error:
logger.warning(
- "Failed to close update_by_row_id abort commit: %s",
+ "Failed to close update_by_row_id commit: %s",
close_error,
exc_info=close_error,
)
diff --git
a/paimon-python/pypaimon/tests/daft/daft_pk_distributed_write_test.py
b/paimon-python/pypaimon/tests/daft/daft_pk_distributed_write_test.py
index 1c2ff559b9..c96ad3c20c 100644
--- a/paimon-python/pypaimon/tests/daft/daft_pk_distributed_write_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_pk_distributed_write_test.py
@@ -125,6 +125,13 @@ def
test_bare_datasink_rejects_multiple_primary_key_write_tasks(catalog):
pa.table({"id": [value], "value": [f"v-{value}"]})
)
results.extend(sink.write(iter([micropartition])))
+ staged_paths = [
+ str(data_file.external_path or data_file.file_path)
+ for result in results
+ for message in result.result
+ for data_file in message.new_files
+ ]
+ assert staged_paths
with pytest.raises(
ValueError, match="require a single non-empty Daft write task"
@@ -132,6 +139,7 @@ def
test_bare_datasink_rejects_multiple_primary_key_write_tasks(catalog):
sink.finalize(results)
assert table.snapshot_manager().get_latest_snapshot() is None
+ assert all(table.file_io.exists(path) for path in staged_paths)
def test_bare_datasink_supports_single_task_dynamic_bucket(catalog):
diff --git a/paimon-python/pypaimon/tests/ray_sink_test.py
b/paimon-python/pypaimon/tests/ray_sink_test.py
index e6a4029b81..41e7365e5f 100644
--- a/paimon-python/pypaimon/tests/ray_sink_test.py
+++ b/paimon-python/pypaimon/tests/ray_sink_test.py
@@ -496,7 +496,6 @@ class RaySinkTest(unittest.TestCase):
datasink._writer_builder.new_commit = mock_new_commit
with self.assertRaises(Exception):
datasink.on_write_complete(write_result)
- self.assertEqual(len(datasink._pending_commit_messages), 1)
def test_on_write_complete_without_on_write_start(self):
from ray.data.datasource.datasink import WriteResult
@@ -580,34 +579,15 @@ class RaySinkTest(unittest.TestCase):
)
def test_on_write_failed(self):
- # Test without pending messages (on_write_complete() never called)
datasink = PaimonDatasink(self.table, overwrite=False)
datasink.on_write_start()
- self.assertEqual(datasink._pending_commit_messages, [])
- error = Exception("Write job failed")
- datasink.on_write_failed(error) # Should not raise exception
-
- # Test with pending messages (on_write_complete() was called but
failed)
- datasink = PaimonDatasink(self.table, overwrite=False)
- datasink.on_write_start()
- commit_msg1 = Mock(spec=CommitMessage)
- commit_msg2 = Mock(spec=CommitMessage)
- datasink._pending_commit_messages = [commit_msg1, commit_msg2]
-
- mock_commit = Mock()
- datasink._writer_builder.new_commit = Mock(return_value=mock_commit)
+ datasink._writer_builder.new_commit = Mock()
error = Exception("Write job failed")
datasink.on_write_failed(error)
- mock_commit.abort.assert_called_once()
- abort_args = mock_commit.abort.call_args[0][0]
- self.assertEqual(len(abort_args), 2)
- self.assertEqual(abort_args[0], commit_msg1)
- self.assertEqual(abort_args[1], commit_msg2)
- mock_commit.close.assert_called_once()
- self.assertEqual(datasink._pending_commit_messages, [])
+ datasink._writer_builder.new_commit.assert_not_called()
- def test_consume_write_results_stages_messages_before_late_failure(self):
+ def test_consume_write_results_reports_late_failure(self):
import pickle
message_col = '__messages__'
@@ -629,9 +609,6 @@ class RaySinkTest(unittest.TestCase):
FailingResults(), coordinator, message_col
)
- coordinator.add_pending_commit_messages.assert_called_once_with(
- ['first']
- )
coordinator.on_write_complete.assert_not_called()
coordinator.on_write_failed.assert_called_once()
@@ -660,29 +637,47 @@ class RaySinkTest(unittest.TestCase):
results, coordinator, message_col, error_col
)
- self.assertEqual(
- [call.args[0] for call in
- coordinator.add_pending_commit_messages.call_args_list],
- [['first'], [], ['last']],
- )
coordinator.on_write_complete.assert_not_called()
coordinator.on_write_failed.assert_called_once()
- # Test abort failure handling (should not raise exception)
- datasink = PaimonDatasink(self.table, overwrite=False)
- datasink.on_write_start()
- commit_msg1 = Mock(spec=CommitMessage)
- datasink._pending_commit_messages = [commit_msg1]
+ def test_consume_write_results_failure_preserves_completed_files(self):
+ import pickle
- mock_commit = Mock()
- mock_commit.abort.side_effect = Exception("Abort failed")
- datasink._writer_builder.new_commit = Mock(return_value=mock_commit)
- error = Exception("Write job failed")
- datasink.on_write_failed(error)
+ writer = self.table.new_batch_write_builder().new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['Alice'],
+ 'value': [1.1],
+ }, schema=self.pk_pa_schema))
+ messages = writer.prepare_commit()
+ writer.close()
+ paths = [
+ file.external_path or file.file_path
+ for message in messages
+ for file in message.new_files
+ ]
- mock_commit.abort.assert_called_once()
- mock_commit.close.assert_called_once()
- self.assertEqual(datasink._pending_commit_messages, [])
+ message_col = '__messages__'
+ error_col = '__errors__'
+ results = Mock()
+ results.iter_batches.return_value = iter([
+ pa.table({
+ message_col: pa.array([
+ pickle.dumps(messages),
+ pickle.dumps([]),
+ ], type=pa.binary()),
+ error_col: pa.array([None, 'worker failure'],
type=pa.string()),
+ }),
+ ])
+ coordinator = PaimonDatasink(self.table, overwrite=False)
+ coordinator.on_write_start()
+
+ with self.assertRaisesRegex(RuntimeError, 'worker failure'):
+ _consume_write_results(
+ results, coordinator, message_col, error_col
+ )
+
+ self.assertTrue(all(self.table.file_io.exists(path) for path in paths))
if __name__ == '__main__':
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 801c35218a..e2bda92237 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
@@ -175,7 +175,7 @@ class RayUpdateByRowIdTest(unittest.TestCase):
update_by_row_id(target, src, self.catalog_options,
update_cols=["age"])
self.assertEqual(captured["base_snapshot_id"], expected_sid)
- def test_new_commit_failure_aborts_pending_messages(self):
+ def test_new_commit_failure_preserves_pending_messages(self):
err = RuntimeError("new_commit failed")
recorder = {}
@@ -186,8 +186,8 @@ class RayUpdateByRowIdTest(unittest.TestCase):
)
self.assertEqual(recorder["commit_calls"], 0)
- self.assertEqual(recorder["abort_calls"], 1)
- self.assertEqual(recorder["abort_msgs"], recorder["msgs"])
+ self.assertEqual(recorder["abort_calls"], 0)
+ self.assertEqual(recorder["new_commit_calls"], 1)
def test_commit_failure_does_not_abort_after_commit_started(self):
err = RuntimeError("commit failed")
diff --git a/paimon-python/pypaimon/tests/table_merge_into_test.py
b/paimon-python/pypaimon/tests/table_merge_into_test.py
index 5ad3a405f3..f69ddfe8b5 100644
--- a/paimon-python/pypaimon/tests/table_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/table_merge_into_test.py
@@ -102,7 +102,7 @@ class TableMergeIntoTest(BatchModeMixin,
DataEvolutionTestBase, unittest.TestCas
update_msg = object()
delete_msg = object()
- table, table_commit = self._mock_ray_commit_table()
+ table, _ = self._mock_ray_commit_table()
with patch.object(
ray_merge,
@@ -125,14 +125,13 @@ 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()
+ table.new_batch_write_builder.assert_not_called()
- def
test_ray_execute_aborts_prepared_messages_on_later_branch_failure(self):
+ def test_ray_execute_preserves_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()
+ table, _ = self._mock_ray_commit_table()
with patch.object(
ray_merge,
@@ -155,15 +154,13 @@ class TableMergeIntoTest(BatchModeMixin,
DataEvolutionTestBase, unittest.TestCas
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()
+ table.new_batch_write_builder.assert_not_called()
- def test_ray_execute_aborts_prepared_messages_on_insert_failure(self):
+ def test_ray_execute_preserves_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()
+ table, _ = self._mock_ray_commit_table()
with patch.object(
ray_merge,
@@ -186,9 +183,7 @@ class TableMergeIntoTest(BatchModeMixin,
DataEvolutionTestBase, unittest.TestCas
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()
+ table.new_batch_write_builder.assert_not_called()
def test_ray_execute_does_not_abort_after_commit_starts(self):
import pypaimon.ray.data_evolution_merge_into as ray_merge
diff --git a/paimon-python/pypaimon/tests/write/commit_callback_test.py
b/paimon-python/pypaimon/tests/write/commit_callback_test.py
index 4e02bf32f6..036031edef 100644
--- a/paimon-python/pypaimon/tests/write/commit_callback_test.py
+++ b/paimon-python/pypaimon/tests/write/commit_callback_test.py
@@ -153,6 +153,53 @@ class CommitCallbackTest(unittest.TestCase):
table_commit.close()
self.assertTrue(callback.closed)
+ def test_callback_error_after_commit_keeps_committed_files(self):
+ table = self._create_table('test_callback_error_after_commit')
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+
+ callback_error = RuntimeError('callback failed after commit')
+
+ class FailingCallback(CommitCallback):
+
+ def call(self, context: CommitCallbackContext) -> None:
+ raise callback_error
+
+ table_commit.add_commit_callback(FailingCallback())
+ try:
+ table_write.write_arrow(pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['committed'],
+ 'dt': ['p1'],
+ }, schema=self.pa_schema))
+ messages = table_write.prepare_commit()
+ data_paths = [
+ file.external_path or file.file_path
+ for message in messages
+ for file in message.new_files
+ ]
+
+ with self.assertRaises(RuntimeError) as context:
+ table_commit.commit(messages)
+
+ latest_snapshot = table.snapshot_manager().get_latest_snapshot()
+ self.assertEqual(1, latest_snapshot.id)
+ self.assertTrue(all(
+ table.file_io.exists(path) for path in data_paths))
+
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(
+ {'id': [1], 'name': ['committed'], 'dt': ['p1']},
+ actual.to_pydict(),
+ )
+ self.assertIs(callback_error, context.exception)
+ finally:
+ table_write.close()
+ table_commit.close()
+
def test_callback_not_invoked_when_no_data(self):
table = self._create_table('test_callback_no_data')
write_builder = table.new_batch_write_builder()
diff --git a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
index f7714f5e99..ffe6c88319 100644
--- a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
+++ b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
@@ -37,7 +37,6 @@ from pypaimon.index.index_file_meta import IndexFileMeta
from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.generic_row import GenericRow
-from pypaimon.write.commit.conflict_detection import CommitConflictError
from pypaimon.write.row_key_extractor import DynamicBucketRowKeyExtractor
@@ -455,7 +454,7 @@ class DynamicBucketTest(unittest.TestCase):
commit2.commit(messages2)
self.assertTrue(all(
- not table.file_io.exists(path)
+ table.file_io.exists(path)
for path in stale_data_paths + stale_index_paths
))
@@ -554,7 +553,7 @@ class DynamicBucketTest(unittest.TestCase):
commit2.commit(messages2)
self.assertTrue(all(
- not table.file_io.exists(path) for path in stale_paths
+ table.file_io.exists(path) for path in stale_paths
))
writer1.close()
writer2.close()
@@ -603,7 +602,7 @@ class DynamicBucketTest(unittest.TestCase):
stale_writer.close()
stale_commit.close()
- def test_retry_then_hash_index_conflict_aborts_prepared_files(self):
+ def test_retry_then_hash_index_conflict_preserves_prepared_files(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(root, 'retry_hash_conflict')
writer, commit, messages = self._prepare_indexed_write(table, [1])
@@ -643,13 +642,13 @@ class DynamicBucketTest(unittest.TestCase):
'_commit_retry_wait',
):
with self.assertRaisesRegex(
- CommitConflictError, 'HASH index assignment conflict'
+ RuntimeError, 'HASH index assignment conflict'
):
commit.commit(messages)
self.assertEqual(1, calls)
self.assertTrue(all(
- not table.file_io.exists(path) for path in prepared_paths
+ table.file_io.exists(path) for path in prepared_paths
))
writer.close()
commit.close()
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index a640b8d31f..e02a688cf9 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -1174,8 +1174,6 @@ class TableWriteTest(unittest.TestCase):
self.assertEqual(1, overwrite_plan.num_buckets(('p',)))
def test_postpone_worker_bucket_plan_mismatch_fails_commit(self):
- from pypaimon.write.commit.conflict_detection import
CommitConflictError
-
table = self._create_postpone_table(
'default.test_postpone_worker_plan_mismatch',
pa_schema=self.postpone_pa_schema,
@@ -1202,7 +1200,7 @@ class TableWriteTest(unittest.TestCase):
large_messages = large_write.prepare_commit()
self.assertEqual({1}, {m.total_buckets for m in small_messages})
self.assertEqual({3}, {m.total_buckets for m in large_messages})
- with self.assertRaisesRegex(CommitConflictError, 'Total buckets'):
+ with self.assertRaisesRegex(RuntimeError, 'Total buckets'):
commit.commit(small_messages + large_messages)
finally:
small_write.close()
@@ -1210,8 +1208,6 @@ class TableWriteTest(unittest.TestCase):
commit.close()
def test_postpone_overwrite_bucket_plan_mismatch_fails_commit(self):
- from pypaimon.write.commit.conflict_detection import
CommitConflictError
-
table = self._create_postpone_table(
'default.test_postpone_overwrite_plan_mismatch',
pa_schema=self.postpone_pa_schema,
@@ -1245,10 +1241,10 @@ class TableWriteTest(unittest.TestCase):
for file in message.new_files
]
- with self.assertRaisesRegex(CommitConflictError, 'Total buckets'):
+ with self.assertRaisesRegex(RuntimeError, 'Total buckets'):
commit.commit(messages)
self.assertTrue(all(
- not table.file_io.exists(path) for path in paths))
+ table.file_io.exists(path) for path in paths))
finally:
small_write.close()
large_write.close()
@@ -1468,8 +1464,6 @@ class TableWriteTest(unittest.TestCase):
)
def test_postpone_concurrent_new_partition_bucket_num_conflict(self):
- from pypaimon.write.commit.conflict_detection import
CommitConflictError
-
table_two_buckets = self._create_postpone_table(
'default.test_postpone_concurrent_bucket_num',
partition_keys=['dt'],
@@ -1530,11 +1524,11 @@ class TableWriteTest(unittest.TestCase):
commit_three.file_store_commit.snapshot_commit.commit = (
fail_cas_after_concurrent_commit)
- with self.assertRaisesRegex(CommitConflictError, "Total buckets"):
+ with self.assertRaisesRegex(RuntimeError, "Total buckets"):
commit_three.commit(messages_three)
self.assertTrue(concurrent_commit['done'])
self.assertTrue(all(
- not table_three_buckets.file_io.exists(path)
+ table_three_buckets.file_io.exists(path)
for path in losing_paths
))
finally:
diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py
b/paimon-python/pypaimon/write/commit/conflict_detection.py
index 62606c183a..4e6026f7b6 100644
--- a/paimon-python/pypaimon/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/write/commit/conflict_detection.py
@@ -149,10 +149,6 @@ class _WriteRange:
self.field_ids = field_ids
-class CommitConflictError(RuntimeError):
- """A deterministic pre-snapshot conflict which is safe to abort."""
-
-
class RowIdExistenceConflict(RuntimeError):
"""A staged row-id file no longer matches the current base-file layout."""
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 7e2b69f97a..adcfa79148 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -39,7 +39,6 @@ from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.write.commit.commit_rollback import CommitRollback
from pypaimon.write.commit.commit_scanner import CommitScanner
from pypaimon.write.commit.conflict_detection import (
- CommitConflictError,
ConflictDetection,
RowIdExistenceConflict,
)
@@ -567,8 +566,6 @@ class FileStoreCommit:
)
if commit_result_may_be_uncertain:
raise RuntimeError(error_msg) from
uncertain_commit_exception
- if retry_result is not None and retry_result.exception is None:
- raise CommitConflictError(error_msg)
if retry_result is not None and retry_result.exception:
raise RuntimeError(error_msg) from retry_result.exception
else:
@@ -597,15 +594,12 @@ class FileStoreCommit:
hash_index_base_snapshot is not None
and latest_snapshot_id != hash_index_base_snapshot
):
- conflict = RuntimeError(
+ raise RuntimeError(
"HASH index assignment conflict detected: assigned from "
"snapshot {}, but the latest snapshot is {}.".format(
hash_index_base_snapshot, latest_snapshot_id
)
)
- if not commit_result_may_be_uncertain:
- raise CommitConflictError(str(conflict)) from conflict
- raise conflict
unique_id = uuid.uuid4()
base_manifest_list = f"manifest-list-{unique_id}-0"
@@ -671,13 +665,6 @@ class FileStoreCommit:
# Rolled back: base/snapshot no longer valid; next
attempt
# re-scans from scratch (matches Java
RollbackRetryResult).
return RetryResult(None, conflict_exception)
- if not commit_result_may_be_uncertain:
- raise CommitConflictError(
- str(conflict_exception)
- ) from conflict_exception
- # A previous attempt may have committed despite returning an
- # error. Preserve the generic, uncertain-result semantics so
- # callers do not delete files which a snapshot may reference.
raise conflict_exception
# Apply row tracking logic after conflict detection (matches Java
ordering)
@@ -886,7 +873,7 @@ class FileStoreCommit:
)
)
if non_compaction_conflict is not None:
- raise CommitConflictError(
+ raise RuntimeError(
str(non_compaction_conflict)
) from non_compaction_conflict
@@ -902,7 +889,7 @@ class FileStoreCommit:
commit_entries,
)
except RuntimeError as rewrite_error:
- raise CommitConflictError(
+ raise RuntimeError(
"{} {}".format(conflict_exception, rewrite_error)
) from conflict_exception
diff --git a/paimon-python/pypaimon/write/ray_datasink.py
b/paimon-python/pypaimon/write/ray_datasink.py
index c87c9cbd9e..e121bd2432 100644
--- a/paimon-python/pypaimon/write/ray_datasink.py
+++ b/paimon-python/pypaimon/write/ray_datasink.py
@@ -82,7 +82,6 @@ class PaimonDatasink(_DatasinkBase):
self._postpone_bucket_plan = postpone_bucket_plan
self._table_name = table.identifier.get_full_name()
self._writer_builder: Optional["WriteBuilder"] = None
- self._pending_commit_messages: List["CommitMessage"] = []
def _is_overwrite(self) -> bool:
return self.overwrite or self.static_partition is not None
@@ -192,11 +191,8 @@ class PaimonDatasink(_DatasinkBase):
msg for msg in all_commit_messages if not msg.is_empty()
]
- self._pending_commit_messages = non_empty_messages
-
if not non_empty_messages and not self._is_overwrite():
logger.info("No data to commit (all commit messages are
empty)")
- self._pending_commit_messages = []
return
# Ray does not call on_write_start when the input has no blocks.
@@ -211,16 +207,12 @@ class PaimonDatasink(_DatasinkBase):
table_commit = self._writer_builder.new_commit()
table_commit.commit(non_empty_messages)
- self._pending_commit_messages = []
-
logger.info(f"Successfully committed write job for table
{self._table_name}")
except Exception as e:
logger.error(
f"Error committing write job for table {self._table_name}:
{e}",
exc_info=e
)
- if table_commit is not None:
- self._pending_commit_messages = []
raise
finally:
if table_commit is not None:
@@ -232,35 +224,13 @@ class PaimonDatasink(_DatasinkBase):
exc_info=e
)
- def add_pending_commit_messages(self, commit_messages) -> None:
- self._pending_commit_messages.extend(
- message for message in commit_messages if not message.is_empty()
- )
-
def on_write_failed(self, error: Exception) -> None:
logger.error(
f"Write job failed for table {self._table_name}. Error: {error}",
exc_info=error
)
-
- if self._pending_commit_messages:
- try:
- table_commit = self._writer_builder.new_commit()
- try:
- table_commit.abort(self._pending_commit_messages)
- logger.info(
- f"Aborted {len(self._pending_commit_messages)} commit
messages "
- f"for table {self._table_name} in on_write_failed()"
- )
- finally:
- table_commit.close()
- except Exception as abort_error:
- logger.error(
- f"Error aborting commit messages in on_write_failed():
{abort_error}",
- exc_info=abort_error
- )
- finally:
- self._pending_commit_messages = []
+ # Do not abort files returned by completed write tasks. Ray or an outer
+ # scheduler may replay those commit messages in another attempt.
def write_paimon_dataset(
@@ -480,7 +450,6 @@ def _consume_write_results(
for blob, error in zip(messages, batch_errors):
commit_messages = pickle.loads(blob)
write_returns.append(commit_messages)
- coordinator.add_pending_commit_messages(commit_messages)
if error is not None:
errors.append(error)
if errors:
diff --git a/paimon-python/pypaimon/write/table_commit.py
b/paimon-python/pypaimon/write/table_commit.py
index a70c52ebd3..3215f7eea7 100644
--- a/paimon-python/pypaimon/write/table_commit.py
+++ b/paimon-python/pypaimon/write/table_commit.py
@@ -23,7 +23,6 @@ from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
logger = logging.getLogger(__name__)
from pypaimon.write.commit_callback import CommitCallback
from pypaimon.write.commit_message import CommitMessage
-from pypaimon.write.commit.conflict_detection import CommitConflictError
from pypaimon.write.file_store_commit import FileStoreCommit
@@ -64,43 +63,33 @@ class TableCommit:
def _commit(self, commit_messages: List[CommitMessage], commit_identifier:
int = BATCH_COMMIT_IDENTIFIER):
non_empty_messages = [msg for msg in commit_messages if not
msg.is_empty()]
- try:
- if self.overwrite_partition is not None:
- # Always call overwrite() even with empty messages, so that
- # FileStoreCommit.overwrite can handle the empty case properly
- # (e.g. static overwrite with empty data should delete the
partition).
- logger.info(
- "Committing overwrite to table %s, %d non-empty messages",
- self.table.identifier, len(non_empty_messages)
- )
- self.file_store_commit.overwrite(
- overwrite_partition=self.overwrite_partition,
- commit_messages=non_empty_messages,
- commit_identifier=commit_identifier
- )
- else:
- if not non_empty_messages:
- return
- logger.info(
- "Committing table %s, %d non-empty messages",
- self.table.identifier, len(non_empty_messages)
- )
- self.file_store_commit.commit(
- commit_messages=non_empty_messages,
- commit_identifier=commit_identifier
- )
- except CommitConflictError:
- # These files are known to be uncommitted. Generic commit failures
- # remain untouched because their success is uncertain.
- try:
- self.file_store_commit.abort(non_empty_messages)
- except Exception:
- logger.warning(
- "Failed to abort files after a deterministic commit "
- "conflict.",
- exc_info=True,
- )
- raise
+ # Never abort files in response to a commit exception. Preserving
+ # possible orphan files is safer than deleting files which another
+ # attempt may still commit or which a snapshot may already reference.
+ if self.overwrite_partition is not None:
+ # Always call overwrite() even with empty messages, so that
+ # FileStoreCommit.overwrite can handle the empty case properly
+ # (e.g. static overwrite with empty data should delete the
partition).
+ logger.info(
+ "Committing overwrite to table %s, %d non-empty messages",
+ self.table.identifier, len(non_empty_messages)
+ )
+ self.file_store_commit.overwrite(
+ overwrite_partition=self.overwrite_partition,
+ commit_messages=non_empty_messages,
+ commit_identifier=commit_identifier
+ )
+ else:
+ if not non_empty_messages:
+ return
+ logger.info(
+ "Committing table %s, %d non-empty messages",
+ self.table.identifier, len(non_empty_messages)
+ )
+ self.file_store_commit.commit(
+ commit_messages=non_empty_messages,
+ commit_identifier=commit_identifier
+ )
def abort(self, commit_messages: List[CommitMessage]):
self.file_store_commit.abort(commit_messages)
