This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 66e400f3213ae78c808108ea9fd7f53c3320074d Author: YeJunHao <[email protected]> AuthorDate: Thu Sep 17 16:44:40 2026 +0800 [python] Preserve manifest sidecar references on commit retries (#9921) --- .../pypaimon/tests/file_store_commit_test.py | 52 ++++++++++++++++++++++ paimon-python/pypaimon/write/file_store_commit.py | 6 +++ 2 files changed, 58 insertions(+) diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index 4a2f9d2450..1daefe1fde 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -17,6 +17,7 @@ import unittest import uuid +from dataclasses import replace from datetime import datetime from unittest.mock import MagicMock, Mock, patch @@ -311,6 +312,8 @@ class TestFileStoreCommit(unittest.TestCase): self._manifest_meta('suffix'), ] replacement = [self._manifest_meta('merged')] + previous[0].extra_files = ['a.avro.sidecar'] + current[1].extra_files = ['a.avro.sidecar'] result = _try_replace_manifest_files( current, previous, replacement) @@ -321,6 +324,31 @@ class TestFileStoreCommit(unittest.TestCase): ) self.assertIsNot(current[1], previous[0]) + def test_replace_manifest_files_rejects_changed_metadata( + self, mock_manifest_list_manager, mock_manifest_file_manager): + previous = self._manifest_meta('before') + replacement = [self._manifest_meta('merged')] + changes = { + 'min_bucket': 0, + 'max_bucket': 3, + 'min_level': 0, + 'max_level': 2, + 'total_buckets': 4, + 'extra_files': ['before.avro.sidecar'], + } + for field, value in changes.items(): + with self.subTest(field=field): + current = replace(previous, **{field: value}) + self.assertIsNone(_try_replace_manifest_files( + [current], [previous], replacement)) + + for old, new in [(None, []), ([], ['sidecar']), (['old'], ['new']), + (['a', 'b'], ['b', 'a'])]: + with self.subTest(old=old, new=new): + self.assertIsNone(_try_replace_manifest_files( + [replace(previous, extra_files=new)], + [replace(previous, extra_files=old)], replacement)) + def test_replace_manifest_files_preserves_order_and_empty_semantics( self, mock_manifest_list_manager, mock_manifest_file_manager): a = self._manifest_meta('a') @@ -605,6 +633,30 @@ class TestFileStoreCommit(unittest.TestCase): [manifest.file_name for manifest in base_manifests], ) + def test_retry_preserves_concurrently_added_manifest_sidecar( + self, mock_manifest_list_manager, mock_manifest_file_manager): + previous = self._manifest_meta('before') + _, retry_result = self._run_manifest_commit_attempt( + commit_result=False, + existing_manifests=[previous], + merged_manifests=[previous], + ) + # A concurrent commit attaches a sidecar to the same manifest file. + current = replace(previous, extra_files=['before.avro.sidecar']) + file_store_commit, result = self._run_manifest_commit_attempt( + commit_result=True, + retry_result=retry_result, + existing_manifests=[current], + ) + + self.assertTrue(result.is_success()) + file_store_commit.manifest_file_merger.merge.assert_not_called() + base_manifests = ( + file_store_commit.manifest_list_manager.write + .call_args_list[-1][0][1]) + self.assertEqual([current], base_manifests) + self.assertEqual(['before.avro.sidecar'], base_manifests[0].extra_files) + def test_retry_skips_manifest_merge_when_previous_input_is_not_contiguous( self, mock_manifest_list_manager, mock_manifest_file_manager): previous_before = [ diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 8c7cab1175..27907f8741 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -120,8 +120,14 @@ def _manifest_file_key(manifest: ManifestFileMeta): GenericRowSerializer.to_bytes(stats.max_values), tuple(stats.null_counts) if stats.null_counts is not None else None, manifest.schema_id, + manifest.min_bucket, + manifest.max_bucket, + manifest.min_level, + manifest.max_level, manifest.min_row_id, manifest.max_row_id, + manifest.total_buckets, + tuple(manifest.extra_files) if manifest.extra_files is not None else None, )
