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 f9476867de [python] Notify callbacks for successful retried commits
(#9589)
f9476867de is described below
commit f9476867dee338db8be9d7a583c00c882d2ab999
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Sep 3 21:05:29 2026 +0800
[python] Notify callbacks for successful retried commits (#9589)
---
.../pypaimon/tests/write/commit_callback_test.py | 88 ++++++++++++++++++++++
paimon-python/pypaimon/write/file_store_commit.py | 56 +++++++++++---
2 files changed, 134 insertions(+), 10 deletions(-)
diff --git a/paimon-python/pypaimon/tests/write/commit_callback_test.py
b/paimon-python/pypaimon/tests/write/commit_callback_test.py
index 036031edef..55ba65dd4d 100644
--- a/paimon-python/pypaimon/tests/write/commit_callback_test.py
+++ b/paimon-python/pypaimon/tests/write/commit_callback_test.py
@@ -23,6 +23,7 @@ import unittest
import pyarrow as pa
from pypaimon import CatalogFactory, Schema
+from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.write.commit_callback import CommitCallback,
CommitCallbackContext
@@ -65,6 +66,22 @@ class CommitCallbackTest(unittest.TestCase):
self.catalog.create_table(f'default.{table_name}', schema, False)
return self.catalog.get_table(f'default.{table_name}')
+ def _lose_commit_response_once(self, table_commit):
+ real_commit = table_commit.file_store_commit.snapshot_commit.commit
+ attempts = []
+
+ def commit_then_lose_response(
+ base_snapshot_uuid, snapshot, statistics):
+ attempts.append(snapshot.id)
+ self.assertTrue(real_commit(
+ base_snapshot_uuid, snapshot, statistics))
+ raise TimeoutError('lost snapshot commit response')
+
+ table_commit.file_store_commit.snapshot_commit.commit = (
+ commit_then_lose_response)
+ table_commit.file_store_commit._commit_retry_wait = lambda _: None
+ return attempts
+
def test_callback_invoked_on_commit(self):
table = self._create_table('test_callback_invoked')
write_builder = table.new_batch_write_builder()
@@ -115,6 +132,77 @@ class CommitCallbackTest(unittest.TestCase):
table_write.close()
table_commit.close()
+ def test_callback_invoked_after_lost_commit_response(self):
+ table = self._create_table(
+ 'test_callback_response_loss',
+ options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ },
+ )
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ callback = RecordingCallback()
+ table_commit.add_commit_callback(callback)
+ attempts = self._lose_commit_response_once(table_commit)
+
+ table_write.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['a', 'b'],
+ 'dt': ['p1', 'p1'],
+ }, schema=self.pa_schema))
+ messages = table_write.prepare_commit()
+ expected_paths = sorted(
+ file.file_path
+ for message in messages
+ for file in message.new_files
+ )
+ table_commit.commit(messages)
+
+ self.assertEqual([1], attempts)
+ self.assertEqual(1, len(callback.contexts))
+ self.assertEqual(1, callback.contexts[0].snapshot.id)
+ self.assertGreater(len(callback.contexts[0].commit_entries), 0)
+ for entry in callback.contexts[0].commit_entries:
+ self.assertIsNotNone(entry.file.first_row_id)
+ self.assertEqual(expected_paths, sorted(
+ entry.file.file_path
+ for entry in callback.contexts[0].commit_entries
+ if entry.kind == 0
+ ))
+ table_write.close()
+ table_commit.close()
+
+ def test_empty_overwrite_callback_after_lost_commit_response(self):
+ table = self._create_table('test_empty_overwrite_response_loss')
+ builder = table.new_batch_write_builder()
+ table_write = builder.new_write()
+ initial_commit = builder.new_commit()
+ table_write.write_arrow(pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['a'],
+ 'dt': ['p1'],
+ }, schema=self.pa_schema))
+ initial_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ initial_commit.close()
+
+ table_commit = table.new_batch_write_builder().new_commit()
+ callback = RecordingCallback()
+ table_commit.add_commit_callback(callback)
+ attempts = self._lose_commit_response_once(table_commit)
+ table_commit.file_store_commit.truncate_table(
+ BATCH_COMMIT_IDENTIFIER)
+
+ self.assertEqual([2], attempts)
+ self.assertEqual(1, len(callback.contexts))
+ self.assertEqual(2, callback.contexts[0].snapshot.id)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ self.assertEqual(0, read_builder.new_read().to_arrow(splits).num_rows)
+ table_commit.close()
+
def test_multiple_callbacks(self):
table = self._create_table('test_multi_callbacks')
write_builder = table.new_batch_write_builder()
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 73ecc8e950..fbcda7fffe 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -497,6 +497,13 @@ class FileStoreCommit:
start_time_ms = int(time.time() * 1000)
while True:
latest_snapshot = self.snapshot_manager.get_latest_snapshot()
+ if retry_result is not None and self._is_duplicate_commit(
+ retry_result,
+ latest_snapshot,
+ commit_identifier,
+ commit_kind,
+ notify_callbacks=True):
+ break
commit_entries = (
rewritten_commit_entries
if rewritten_commit_entries is not None
@@ -602,7 +609,11 @@ class FileStoreCommit:
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):
+ retry_result,
+ latest_snapshot,
+ commit_identifier,
+ commit_kind,
+ notify_callbacks=True):
return SuccessResult()
latest_snapshot_id = latest_snapshot.id if latest_snapshot else 0
@@ -865,14 +876,8 @@ class FileStoreCommit:
commit_kind,
)
- if self.commit_callbacks:
- context = CommitCallbackContext(
- snapshot=snapshot_data,
- commit_entries=commit_entries,
- identifier=commit_identifier,
- )
- for callback in self.commit_callbacks:
- callback.call(context)
+ self._notify_commit_callbacks(
+ snapshot_data, commit_entries, commit_identifier)
return SuccessResult()
@@ -930,7 +935,8 @@ class FileStoreCommit:
retry_result,
latest_snapshot,
commit_identifier,
- commit_kind) -> bool:
+ commit_kind,
+ notify_callbacks=False) -> bool:
if (isinstance(retry_result, CommitFailRetryResult)
and latest_snapshot is not None):
start_check_snapshot_id = 1 # Snapshot.FIRST_SNAPSHOT_ID
@@ -956,9 +962,39 @@ class FileStoreCommit:
f"Commit already completed (snapshot {snapshot_id}), "
f"user: {self.commit_user}, identifier:
{commit_identifier}"
)
+ if notify_callbacks and self.commit_callbacks:
+ entries = []
+ for manifest in self.manifest_list_manager.read_delta(
+ snapshot):
+ entries.extend(self.manifest_file_manager.read(
+ manifest.file_name, drop_stats=False))
+ path_factory = self.table.path_factory()
+ for entry in entries:
+ file = entry.file
+ file.file_path = file.external_path or "%s/%s" % (
+ path_factory.bucket_path(
+ tuple(entry.partition.values),
+ entry.bucket,
+ ).rstrip("/"),
+ file.file_name,
+ )
+ self._notify_commit_callbacks(
+ snapshot, entries, commit_identifier)
return True
return False
+ def _notify_commit_callbacks(
+ self, snapshot, commit_entries, commit_identifier):
+ if not self.commit_callbacks:
+ return
+ context = CommitCallbackContext(
+ snapshot=snapshot,
+ commit_entries=commit_entries,
+ identifier=commit_identifier,
+ )
+ for callback in self.commit_callbacks:
+ callback.call(context)
+
def _create_dynamic_partition_filter(self, commit_messages:
List[CommitMessage]):
"""Build a partition filter from the unique partitions present in
commit_messages."""
predicate_builder = PredicateBuilder(self.table.partition_keys_fields)
