From 75ce10f2e60e43a712cf4427732c7f695940240f Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Thu, 3 Sep 2026 17:27:15 +0800 Subject: [PATCH 1/2] feat: add rewrite_if predicate to RewriteManifests - Support selective manifest rewriting via rewrite_if(predicate) - Allow single manifest rewriting when matching predicate (needed for #3840) - Update rewrites_needed() to evaluate predicate - Add unit tests for selective rewriting and single manifest predicates --- pyiceberg/table/update/snapshot.py | 34 ++++++++++++--- tests/table/test_rewrite_manifests.py | 60 +++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 6 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index b54cd6bcc5..11628e62fb 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -1326,6 +1326,7 @@ class RewriteManifests(_SnapshotProducer["RewriteManifests"]): """ _computed_manifests: list[ManifestFile] | None + _predicate: Callable[[ManifestFile], bool] | None _rewritten_count: int _created_count: int @@ -1347,12 +1348,25 @@ def __init__( "the first-row-id of rewritten manifests must be preserved, " "see: https://github.com/apache/iceberg-python/issues/3621" ) + self._predicate = None self._rewritten_count = 0 self._created_count = 0 self._kept_count = 0 self._entries_processed = 0 self._computed_manifests = None + def rewrite_if(self, predicate: Callable[[ManifestFile], bool]) -> RewriteManifests: + """Filter which manifests should be rewritten. + + Args: + predicate: A function that takes a ManifestFile and returns True if it should be rewritten. + + Returns: + This RewriteManifests instance for method chaining. + """ + self._predicate = predicate + return self + def _deleted_entries(self) -> list[ManifestEntry]: return [] @@ -1395,25 +1409,31 @@ def _existing_manifests(self) -> list[ManifestFile]: kept_manifests: list[ManifestFile] = [] for manifest in snapshot.manifests(self._io): if manifest.content == ManifestContent.DATA: - data_manifests_by_spec[manifest.partition_spec_id].append(manifest) + if self._predicate is None or self._predicate(manifest): + data_manifests_by_spec[manifest.partition_spec_id].append(manifest) + else: + kept_manifests.append(manifest) else: kept_manifests.append(manifest) new_manifests: list[ManifestFile] = [] for spec_id, manifests in data_manifests_by_spec.items(): for group in self._group_by_target_size(manifests): - if len(group) == 1: - # nothing to merge; keep the manifest as-is + if len(group) == 1 and self._predicate is None: + # nothing to merge and no predicate specified; keep the manifest as-is kept_manifests.append(group[0]) continue + entries_in_group = 0 with self.new_manifest_writer(self.spec(spec_id)) as writer: for manifest in group: for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True): writer.existing(entry) self._entries_processed += 1 - new_manifests.append(writer.to_manifest_file()) + entries_in_group += 1 + if entries_in_group > 0: + new_manifests.append(writer.to_manifest_file()) + self._created_count += 1 self._rewritten_count += len(group) - self._created_count += 1 self._kept_count = len(kept_manifests) self.snapshot_properties = { @@ -1434,9 +1454,11 @@ def _commit(self) -> UpdatesAndRequirements: return super()._commit() def rewrites_needed(self) -> bool: - """Return whether the current snapshot has more than one data manifest to merge.""" + """Return whether the current snapshot has data manifests to rewrite.""" snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch or MAIN_BRANCH) if snapshot is None: return False data_manifests = [m for m in snapshot.manifests(self._io) if m.content == ManifestContent.DATA] + if self._predicate is not None: + return any(self._predicate(m) for m in data_manifests) return len(data_manifests) > 1 diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py index 2c1f841c37..7d6b5ecc8b 100644 --- a/tests/table/test_rewrite_manifests.py +++ b/tests/table/test_rewrite_manifests.py @@ -141,3 +141,63 @@ def test_rewrites_needed(catalog: Catalog) -> None: table.append(_arrow_table(offset=3)) table = catalog.load_table("default.test_rewrite") assert table.maintenance.rewrite_manifests().rewrites_needed() is True + + +def test_rewrite_manifests_with_predicate_selective(catalog: Catalog) -> None: + table = _create_table_with_appends(catalog, appends=3) + manifests_before = _data_manifests(table) + assert len(manifests_before) == 3 + target_manifest = manifests_before[0] + target_path = target_manifest.manifest_path + rows_before = table.scan().to_arrow().sort_by("id") + + table.maintenance.rewrite_manifests().rewrite_if(lambda m: m.manifest_path == target_path).commit() + + table = catalog.load_table("default.test_rewrite") + manifests_after = _data_manifests(table) + assert len(manifests_after) == 3 + assert table.scan().to_arrow().sort_by("id") == rows_before + + snapshot = table.current_snapshot() + assert snapshot is not None + assert snapshot.summary is not None + assert snapshot.summary["manifests-created"] == "1" + assert snapshot.summary["manifests-replaced"] == "1" + assert snapshot.summary["manifests-kept"] == "2" + assert snapshot.summary["entries-processed"] == "1" + + +def test_rewrite_manifests_single_manifest_with_predicate(catalog: Catalog) -> None: + table = _create_table_with_appends(catalog, appends=1) + snapshot_before = table.current_snapshot() + assert snapshot_before is not None + manifest_path_before = _data_manifests(table)[0].manifest_path + + # with predicate, even single manifest should be rewritten + table.maintenance.rewrite_manifests().rewrite_if(lambda m: True).commit() + + table = catalog.load_table("default.test_rewrite") + manifests_after = _data_manifests(table) + assert len(manifests_after) == 1 + assert manifests_after[0].manifest_path != manifest_path_before + assert manifests_after[0].existing_files_count == 1 + assert manifests_after[0].added_files_count == 0 + + snapshot = table.current_snapshot() + assert snapshot is not None + assert snapshot.snapshot_id != snapshot_before.snapshot_id + assert snapshot.summary is not None + assert snapshot.summary["manifests-created"] == "1" + assert snapshot.summary["manifests-replaced"] == "1" + assert snapshot.summary["manifests-kept"] == "0" + assert snapshot.summary["entries-processed"] == "1" + + +def test_rewrites_needed_with_predicate(catalog: Catalog) -> None: + table = _create_table_with_appends(catalog, appends=1) + # single manifest without predicate: False + assert table.maintenance.rewrite_manifests().rewrites_needed() is False + # single manifest with matching predicate: True + assert table.maintenance.rewrite_manifests().rewrite_if(lambda m: True).rewrites_needed() is True + # single manifest with non-matching predicate: False + assert table.maintenance.rewrite_manifests().rewrite_if(lambda m: False).rewrites_needed() is False From 6afbd5f7dedd50838a2bd865400a1abf55a3bd0f Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Fri, 4 Sep 2026 13:39:00 +0800 Subject: [PATCH 2/2] fix(table): avoid writing empty manifest for fully-deleted manifests and V1 edge cases - Guard ManifestWriter by peeking first live entry with itertools.chain to avoid empty manifest files - Retain plain rewrite_manifests behavior merging live and fully-deleted manifests into one - Clarify rewrite_if docstring regarding single-manifest optimization - Add regression tests for kept paths, fully deleted manifests, and merging dead manifests --- pyiceberg/table/update/snapshot.py | 32 +++++----- tests/table/test_rewrite_manifests.py | 87 ++++++++++++++++++++++++++- 2 files changed, 103 insertions(+), 16 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 11628e62fb..1cb17ea126 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -1358,6 +1358,9 @@ def __init__( def rewrite_if(self, predicate: Callable[[ManifestFile], bool]) -> RewriteManifests: """Filter which manifests should be rewritten. + Passing a predicate also disables the optimization that keeps single-manifest + groups as-is, allowing single manifests to be rewritten when they match the predicate. + Args: predicate: A function that takes a ManifestFile and returns True if it should be rewritten. @@ -1408,11 +1411,8 @@ def _existing_manifests(self) -> list[ManifestFile]: data_manifests_by_spec: defaultdict[int, list[ManifestFile]] = defaultdict(list) kept_manifests: list[ManifestFile] = [] for manifest in snapshot.manifests(self._io): - if manifest.content == ManifestContent.DATA: - if self._predicate is None or self._predicate(manifest): - data_manifests_by_spec[manifest.partition_spec_id].append(manifest) - else: - kept_manifests.append(manifest) + if manifest.content == ManifestContent.DATA and (self._predicate is None or self._predicate(manifest)): + data_manifests_by_spec[manifest.partition_spec_id].append(manifest) else: kept_manifests.append(manifest) @@ -1423,16 +1423,20 @@ def _existing_manifests(self) -> list[ManifestFile]: # nothing to merge and no predicate specified; keep the manifest as-is kept_manifests.append(group[0]) continue - entries_in_group = 0 + + entries = (entry for manifest in group for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True)) + first_entry = next(entries, None) + if first_entry is None: + kept_manifests.extend(group) + continue + with self.new_manifest_writer(self.spec(spec_id)) as writer: - for manifest in group: - for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True): - writer.existing(entry) - self._entries_processed += 1 - entries_in_group += 1 - if entries_in_group > 0: - new_manifests.append(writer.to_manifest_file()) - self._created_count += 1 + for entry in itertools.chain([first_entry], entries): + writer.existing(entry) + self._entries_processed += 1 + + new_manifests.append(writer.to_manifest_file()) + self._created_count += 1 self._rewritten_count += len(group) self._kept_count = len(kept_manifests) diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py index 7d6b5ecc8b..5e00d53d2c 100644 --- a/tests/table/test_rewrite_manifests.py +++ b/tests/table/test_rewrite_manifests.py @@ -147,16 +147,39 @@ def test_rewrite_manifests_with_predicate_selective(catalog: Catalog) -> None: table = _create_table_with_appends(catalog, appends=3) manifests_before = _data_manifests(table) assert len(manifests_before) == 3 - target_manifest = manifests_before[0] - target_path = target_manifest.manifest_path + + # 1. Extract all manifest paths and underlying data file paths before rewrite + paths_before = [m.manifest_path for m in manifests_before] + target_path = paths_before[0] + kept_paths_before = set(paths_before[1:]) rows_before = table.scan().to_arrow().sort_by("id") + data_files_before = [ + entry.data_file.file_path for m in manifests_before for entry in m.fetch_manifest_entry(table.io, discard_deleted=True) + ] + # 2. Execute selective rewrite table.maintenance.rewrite_manifests().rewrite_if(lambda m: m.manifest_path == target_path).commit() + # 3. Reload table and extract new paths for comprehensive verification table = catalog.load_table("default.test_rewrite") manifests_after = _data_manifests(table) + paths_after = {m.manifest_path for m in manifests_after} + assert len(manifests_after) == 3 + # Manifest paths not rewritten remain unchanged + assert kept_paths_before.issubset(paths_after) + # Old path of rewritten target manifest is gone + assert target_path not in paths_after + # Exactly one new manifest was created + new_manifest_paths = paths_after - kept_paths_before + assert len(new_manifest_paths) == 1 + + # Verify table data and underlying data files are completely preserved assert table.scan().to_arrow().sort_by("id") == rows_before + data_files_after = [ + entry.data_file.file_path for m in manifests_after for entry in m.fetch_manifest_entry(table.io, discard_deleted=True) + ] + assert set(data_files_before) == set(data_files_after) snapshot = table.current_snapshot() assert snapshot is not None @@ -201,3 +224,63 @@ def test_rewrites_needed_with_predicate(catalog: Catalog) -> None: assert table.maintenance.rewrite_manifests().rewrite_if(lambda m: True).rewrites_needed() is True # single manifest with non-matching predicate: False assert table.maintenance.rewrite_manifests().rewrite_if(lambda m: False).rewrites_needed() is False + + +def test_rewrite_manifests_with_fully_deleted_manifest(catalog: Catalog) -> None: + table = catalog.create_table("default.test_fully_deleted", schema=pa.schema([pa.field("id", pa.int64())])) + table.append(_arrow_table(offset=0)) # manifest 1: id 1, 2, 3 + table.append(_arrow_table(offset=3)) # manifest 2: id 4, 5, 6 + table.delete("id <= 3") # deletes id 1, 2, 3 + + snapshot_before = table.current_snapshot() + manifests_before = _data_manifests(table) + assert len(manifests_before) == 2 + paths_before = [m.manifest_path for m in manifests_before] + + # Target rewrite for manifest containing only deleted entries + table.maintenance.rewrite_manifests().rewrite_if(lambda m: (m.deleted_files_count or 0) > 0).commit() + + table = catalog.load_table("default.test_fully_deleted") + assert table.current_snapshot() == snapshot_before + + # Verify both manifest paths remain unchanged + manifests_after = _data_manifests(table) + assert len(manifests_after) == 2 + assert [m.manifest_path for m in manifests_after] == paths_before + + # Verify the second manifest with live data (ids 4, 5, 6) is intact and readable + assert table.scan().to_arrow().sort_by("id") == _arrow_table(offset=3) + live_entries_manifest2 = manifests_after[1].fetch_manifest_entry(table.io, discard_deleted=True) + assert len(live_entries_manifest2) == 1 + assert live_entries_manifest2[0].data_file.record_count == 3 + + +def test_rewrite_manifests_predicate_matching_nothing(catalog: Catalog) -> None: + table = _create_table_with_appends(catalog, appends=2) + snapshot_before = table.current_snapshot() + table.maintenance.rewrite_manifests().rewrite_if(lambda m: False).commit() + table = catalog.load_table("default.test_rewrite") + assert table.current_snapshot() == snapshot_before + + +def test_rewrite_manifests_merges_live_and_fully_deleted_manifests(catalog: Catalog) -> None: + table = catalog.create_table("default.test_merge_deleted", schema=pa.schema([pa.field("id", pa.int64())])) + table.append(_arrow_table(offset=0)) # manifest 1: id 1, 2, 3 + table.append(_arrow_table(offset=3)) # manifest 2: id 4, 5, 6 + table.delete("id <= 3") # fully deletes manifest 1 + + manifests_before = _data_manifests(table) + assert len(manifests_before) == 2 + + # Plain rewrite_manifests() without predicate should merge live and fully-deleted manifests into 1 + assert table.maintenance.rewrite_manifests().rewrites_needed() is True + table.maintenance.rewrite_manifests().commit() + + table = catalog.load_table("default.test_merge_deleted") + manifests_after = _data_manifests(table) + assert len(manifests_after) == 1 + assert manifests_after[0].existing_files_count == 1 + assert manifests_after[0].added_files_count == 0 + + # Verify table data is fully preserved and matches remaining live records + assert table.scan().to_arrow().sort_by("id") == _arrow_table(offset=3)