This is an automated email from the ASF dual-hosted git repository.
leaves12138 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 46214120d0 [python] Preserve manifest sidecar references on commit
retries (#9921)
46214120d0 is described below
commit 46214120d0ae613b836b6f10f48728d1b601fb4d
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,
)
