From a52df278a759feaad523ee60ccf0461c76a339d0 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 7 Aug 2026 13:49:34 +0800 Subject: [PATCH] [python] Inherit watermark in PyPaimon commits --- .../pypaimon/tests/file_store_commit_test.py | 17 ++++++++++++++++- .../pypaimon/write/file_store_commit.py | 2 ++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index 50f39d3ed3fb..1ce3fb93c4d4 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -140,7 +140,8 @@ def test_manifest_merge_result_copies_and_freezes_lists( def _run_manifest_commit_attempt(self, commit_side_effect=None, commit_result=None, retry_result=None, existing_manifests=None, - merged_manifests=None): + merged_manifests=None, + latest_watermark=None): file_store_commit = self._create_file_store_commit() self.mock_table.identifier = 'default.test_table' self.mock_table.table_schema.id = 7 @@ -179,6 +180,7 @@ def _run_manifest_commit_attempt(self, commit_side_effect=None, uuid='base-snapshot-uuid', total_record_count=10, index_manifest=None, + watermark=latest_watermark, ) commit_entry = Mock(kind=0) commit_entry.file.row_count = 2 @@ -192,6 +194,18 @@ def _run_manifest_commit_attempt(self, commit_side_effect=None, ) return file_store_commit, result + def test_append_commit_inherits_watermark( + self, mock_manifest_list_manager, mock_manifest_file_manager): + file_store_commit, result = self._run_manifest_commit_attempt( + commit_result=True, + latest_watermark=123, + ) + + self.assertTrue(result.is_success()) + committed_snapshot = ( + file_store_commit.snapshot_commit.commit.call_args[0][1]) + self.assertEqual(123, committed_snapshot.watermark) + def test_false_atomic_commit_retains_manifest_merge_result( self, mock_manifest_list_manager, mock_manifest_file_manager): file_store_commit, result = self._run_manifest_commit_attempt( @@ -750,6 +764,7 @@ def test_append_commit_inherits_index_manifest( latest_snapshot.uuid = "base-snapshot-uuid" latest_snapshot.total_record_count = 10 latest_snapshot.index_manifest = "index-manifest-existing" + latest_snapshot.watermark = None commit_entry = Mock() commit_entry.kind = 0 diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index f9012c5eafef..28fbd441a992 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -742,6 +742,8 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str commit_identifier=commit_identifier, commit_kind=commit_kind, time_millis=int(time.time() * 1000), + watermark=( + latest_snapshot.watermark if latest_snapshot else None), next_row_id=next_row_id, index_manifest=index_manifest, )