Support rewrite_manifests table maintenance

mkdocs/docs/api.md

@@ -1485,7 +1485,7 @@ table.manage_snapshots().remove_branch("dev").commit()
 
 ## Table Maintenance
 
-PyIceberg provides table maintenance operations through the `table.maintenance` API. This provides a clean interface for performing maintenance tasks like snapshot expiration.
+PyIceberg provides table maintenance operations through the `table.maintenance` API. This provides a clean interface for performing maintenance tasks like snapshot expiration and manifest rewriting.
 
 ### Snapshot Expiration
 
@@ -1527,6 +1527,45 @@ def cleanup_old_snapshots(table_name: str, snapshot_ids: list[int]):
 cleanup_old_snapshots("analytics.user_events", [12345, 67890, 11111])
 ```
 
+### Manifest Rewriting
+
+Rewrite the current snapshot's data manifests without changing any data. Live entries are regrouped into new manifests sized by the `commit.manifest.target-size-bytes` table property and written as `EXISTING` entries that keep their sequence numbers, which keeps scan planning fast on tables that accumulate many small manifests through frequent appends. Delete manifests are kept as-is, and the result is committed as a `replace` snapshot:
+
+```python
+table.maintenance.rewrite_manifests().commit()
+```
+
+Nothing is committed when no manifests would be merged, so the call is safe to schedule periodically. To find out in advance whether a rewrite would do anything:
+
+```python
+if table.maintenance.rewrite_manifests().rewrites_needed():
+    table.maintenance.rewrite_manifests().commit()
+```
+
+The `replace` snapshot records what the operation did in its summary:
+
+```python
+table.current_snapshot().summary
+# {'operation': 'replace', 'manifests-created': '1', 'manifests-kept': '0',
+#  'manifests-replaced': '3', 'entries-processed': '3', ...}
+```
+
+By default every data manifest is a candidate. Pass a predicate to `rewrite_if` to choose which ones to rewrite; manifests that do not match are kept exactly as they are. This is the way to target a subset, such as manifests written by an older library version whose contents need re-encoding:
+
+```python
+# Rewrite only the manifests smaller than 1 MiB, leaving larger ones untouched
+table.maintenance.rewrite_manifests().rewrite_if(lambda manifest: manifest.manifest_length < 1024 * 1024).commit()
+```
+
+A predicate also turns off the shortcut that leaves a lone manifest alone: a manifest that matches is rewritten even when there is nothing to merge it with. `rewrite_if(lambda manifest: True)` therefore rewrites every data manifest, which is not the same as passing no predicate at all.
+
+<!-- prettier-ignore-start -->
+
+!!! note "V3 tables"
+    Rewriting manifests on V3 tables raises `NotImplementedError`, because the `first-row-id` of rewritten manifests has to be preserved and that support is still pending ([#3621](https://github.com/apache/iceberg-python/issues/3621)).
+
+<!-- prettier-ignore-end -->
+
 ## Views
 
 If PyIceberg is unable to automatically determine view support on your REST Catalog, you can manually specify, `"view-endpoints-supported": "true"`:

pyiceberg/table/maintenance.py

@@ -24,7 +24,7 @@
 
 if TYPE_CHECKING:
     from pyiceberg.table import Table
-    from pyiceberg.table.update.snapshot import ExpireSnapshots
+    from pyiceberg.table.update.snapshot import ExpireSnapshots, RewriteManifests
 
 
 class MaintenanceTable:
@@ -43,3 +43,17 @@ def expire_snapshots(self) -> ExpireSnapshots:
         from pyiceberg.table.update.snapshot import ExpireSnapshots
 
         return ExpireSnapshots(transaction=Transaction(self.tbl, autocommit=True))
+
+    def rewrite_manifests(self) -> RewriteManifests:
+        """Return a RewriteManifests operation that merges the current snapshot's data manifests.
+
+        Entries are rewritten as EXISTING, keeping their sequence numbers; delete
+        manifests are kept as-is. The result is committed as a `replace` snapshot.
+
+        Returns:
+            RewriteManifests operation; call commit() to execute it.
+        """
+        from pyiceberg.table import Transaction
+        from pyiceberg.table.update.snapshot import RewriteManifests
+
+        return RewriteManifests(transaction=Transaction(self.tbl, autocommit=True), io=self.tbl.io)

pyiceberg/table/snapshots.py

@@ -351,7 +351,7 @@ def _partition_summary(self, update_metrics: UpdateMetrics) -> str:
 
 
 def update_snapshot_summaries(summary: Summary, previous_summary: Mapping[str, str] | None = None) -> Summary:
-    if summary.operation not in {Operation.APPEND, Operation.OVERWRITE, Operation.DELETE}:
+    if summary.operation not in {Operation.APPEND, Operation.OVERWRITE, Operation.DELETE, Operation.REPLACE}:
         raise ValueError(f"Operation not implemented: {summary.operation}")
 
     if not previous_summary:

pyiceberg/table/update/snapshot.py

@@ -1312,3 +1312,185 @@ def older_than(self, dt: datetime) -> ExpireSnapshots:
             if snapshot.timestamp_ms < expire_from and snapshot.snapshot_id not in protected_ids:
                 self._snapshot_ids_to_expire.add(snapshot.snapshot_id)
         return self
+
+
+class RewriteManifests(_SnapshotProducer["RewriteManifests"]):
+    """Rewrite the current snapshot's data manifests without changing data.
+
+    Live entries from the rewritten data manifests are regrouped into new
+    manifests sized by `commit.manifest.target-size-bytes`, written as EXISTING
+    entries that keep their sequence numbers. Entries with status DELETED are
+    dropped, matching the reference implementation, which rewrites live entries
+    only. Delete manifests are kept as-is. The result is committed as a
+    `replace` snapshot; if no manifests need merging, no snapshot is committed.
+    """
+
+    _computed_manifests: list[ManifestFile] | None
+    _manifest_predicate: Callable[[ManifestFile], bool] | None
+
+    _rewritten_count: int
+    _created_count: int
+    _kept_count: int
+    _entries_processed: int
+
+    def __init__(
+        self,
+        transaction: Transaction,
+        io: FileIO,
+        commit_uuid: uuid.UUID | None = None,
+        snapshot_properties: dict[str, str] = EMPTY_DICT,
+        branch: str | None = MAIN_BRANCH,
+    ) -> None:
+        super().__init__(Operation.REPLACE, transaction, io, commit_uuid, snapshot_properties, branch)
+        if transaction.table_metadata.format_version >= 3:
+            raise NotImplementedError(
+                "Rewriting manifests is not yet supported for V3 tables: "
+                "the first-row-id of rewritten manifests must be preserved, "
+                "see: https://github.com/apache/iceberg-python/issues/3621"
+            )
+        self._manifest_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.
+
+        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.
+
+        Returns:
+            This RewriteManifests instance for method chaining.
+        """
+        self._manifest_predicate = predicate
+        return self
+
+    def _deleted_entries(self) -> list[ManifestEntry]:
+        return []
+
+    def _target_size_bytes(self) -> int:
+        from pyiceberg.table import TableProperties
+
+        return property_as_int(  # type: ignore
+            self._transaction.table_metadata.properties,
+            TableProperties.MANIFEST_TARGET_SIZE_BYTES,
+            TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT,
+        )
+
+    def _group_by_target_size(self, manifests: list[ManifestFile]) -> list[list[ManifestFile]]:
+        """Pack manifests into groups whose source sizes add up to roughly the target size."""
+        target_size = self._target_size_bytes()
+        groups: list[list[ManifestFile]] = []
+        current_group: list[ManifestFile] = []
+        current_size = 0
+        for manifest in manifests:
+            if current_group and current_size + manifest.manifest_length > target_size:
+                groups.append(current_group)
+                current_group = []
+                current_size = 0
+            current_group.append(manifest)
+            current_size += manifest.manifest_length
+        if current_group:
+            groups.append(current_group)
+        return groups
+
+    def _existing_manifests(self) -> list[ManifestFile]:
+        if self._computed_manifests is not None:
+            return self._computed_manifests
+
+        snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch or MAIN_BRANCH)
+        if snapshot is None:
+            self._computed_manifests = []
+            return self._computed_manifests
+
+        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 and (
+                self._manifest_predicate is None or self._manifest_predicate(manifest)
+            ):
+                data_manifests_by_spec[manifest.partition_spec_id].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 and self._manifest_predicate is None:
+                    # nothing to merge and no predicate specified; keep the manifest as-is
+                    kept_manifests.append(group[0])
+                    continue
+
+                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 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)
+        self.snapshot_properties = {
+            **self.snapshot_properties,
+            "manifests-created": str(self._created_count),
+            "manifests-kept": str(self._kept_count),
+            "manifests-replaced": str(self._rewritten_count),
+            "entries-processed": str(self._entries_processed),
+        }
+        self._computed_manifests = new_manifests + kept_manifests
+        return self._computed_manifests
+
+    def _refresh_for_retry(self) -> None:
+        """Reset state for a retry attempt, discarding the plan built from the stale branch head."""
+        super()._refresh_for_retry()
+        # The plan and its counters depend on the branch head, which changes on retry. Keeping them
+        # would rebuild the replacement snapshot from the manifests of the superseded snapshot,
+        # dropping whatever was committed concurrently.
+        self._computed_manifests = None
+        self._rewritten_count = 0
+        self._created_count = 0
+        self._kept_count = 0
+        self._entries_processed = 0
+
+    def _commit(self) -> UpdatesAndRequirements:
+        self._existing_manifests()
+        if self._created_count == 0:
+            # nothing was merged; committing would only produce a pointless replace snapshot
+            return (), ()
+        return super()._commit()
+
+    def rewrites_needed(self) -> bool:
+        """Return whether committing would rewrite any manifest.
+
+        This mirrors the grouping that `_existing_manifests` performs, so a snapshot whose data
+        manifests all end up kept as-is reports False. A predicate that matches only manifests
+        holding no live entries still reports True, because deciding that requires reading them.
+        """
+        snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch or MAIN_BRANCH)
+        if snapshot is None:
+            return False
+
+        data_manifests_by_spec: defaultdict[int, list[ManifestFile]] = defaultdict(list)
+        for manifest in snapshot.manifests(self._io):
+            if manifest.content == ManifestContent.DATA and (
+                self._manifest_predicate is None or self._manifest_predicate(manifest)
+            ):
+                data_manifests_by_spec[manifest.partition_spec_id].append(manifest)
+
+        return any(
+            len(group) > 1 or self._manifest_predicate is not None
+            for manifests in data_manifests_by_spec.values()
+            for group in self._group_by_target_size(manifests)
+        )

tests/integration/test_rewrite_manifests_interop.py

@@ -0,0 +1,74 @@
+# 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.
+# pylint:disable=redefined-outer-name
+from typing import TYPE_CHECKING
+
+import pyarrow as pa
+import pytest
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.exceptions import NoSuchTableError
+from pyiceberg.manifest import ManifestContent
+
+if TYPE_CHECKING:
+    from pyspark.sql import SparkSession
+
+
+@pytest.mark.integration
+def test_spark_reads_table_after_rewrite_manifests(session_catalog: Catalog, spark: "SparkSession") -> None:
+    identifier = "default.test_rewrite_manifests_interop"
+    try:
+        session_catalog.drop_table(identifier)
+    except NoSuchTableError:
+        pass
+
+    table = session_catalog.create_table(identifier, schema=pa.schema([pa.field("id", pa.int64())]))
+    for i in range(3):
+        table.append(pa.table({"id": pa.array([i * 3 + 1, i * 3 + 2, i * 3 + 3], type=pa.int64())}))
+
+    table = session_catalog.load_table(identifier)
+    snapshot = table.current_snapshot()
+    assert snapshot is not None
+    assert len([m for m in snapshot.manifests(table.io) if m.content == ManifestContent.DATA]) == 3
+
+    table.maintenance.rewrite_manifests().commit()
+
+    table = session_catalog.load_table(identifier)
+    snapshot = table.current_snapshot()
+    assert snapshot is not None
+    assert len([m for m in snapshot.manifests(table.io) if m.content == ManifestContent.DATA]) == 1
+
+    # Spark must read the rewritten table with the same data
+    spark_rows = spark.table(f"{identifier}").collect()
+    assert sorted(row.id for row in spark_rows) == list(range(1, 10))
+
+    # Spark must see the replace snapshot and the preserved data files
+    snapshots = spark.sql(f"SELECT operation FROM {identifier}.snapshots ORDER BY committed_at").collect()
+    assert [row.operation for row in snapshots] == ["append", "append", "append", "replace"]
+
+    files = spark.sql(f"SELECT file_path FROM {identifier}.files").collect()
+    assert len(files) == 3
+
+    # Spark sees the same manifest consolidation
+    manifests = spark.sql(f"SELECT path FROM {identifier}.manifests").collect()
+    assert len(manifests) == 1
+
+    # time travel to the pre-rewrite snapshot still works from Spark
+    previous_snapshot_id = snapshot.parent_snapshot_id
+    assert previous_snapshot_id is not None
+    previous_rows = spark.sql(f"SELECT id FROM {identifier} VERSION AS OF {previous_snapshot_id}").collect()
+    assert sorted(row.id for row in previous_rows) == list(range(1, 10))

tests/table/test_commit_retry.py

@@ -302,6 +302,40 @@ def test_delete_files_refresh_clears_compute_deletes_cache(catalog: Catalog) ->
     assert "_compute_deletes" not in producer.__dict__
 
 
+def test_rewrite_manifests_refresh_clears_computed_manifests(catalog: Catalog) -> None:
+    """Verify that _refresh_for_retry discards the plan RewriteManifests computed for the old head."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    table = catalog.create_table("default.rewrite_cache_test", schema=schema)
+
+    import pyarrow as pa
+
+    table.append(pa.table({"x": [1, 2, 3]}))
+    table.append(pa.table({"x": [4, 5, 6]}))
+    table = catalog.load_table("default.rewrite_cache_test")
+
+    from pyiceberg.table.update.snapshot import RewriteManifests
+
+    tx = Transaction(table, autocommit=False)
+    producer = RewriteManifests(transaction=tx, io=table.io)
+
+    # Plan against the current head, populating the cache and the counters
+    planned = producer._existing_manifests()
+
+    assert planned
+    assert producer._created_count == 1
+    assert producer._rewritten_count == 2
+
+    producer._refresh_for_retry()
+
+    # Reusing either would rebuild the replacement snapshot from the superseded head
+    assert producer._computed_manifests is None
+    assert producer._created_count == 0
+    assert producer._rewritten_count == 0
+    assert producer._kept_count == 0
+    assert producer._entries_processed == 0
+
+
 def test_concurrent_overwrite_overwrite_raises_validation_exception(catalog: Catalog) -> None:
     """Concurrent overwrites on the same data should fail with ValidationException."""
     catalog.create_namespace("default")