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 633bb37a84 [python] Remove manifest merging from commits (#10082)
633bb37a84 is described below
commit 633bb37a849f2df99d6d93306aaa3fb61de073ec
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 15:19:54 2026 +0800
[python] Remove manifest merging from commits (#10082)
---
docs/docs/pypaimon/writing.md | 17 +-
.../pypaimon/common/options/core_options.py | 39 ---
.../pypaimon/manifest/manifest_file_merger.py | 100 -------
.../pypaimon/tests/file_store_commit_test.py | 330 ++++-----------------
.../manifest/manifest_entry_identifier_test.py | 47 ---
.../pypaimon/tests/test_early_row_range_filter.py | 27 +-
.../pypaimon/tests/write/table_write_test.py | 36 +--
paimon-python/pypaimon/write/file_store_commit.py | 154 +---------
8 files changed, 110 insertions(+), 640 deletions(-)
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index e1d79ce544..68d2c8aa0f 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -95,16 +95,13 @@ compatibility when upgrading; PyArrow reads do not
currently use these indexes.
### Manifest Merging
-`manifest.merge.skip-on-write-only` defaults to `false` in both Python and
Java,
-so commits keep their automatic manifest merging behavior. Set both this option
-and `write-only` to `true` to retain existing manifest files during commit and
-avoid the cost of reading and rewriting them. This option has no effect when
-`write-only=false`, which is also the default.
-
-Python supports minor manifest compaction, using `manifest.merge-min-count` and
-`manifest.target-file-size`. Python does not support manifest sort rewrite.
-In Java, skipping automatic manifest merging also skips automatic manifest sort
-rewrite; explicit manifest compaction remains available.
+Python commits retain existing manifest files without merging or rewriting
them,
+including during commit retries. Run manifest compaction through a centralized
+maintenance service, for example using the Java engines' `compact_manifest`
procedure.
+
+The table options `manifest.merge-min-count`,
`manifest.merge.skip-on-write-only`,
+and `write-only` do not enable manifest merging in Python.
+`manifest.target-file-size` still controls the size of newly written manifest
files.
### Commit Callback
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 674cc1dcf6..54bd771310 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -285,16 +285,6 @@ class CoreOptions:
)
)
- WRITE_ONLY: ConfigOption[bool] = (
- ConfigOptions.key("write-only")
- .boolean_type()
- .default_value(False)
- .with_description(
- "Whether to use write-only mode. Automatic manifest merging is
skipped "
- "when both this option and manifest.merge.skip-on-write-only are
true."
- )
- )
-
SCAN_MANIFEST_PARALLELISM: ConfigOption[int] = (
ConfigOptions.key("scan.manifest.parallelism")
.int_type()
@@ -330,26 +320,6 @@ class CoreOptions:
.with_description("Suggested file size of a manifest file.")
)
- MANIFEST_MERGE_SKIP_ON_WRITE_ONLY: ConfigOption[bool] = (
- ConfigOptions.key("manifest.merge.skip-on-write-only")
- .boolean_type()
- .default_value(False)
- .with_description(
- "Whether to skip automatic manifest merging during commit when
write-only is true. "
- "Python only supports minor manifest compaction, without manifest
sort rewrite."
- )
- )
-
- MANIFEST_MERGE_MIN_COUNT: ConfigOption[int] = (
- ConfigOptions.key("manifest.merge-min-count")
- .int_type()
- .default_value(30)
- .with_description(
- "To avoid frequent manifest merges, this parameter specifies the
minimum number "
- "of ManifestFileMeta to merge."
- )
- )
-
# File format options
PARQUET_WRITE_PAGE_INDEX_ENABLED: ConfigOption[bool] = (
ConfigOptions.key("parquet.write-page-index.enabled")
@@ -1329,9 +1299,6 @@ class CoreOptions:
CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET, default
).get_bytes()
- def write_only(self, default=None):
- return self.options.get(CoreOptions.WRITE_ONLY, default)
-
def scan_manifest_parallelism(self, default=None):
return self.options.get(CoreOptions.SCAN_MANIFEST_PARALLELISM, default)
@@ -1350,12 +1317,6 @@ class CoreOptions:
def manifest_sort_enabled(self):
return self.options.get(CoreOptions.MANIFEST_SORT_ENABLED)
- def manifest_merge_skip_on_write_only(self, default=None):
- return self.options.get(CoreOptions.MANIFEST_MERGE_SKIP_ON_WRITE_ONLY,
default)
-
- def manifest_merge_min_count(self, default=None):
- return self.options.get(CoreOptions.MANIFEST_MERGE_MIN_COUNT, default)
-
def file_format(self, default=None):
return self.options.get(CoreOptions.FILE_FORMAT, default)
diff --git a/paimon-python/pypaimon/manifest/manifest_file_merger.py
b/paimon-python/pypaimon/manifest/manifest_file_merger.py
deleted file mode 100644
index 821b12aef5..0000000000
--- a/paimon-python/pypaimon/manifest/manifest_file_merger.py
+++ /dev/null
@@ -1,100 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-import uuid
-from typing import List, Tuple
-
-from pypaimon.manifest.schema.file_entry import FileEntry
-from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
-
-
-class ManifestFileMerger:
- """Minor manifest compaction for Python commits.
-
- This intentionally implements only the minor compaction path from Java
- ManifestFileMerger. It does not do full compaction or manifest sort
rewrite.
- """
-
- def __init__(self, manifest_file_manager, suggested_meta_size: int,
- suggested_min_meta_count: int):
- self.manifest_file_manager = manifest_file_manager
- self.suggested_meta_size = suggested_meta_size
- self.suggested_min_meta_count = suggested_min_meta_count
-
- def merge(self, manifest_files: List[ManifestFileMeta]) ->
Tuple[List[ManifestFileMeta],
-
List[ManifestFileMeta]]:
- new_files = []
- try:
- return self._try_minor_compaction(manifest_files, new_files),
new_files
- except Exception:
- self._delete_manifests(new_files)
- raise
-
- def _try_minor_compaction(self, manifest_files: List[ManifestFileMeta],
- new_files: List[ManifestFileMeta]) ->
List[ManifestFileMeta]:
- result = []
- candidates = []
- total_size = 0
-
- for manifest in manifest_files:
- total_size += manifest.file_size
- candidates.append(manifest)
- if total_size >= self.suggested_meta_size:
- self._merge_candidates(candidates, result, new_files)
- candidates = []
- total_size = 0
-
- if len(candidates) >= self.suggested_min_meta_count:
- self._merge_candidates(candidates, result, new_files)
- else:
- result.extend(candidates)
-
- return result
-
- def _merge_candidates(self, candidates: List[ManifestFileMeta],
- result: List[ManifestFileMeta],
- new_files: List[ManifestFileMeta]):
- if len(candidates) == 1:
- result.append(candidates[0])
- return
-
- entries = []
- for manifest in candidates:
- entries.extend(
- self.manifest_file_manager.read(
- manifest.file_name,
- drop_stats=False,
- )
- )
-
- merged_entries = FileEntry.merge_entries(entries)
- if not merged_entries:
- return
-
- manifest_file = "manifest-{}".format(str(uuid.uuid4()))
- merged_metas = self.manifest_file_manager.rolling_write(
- merged_entries, self.suggested_meta_size, manifest_file)
- result.extend(merged_metas)
- new_files.extend(merged_metas)
-
- def _delete_manifests(self, manifests: List[ManifestFileMeta]):
- for manifest in manifests:
- manifest_path = "{}/{}".format(
- self.manifest_file_manager.manifest_path,
- manifest.file_name,
- )
- self.manifest_file_manager.file_io.delete_quietly(manifest_path)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 1daefe1fde..649812d7fd 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -35,11 +35,9 @@ from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.file_store_commit import (
CommitFailRetryResult,
FileStoreCommit,
- ManifestMergeResult,
RollbackRetryResult,
RewriteResult,
_abort_commit_messages,
- _try_replace_manifest_files,
)
@@ -69,9 +67,6 @@ class TestFileStoreCommitRowTracking(unittest.TestCase):
self.mock_table.table_path = '/test/table/path'
self.mock_table.file_io = Mock()
self.mock_table.options.manifest_target_size.return_value = 8 * 1024 *
1024
- self.mock_table.options.manifest_merge_min_count.return_value = 30
- self.mock_table.options.write_only.return_value = False
- self.mock_table.options.manifest_merge_skip_on_write_only.return_value
= False
self.mock_snapshot_commit = Mock()
def _create_file_store_commit(self):
@@ -191,8 +186,6 @@ class TestFileStoreCommitRowTracking(unittest.TestCase):
snapshot_commit.commit.return_value = True
file_store_commit.snapshot_commit = snapshot_commit
file_store_commit.manifest_list_manager.read_all.return_value = []
- file_store_commit.manifest_file_merger = Mock()
- file_store_commit.manifest_file_merger.merge.return_value = ([], [])
file_store_commit._generate_partition_statistics = Mock(
return_value=[])
@@ -274,9 +267,6 @@ class TestFileStoreCommit(unittest.TestCase):
self.mock_table.table_path = '/test/table/path'
self.mock_table.file_io = Mock()
self.mock_table.options.manifest_target_size.return_value = 8 * 1024 *
1024
- self.mock_table.options.manifest_merge_min_count.return_value = 30
- self.mock_table.options.write_only.return_value = False
- self.mock_table.options.manifest_merge_skip_on_write_only.return_value
= False
# Mock snapshot commit
self.mock_snapshot_commit = Mock()
@@ -302,92 +292,6 @@ class TestFileStoreCommit(unittest.TestCase):
schema_id=0,
)
- def test_replace_manifest_files_uses_stable_value_equality(
- self, mock_manifest_list_manager, mock_manifest_file_manager):
- previous = [self._manifest_meta('a'), self._manifest_meta('b')]
- current = [
- self._manifest_meta('prefix'),
- self._manifest_meta('a'),
- self._manifest_meta('b'),
- 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)
-
- self.assertEqual(
- ['prefix', 'merged', 'suffix'],
- [manifest.file_name for manifest in result],
- )
- 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')
- b = self._manifest_meta('b')
- merged = self._manifest_meta('merged')
-
- self.assertIsNone(_try_replace_manifest_files(
- [a, self._manifest_meta('x'), b], [a, b], [merged]))
- self.assertEqual(
- ['a', 'merged'],
- [manifest.file_name for manifest in _try_replace_manifest_files(
- [a, self._manifest_meta('a'), b], [a, b], [merged])],
- )
- self.assertEqual(
- [merged], _try_replace_manifest_files([], [], [merged]))
- self.assertIsNone(_try_replace_manifest_files([a], [], [merged]))
-
- def test_manifest_merge_result_copies_and_freezes_lists(
- self, mock_manifest_list_manager, mock_manifest_file_manager):
- before = [self._manifest_meta('before')]
- after = [self._manifest_meta('after')]
-
- result = ManifestMergeResult(before, after)
- before.clear()
- after.clear()
-
- self.assertIsInstance(result.merge_before_manifests, tuple)
- self.assertIsInstance(result.merge_after_manifests, tuple)
- self.assertEqual(
- ['before'],
- [manifest.file_name
- for manifest in result.merge_before_manifests],
- )
- self.assertEqual(
- ['after'],
- [manifest.file_name
- for manifest in result.merge_after_manifests],
- )
-
def test_conflict_rollback_retry_skips_history_and_rescans_base(
self, mock_manifest_list_manager, mock_manifest_file_manager):
file_store_commit = self._create_file_store_commit()
@@ -462,7 +366,6 @@ class TestFileStoreCommit(unittest.TestCase):
def _run_manifest_commit_attempt(self, commit_side_effect=None,
commit_result=None, retry_result=None,
existing_manifests=None,
- merged_manifests=None,
latest_watermark=None):
file_store_commit = self._create_file_store_commit()
self.mock_table.identifier = 'default.test_table'
@@ -477,13 +380,9 @@ class TestFileStoreCommit(unittest.TestCase):
file_store_commit.snapshot_commit = snapshot_commit
before = self._manifest_meta('before')
- after = self._manifest_meta('after')
existing_manifests = (
[before] if existing_manifests is None
else existing_manifests)
- merged_manifests = (
- [after] if merged_manifests is None
- else merged_manifests)
delta = self._manifest_meta('delta')
file_store_commit._write_manifest_files = Mock(
return_value=[delta])
@@ -491,11 +390,6 @@ class TestFileStoreCommit(unittest.TestCase):
return_value=[])
file_store_commit.manifest_list_manager.read_all.return_value = (
existing_manifests)
- file_store_commit.manifest_file_merger = Mock()
- file_store_commit.manifest_file_merger.merge.return_value = (
- merged_manifests, merged_manifests)
- file_store_commit._clean_up_reuse_tmp_manifests = Mock()
- file_store_commit._clean_up_no_reuse_tmp_manifests = Mock()
latest_snapshot = Mock(
id=3,
@@ -528,109 +422,73 @@ class TestFileStoreCommit(unittest.TestCase):
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(
+ def test_commit_retries_preserve_latest_manifests(
self, mock_manifest_list_manager, mock_manifest_file_manager):
- file_store_commit, result = self._run_manifest_commit_attempt(
- commit_result=False)
-
- self.assertIsInstance(result, CommitFailRetryResult)
- self.assertIsNone(result.exception)
- self.assertEqual(
- ['before'],
- [manifest.file_name for manifest
- in result.manifest_merge_result.merge_before_manifests],
- )
- self.assertEqual(
- ['after'],
- [manifest.file_name for manifest
- in result.manifest_merge_result.merge_after_manifests],
- )
- file_store_commit.manifest_file_merger.merge.assert_called_once()
-
- def test_disabled_manifest_merge_preserves_manifests_on_retry(
- self, mock_manifest_list_manager, mock_manifest_file_manager):
- options = CoreOptions(Options({
- 'write-only': 'true',
- 'manifest.merge.skip-on-write-only': 'true',
- }))
- self.mock_table.options.write_only.side_effect = options.write_only
- self.mock_table.options.manifest_merge_skip_on_write_only.side_effect
= (
- options.manifest_merge_skip_on_write_only)
- current = [self._manifest_meta('before-a'),
self._manifest_meta('before-b')]
- first_commit, retry_result = self._run_manifest_commit_attempt(
- commit_result=False, existing_manifests=current)
-
- self.assertIsInstance(retry_result, CommitFailRetryResult)
- self.assertIsNone(retry_result.manifest_merge_result)
- first_commit.manifest_file_merger.merge.assert_not_called()
- self.assertEqual(
- current,
first_commit.manifest_list_manager.write.call_args_list[-1].args[1])
-
- current.append(self._manifest_meta('concurrent'))
- retry_commit, result = self._run_manifest_commit_attempt(
- commit_result=True, retry_result=retry_result,
existing_manifests=current)
-
- self.assertTrue(result.is_success())
- retry_commit.manifest_file_merger.merge.assert_not_called()
- self.assertEqual(
- current,
retry_commit.manifest_list_manager.write.call_args_list[-1].args[1])
-
- def test_atomic_commit_exception_does_not_retain_manifest_merge_result(
+ retry_result = None
+ attempts = [
+ ['before-a', 'before-b'],
+ ['before-a', 'before-b', 'concurrent'],
+ # External maintenance may replace the manifests between attempts.
+ ['compacted', 'concurrent'],
+ ]
+ for i, names in enumerate(attempts):
+ current = [self._manifest_meta(name) for name in names]
+ file_store_commit, retry_result =
self._run_manifest_commit_attempt(
+ commit_result=i == len(attempts) - 1,
+ retry_result=retry_result,
+ existing_manifests=current,
+ )
+ snapshot = file_store_commit.snapshot_commit.commit.call_args[0][1]
+ file_store_commit.manifest_list_manager.write.assert_any_call(
+ snapshot.base_manifest_list, current)
+ self.assertEqual(12, snapshot.total_record_count)
+ self.assertEqual(2, snapshot.delta_record_count)
+ self.mock_table.file_io.delete_quietly.assert_not_called()
+ if i < len(attempts) - 1:
+ self.assertIsInstance(retry_result, CommitFailRetryResult)
+ self.assertFalse(retry_result.commit_result_may_be_uncertain)
+ self.assertIsNone(retry_result.exception)
+ else:
+ self.assertTrue(retry_result.is_success())
+
+ def test_atomic_commit_exception_preserves_manifest_files(
self, mock_manifest_list_manager, mock_manifest_file_manager):
failure = TimeoutError('lost commit response')
- file_store_commit, result = self._run_manifest_commit_attempt(
+ _, result = self._run_manifest_commit_attempt(
commit_side_effect=failure)
self.assertIsInstance(result, CommitFailRetryResult)
self.assertIs(failure, result.exception)
self.assertTrue(result.commit_result_may_be_uncertain)
- self.assertIsNone(result.manifest_merge_result)
- file_store_commit._clean_up_reuse_tmp_manifests.assert_not_called()
- file_store_commit._clean_up_no_reuse_tmp_manifests.assert_not_called()
+ self.mock_table.file_io.delete_quietly.assert_not_called()
- def test_retry_reuses_manifest_merge_and_preserves_surrounding_files(
+ def test_prepare_failure_cleans_new_files_and_preserves_base_manifests(
self, mock_manifest_list_manager, mock_manifest_file_manager):
- previous_before = [
- self._manifest_meta('before-a'),
- self._manifest_meta('before-b'),
- ]
- previous_after = [self._manifest_meta('merged')]
- retry_result = CommitFailRetryResult(
- Mock(id=3),
- manifest_merge_result=ManifestMergeResult(
- previous_before, previous_after),
- )
- current = [
- self._manifest_meta('prefix'),
- self._manifest_meta('before-a'),
- self._manifest_meta('before-b'),
- self._manifest_meta('suffix'),
- ]
+ manager = mock_manifest_list_manager.return_value
+ manager.manifest_path = '/table/manifest'
+ mock_manifest_file_manager.return_value.manifest_path =
'/table/manifest'
+ base = self._manifest_meta('base')
+ base.extra_files = ['base.sidecar']
+ delta = self._manifest_meta('delta')
+ lists = {}
- file_store_commit, result = self._run_manifest_commit_attempt(
- commit_result=False,
- retry_result=retry_result,
- existing_manifests=current,
- )
+ def write_list(name, manifests):
+ lists[name] = manifests
+ if name.endswith('-0'):
+ raise OSError('base manifest list write failed')
- self.assertIsInstance(result, CommitFailRetryResult)
- self.assertEqual(
- ['prefix', 'before-a', 'before-b', 'suffix'],
- [manifest.file_name for manifest
- in result.manifest_merge_result.merge_before_manifests],
- )
- self.assertEqual(
- ['prefix', 'merged', 'suffix'],
- [manifest.file_name for manifest
- in result.manifest_merge_result.merge_after_manifests],
- )
- file_store_commit.manifest_file_merger.merge.assert_not_called()
- base_manifests = (
- file_store_commit.manifest_list_manager.write
- .call_args_list[-1].args[1])
+ manager.write.side_effect = write_list
+ manager.read.side_effect = lambda name: lists[name]
+ with self.assertRaisesRegex(RuntimeError, 'base manifest list write
failed'):
+ self._run_manifest_commit_attempt(existing_manifests=[base])
+
+ deleted = {
+ args[0] for args, _ in
self.mock_table.file_io.delete_quietly.call_args_list
+ }
self.assertEqual(
- ['prefix', 'merged', 'suffix'],
- [manifest.file_name for manifest in base_manifests],
+ {'/table/manifest/' + name for name in lists}
+ | {'/table/manifest/' + delta.file_name},
+ deleted,
)
def test_retry_preserves_concurrently_added_manifest_sidecar(
@@ -639,7 +497,6 @@ class TestFileStoreCommit(unittest.TestCase):
_, 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'])
@@ -650,91 +507,12 @@ class TestFileStoreCommit(unittest.TestCase):
)
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 = [
- self._manifest_meta('before-a'),
- self._manifest_meta('before-b'),
- ]
- retry_result = CommitFailRetryResult(
- Mock(id=3),
- manifest_merge_result=ManifestMergeResult(
- previous_before, [self._manifest_meta('merged')]),
- )
- current = [
- self._manifest_meta('before-a'),
- self._manifest_meta('between'),
- self._manifest_meta('before-b'),
- ]
-
- file_store_commit, result = self._run_manifest_commit_attempt(
- commit_result=False,
- retry_result=retry_result,
- existing_manifests=current,
- )
-
- self.assertIsInstance(result, CommitFailRetryResult)
- self.assertIsNone(result.manifest_merge_result)
- file_store_commit.manifest_file_merger.merge.assert_not_called()
- base_manifests = (
- file_store_commit.manifest_list_manager.write
- .call_args_list[-1].args[1])
- self.assertEqual(current, base_manifests)
-
- def test_manifest_merge_runs_once_across_multiple_retries(
- self, mock_manifest_list_manager, mock_manifest_file_manager):
- first_commit, retry_result = self._run_manifest_commit_attempt(
- commit_result=False,
- existing_manifests=[self._manifest_meta('before')],
- merged_manifests=[self._manifest_meta('merged')],
- )
- first_commit.manifest_file_merger.merge.assert_called_once()
-
- unchanged_retry, retry_result = self._run_manifest_commit_attempt(
- commit_result=False,
- retry_result=retry_result,
- existing_manifests=[self._manifest_meta('before')],
- )
- retry_commits = [unchanged_retry]
- current_names = ['before']
- for suffix in ['concurrent-1', 'concurrent-2']:
- current_names.append(suffix)
- retry_commit, retry_result = self._run_manifest_commit_attempt(
- commit_result=False,
- retry_result=retry_result,
- existing_manifests=[
- self._manifest_meta(name) for name in current_names
- ],
- )
- retry_commits.append(retry_commit)
-
- final_names = current_names + ['concurrent-3']
- final_commit, result = self._run_manifest_commit_attempt(
- commit_result=True,
- retry_result=retry_result,
- existing_manifests=[
- self._manifest_meta(name) for name in final_names
- ],
- )
-
- self.assertTrue(result.is_success())
- for retry_commit in retry_commits + [final_commit]:
- retry_commit.manifest_file_merger.merge.assert_not_called()
- base_manifests = (
- final_commit.manifest_list_manager.write
- .call_args_list[-1].args[1])
- self.assertEqual(
- ['merged', 'concurrent-1', 'concurrent-2', 'concurrent-3'],
- [manifest.file_name for manifest in base_manifests],
- )
-
def test_generate_partition_statistics_single_partition_single_file(
self, mock_manifest_list_manager, mock_manifest_file_manager):
"""Test partition statistics generation with single partition and
single file."""
diff --git
a/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
index 9638f04707..9aa6522246 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
@@ -24,7 +24,6 @@ from pypaimon.common.identifier import Identifier
from pypaimon.common.options import Options
from pypaimon.common.options.config import CatalogOptions
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
-from pypaimon.manifest.manifest_file_merger import ManifestFileMerger
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
@@ -153,52 +152,6 @@ class ManifestEntryIdentifierTest(unittest.TestCase):
len(final_entries), 0,
"ADD and DELETE entries with same identifier should both be
removed")
- def test_minor_compaction_cancels_add_delete_matching_same_file(self):
- partition = GenericRow([], [])
- add_entry = ManifestEntry(
- kind=0,
- partition=partition,
- bucket=0,
- total_buckets=1,
- file=self._create_file_meta("data-1.parquet", level=0)
- )
- delete_entry = ManifestEntry(
- kind=1,
- partition=partition,
- bucket=0,
- total_buckets=1,
- file=self._create_file_meta("data-1.parquet", level=0)
- )
-
- manifest_file_1 = ManifestFileMeta(
- file_name="manifest-minor-1.avro",
- file_size=1024,
- num_added_files=1,
- num_deleted_files=0,
- partition_stats=SimpleStats.empty_stats(),
- schema_id=0
- )
- manifest_file_2 = ManifestFileMeta(
- file_name="manifest-minor-2.avro",
- file_size=1024,
- num_added_files=0,
- num_deleted_files=1,
- partition_stats=SimpleStats.empty_stats(),
- schema_id=0
- )
- self.manifest_file_manager.write(manifest_file_1.file_name,
[add_entry])
- self.manifest_file_manager.write(manifest_file_2.file_name,
[delete_entry])
-
- merger = ManifestFileMerger(
- self.manifest_file_manager,
- suggested_meta_size=8 * 1024 * 1024,
- suggested_min_meta_count=2,
- )
- merged_files, new_files = merger.merge([manifest_file_1,
manifest_file_2])
-
- self.assertEqual(merged_files, [])
- self.assertEqual(new_files, [])
-
def test_add_delete_different_levels(self):
"""
Test that entries with different levels are NOT matched.
diff --git a/paimon-python/pypaimon/tests/test_early_row_range_filter.py
b/paimon-python/pypaimon/tests/test_early_row_range_filter.py
index b27184ab55..9961b29970 100644
--- a/paimon-python/pypaimon/tests/test_early_row_range_filter.py
+++ b/paimon-python/pypaimon/tests/test_early_row_range_filter.py
@@ -26,6 +26,8 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.predicate import Predicate
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
@@ -50,7 +52,6 @@ class TestManifestReadRowRangePerformance(unittest.TestCase):
schema = Schema.from_pyarrow_schema(pa_schema, options={
'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
- 'manifest.merge-min-count': '1',
})
cls.catalog.create_table('default.test_row_range_perf', schema, False)
cls.table = cls.catalog.get_table('default.test_row_range_perf')
@@ -74,16 +75,27 @@ class
TestManifestReadRowRangePerformance(unittest.TestCase):
tw.close()
tc.close()
+ # Build an externally compacted base manifest fixture so entry-level
+ # pruning is exercised even though Python commits never merge
manifests.
+ snapshot = cls.table.snapshot_manager().get_latest_snapshot()
+ manifest_lists = ManifestListManager(cls.table)
+ manifest_files = ManifestFileManager(cls.table)
+ entries = []
+ for meta in manifest_lists.read_base(snapshot):
+ entries.extend(manifest_files.read(meta.file_name,
drop_stats=False))
+ compacted = manifest_files.rolling_write(
+ entries, cls.table.options.manifest_target_size(),
'manifest-compacted-fixture')
+ assert len(compacted) == 1
+ manifest_lists.write(snapshot.base_manifest_list, compacted)
+
@classmethod
def tearDownClass(cls):
shutil.rmtree(cls.tempdir, ignore_errors=True)
@pytest.mark.python_plan
- def test_scan_constructs_all_entries_without_early_row_range_filter(self):
- """With manifest.merge-min-count=1, all entries are in one manifest.
- Querying with _ROW_ID BETWEEN 5 AND 14 should return 2 files, but
- the current code constructs DataFileMeta for ALL 20 entries because
- the row-range filter runs after full object construction."""
+ def test_scan_constructs_only_matching_entries_in_compacted_manifest(self):
+ """Querying _ROW_ID BETWEEN 5 AND 14 should construct only the two
+ matching files from the externally compacted base manifest."""
construction_count = [0]
original_init = DataFileMeta.__init__
@@ -106,7 +118,7 @@ class
TestManifestReadRowRangePerformance(unittest.TestCase):
self.assertEqual(sorted(actual.column('id').to_pylist()),
list(range(5, 15)))
- # 2 matching files × 2 (ADD + DELETE from manifest merge) = 4
+ # Each matching file is deserialized and then copied without stats.
self.assertLessEqual(
construction_count[0], 2 * total_files,
f"Expected at most {2 * total_files} DataFileMeta constructions, "
@@ -122,7 +134,6 @@ class
TestManifestReadRowRangePerformance(unittest.TestCase):
Schema.from_pyarrow_schema(pa_schema,
options={
'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
- 'manifest.merge-min-count': '1',
}), False)
table = self.catalog.get_table('default.test_add_delete_pair')
wb = table.new_batch_write_builder()
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index f2875f4030..8e1bbffaf2 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -642,17 +642,18 @@ class TableWriteTest(unittest.TestCase):
self.assertEqual(self.expected, actual)
@parameterized.expand([
- ('default', None, None, True),
- ('not_write_only_default', 'false', None, True),
- ('write_only_default', 'true', None, True),
- ('skip_enabled', None, 'true', True),
- ('not_write_only_skip_enabled', 'false', 'true', True),
- ('write_only_skip_enabled', 'true', 'true', False),
- ('skip_disabled', None, 'false', True),
- ('not_write_only_skip_disabled', 'false', 'false', True),
- ('write_only_skip_disabled', 'true', 'false', True),
+ ('default', None, None),
+ ('not_write_only_default', 'false', None),
+ ('write_only_default', 'true', None),
+ ('skip_enabled', None, 'true'),
+ ('not_write_only_skip_enabled', 'false', 'true'),
+ ('write_only_skip_enabled', 'true', 'true'),
+ ('skip_disabled', None, 'false'),
+ ('not_write_only_skip_disabled', 'false', 'false'),
+ ('write_only_skip_disabled', 'true', 'false'),
])
- def test_commit_manifest_merge(self, name, write_only, skip_on_write_only,
merge_enabled):
+ def test_commit_preserves_existing_manifests(self, name, write_only,
skip_on_write_only):
+ # Shared tables may still carry Java manifest maintenance options.
options = {'manifest.merge-min-count': '2'}
if write_only is not None:
options['write-only'] = write_only
@@ -700,16 +701,11 @@ class TableWriteTest(unittest.TestCase):
base_manifests =
manifest_list_manager.read(snapshot.base_manifest_list)
delta_manifests =
manifest_list_manager.read(snapshot.delta_manifest_list)
- if merge_enabled:
- self.assertEqual(len(base_manifests), 1)
- self.assertEqual(base_manifests[0].num_added_files, 2)
- self.assertEqual(base_manifests[0].num_deleted_files, 0)
- else:
- self.assertEqual(len(base_manifests), 2)
- self.assertEqual(
- [manifest.file_name for manifest in previous_manifests],
- [manifest.file_name for manifest in base_manifests],
- )
+ self.assertEqual(len(base_manifests), 2)
+ self.assertEqual(
+ [manifest.file_name for manifest in previous_manifests],
+ [manifest.file_name for manifest in base_manifests],
+ )
self.assertEqual(len(delta_manifests), 1)
expected = pa.Table.from_pydict(expected_data, schema=self.pa_schema)
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index a06c6e9d63..3fc49b1243 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -25,17 +25,15 @@ from pypaimon.build_info import full_version as
build_full_version
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
-from pypaimon.manifest.manifest_file_merger import ManifestFileMerger
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.file_entry import FileEntry
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
-from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import (PartitionStatistics,
SnapshotCommit)
-from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.write.commit.commit_rollback import CommitRollback
from pypaimon.write.commit.commit_scanner import CommitScanner
@@ -109,82 +107,17 @@ class SuccessResult(CommitResult):
return True
-def _manifest_file_key(manifest: ManifestFileMeta):
- stats = manifest.partition_stats
- return (
- manifest.file_name,
- manifest.file_size,
- manifest.num_added_files,
- manifest.num_deleted_files,
- GenericRowSerializer.to_bytes(stats.min_values),
- 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,
- )
-
-
-def _try_replace_manifest_files(current, replaced, replacement):
- """Replace the first contiguous occurrence while preserving list order."""
- current = list(current)
- replaced = list(replaced)
- replacement = list(replacement)
- if not replaced:
- return replacement if not current else None
-
- current_keys = [_manifest_file_key(manifest) for manifest in current]
- replaced_keys = [_manifest_file_key(manifest) for manifest in replaced]
- for start in range(len(current) - len(replaced) + 1):
- if current_keys[start:start + len(replaced)] == replaced_keys:
- return (
- current[:start]
- + replacement
- + current[start + len(replaced):]
- )
- return None
-
-
-class ManifestMergeResult:
- """Manifest merge input and output retained for a deterministic retry."""
-
- def __init__(self, merge_before_manifests, merge_after_manifests):
- self.merge_before_manifests = tuple(merge_before_manifests)
- self.merge_after_manifests = tuple(merge_after_manifests)
-
-
-def _try_reuse_manifest_merge_result(retry_result, current_manifests):
- if (not isinstance(retry_result, CommitFailRetryResult)
- or retry_result.commit_result_may_be_uncertain
- or retry_result.manifest_merge_result is None):
- return None
- previous = retry_result.manifest_merge_result
- return _try_replace_manifest_files(
- current_manifests,
- previous.merge_before_manifests,
- previous.merge_after_manifests,
- )
-
-
class RetryResult(CommitResult):
def __init__(self, latest_snapshot, exception: Optional[Exception] = None,
base_data_files: Optional[List[ManifestEntry]] = None,
- commit_result_may_be_uncertain: bool = False,
- manifest_merge_result: Optional[ManifestMergeResult] = None):
+ commit_result_may_be_uncertain: bool = False):
self.latest_snapshot = latest_snapshot
self.exception = exception
self.commit_result_may_be_uncertain = commit_result_may_be_uncertain
# Base entries as of latest_snapshot, carried so the next attempt
reuses
# them and reads only the incremental changes.
self.base_data_files = base_data_files
- self.manifest_merge_result = manifest_merge_result
def is_success(self) -> bool:
return False
@@ -232,14 +165,6 @@ class FileStoreCommit:
self.manifest_list_manager = ManifestListManager(table)
self.manifest_target_size = table.options.manifest_target_size()
- self.skip_manifest_merge_on_write_only = (
- table.options.write_only() and
table.options.manifest_merge_skip_on_write_only())
- self.manifest_merge_min_count =
table.options.manifest_merge_min_count()
- self.manifest_file_merger = ManifestFileMerger(
- self.manifest_file_manager,
- self.manifest_target_size,
- self.manifest_merge_min_count,
- )
self.commit_max_retries = table.options.commit_max_retries()
self.commit_timeout = table.options.commit_timeout()
@@ -707,10 +632,6 @@ class FileStoreCommit:
changelog_manifest_list_name = None
changelog_manifest_list_size = None
changelog_record_count = None
- merge_before_manifests = []
- merge_after_manifests = []
- merge_new_files = []
- skip_manifest_merge = False
try:
new_manifest_file_metas =
self._write_manifest_files(commit_entries, new_manifest_file)
self.manifest_list_manager.write(delta_manifest_list,
new_manifest_file_metas)
@@ -730,38 +651,17 @@ class FileStoreCommit:
changelog_record_count = sum(
entry.file.row_count for entry in changelog_entries if
entry.kind == 0)
- # process existing_manifest
+ # Manifest compaction is handled by external maintenance.
+ existing_manifests = []
total_record_count = 0
if latest_snapshot:
- merge_before_manifests = self.manifest_list_manager.read_all(
+ existing_manifests = self.manifest_list_manager.read_all(
latest_snapshot)
previous_record_count = latest_snapshot.total_record_count
if previous_record_count:
total_record_count += previous_record_count
- reused_manifests = (
- _try_reuse_manifest_merge_result(retry_result,
merge_before_manifests)
- if not self.skip_manifest_merge_on_write_only else None)
- skip_manifest_merge = (
- self.skip_manifest_merge_on_write_only
- or (reused_manifests is None and retry_result is not None))
- if reused_manifests is not None:
- merge_after_manifests = reused_manifests
- old_names = {
- manifest.file_name for manifest in merge_before_manifests
- }
- merge_new_files = [
- manifest for manifest in merge_after_manifests
- if manifest.file_name not in old_names
- ]
- elif skip_manifest_merge:
- merge_after_manifests = merge_before_manifests
- else:
- merge_after_manifests, merge_new_files = (
- self.manifest_file_merger.merge(
- merge_before_manifests))
- self.manifest_list_manager.write(
- base_manifest_list, merge_after_manifests)
+ self.manifest_list_manager.write(base_manifest_list,
existing_manifests)
delta_record_count = 0
for entry in commit_entries:
@@ -808,10 +708,9 @@ class FileStoreCommit:
statistics = self._generate_partition_statistics(commit_entries)
except Exception as e:
try:
- self._clean_up_reuse_tmp_manifests(
- delta_manifest_list, changelog_manifest_list_name,
new_index_manifest)
- self._clean_up_no_reuse_tmp_manifests(
- base_manifest_list, merge_new_files)
+ self._clean_up_tmp_manifests(
+ base_manifest_list, delta_manifest_list,
+ changelog_manifest_list_name, new_index_manifest)
except Exception as cleanup_err:
logger.warning(f"Failed to clean up temporary files:
{cleanup_err}",
exc_info=True)
@@ -837,19 +736,10 @@ class FileStoreCommit:
commit_kind,
commit_time_s,
)
- manifest_merge_result = (
- None
- if skip_manifest_merge
- else ManifestMergeResult(
- merge_before_manifests,
- merge_after_manifests,
- )
- )
return CommitFailRetryResult(
latest_snapshot,
None,
base_data_files=base_data_files,
- manifest_merge_result=manifest_merge_result,
)
except Exception as e:
# Commit exception, not sure about the situation and should not
clean up the files
@@ -859,7 +749,6 @@ class FileStoreCommit:
e,
base_data_files=base_data_files,
commit_result_may_be_uncertain=True,
- manifest_merge_result=None,
)
logger.info(
@@ -1100,16 +989,16 @@ class FileStoreCommit:
))
return commit_entries
- def _clean_up_reuse_tmp_manifests(
+ def _clean_up_tmp_manifests(
self,
+ base_manifest_list: Optional[str],
delta_manifest_list: Optional[str],
changelog_manifest_list: Optional[str],
index_manifest: Optional[str] = None):
- """Clean up delta/changelog manifests and index manifest.
-
- Mirrors Java CommitCleaner.cleanUpReuseTmpManifests.
- """
+ """Delete files created while preparing a snapshot, preserving base
manifests."""
manifest_path = self.manifest_list_manager.manifest_path
+ if base_manifest_list:
+
self.table.file_io.delete_quietly(f"{manifest_path}/{base_manifest_list}")
for ml_name in (delta_manifest_list, changelog_manifest_list):
if ml_name:
try:
@@ -1122,21 +1011,6 @@ class FileStoreCommit:
if index_manifest:
self.table.file_io.delete_quietly(f"{manifest_path}/{index_manifest}")
- def _clean_up_no_reuse_tmp_manifests(
- self,
- base_manifest_list: Optional[str],
- merge_new_files: List[ManifestFileMeta]):
- """Clean up base manifest list and newly created merge manifests.
-
- Mirrors Java CommitCleaner.cleanUpNoReuseTmpManifests.
- """
- manifest_path = self.manifest_list_manager.manifest_path
- if base_manifest_list:
-
self.table.file_io.delete_quietly(f"{manifest_path}/{base_manifest_list}")
- for meta in merge_new_files:
- self.table.file_io.delete_quietly(
- f"{self.manifest_file_manager.manifest_path}/{meta.file_name}")
-
def abort(self, commit_messages: List[CommitMessage]):
"""Abort commit and delete files. Uses external_path if available to
ensure proper scheme handling."""
_abort_commit_messages(self.table, commit_messages)
