From 919f59822005741c6072207298a1b9a990ca74a2 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 08:29:57 +0900
Subject: [PATCH 01/12] Support rewrite_manifests table maintenance

Add table.maintenance.rewrite_manifests(), which merges the current
snapshot's data manifests into fewer manifests sized by
commit.manifest.target-size-bytes. Entries are rewritten as EXISTING
and keep their sequence numbers; delete manifests and manifests that
need no merging are kept as-is. The result is committed as a replace
snapshot whose totals carry over unchanged.

V3 tables are rejected for now: rewriting must preserve the
first-row-id of rewritten manifests, which needs the read side of row
lineage (#3621).

Closes #3629
---
 pyiceberg/table/maintenance.py        |  16 ++-
 pyiceberg/table/snapshots.py          |   2 +-
 pyiceberg/table/update/snapshot.py    | 111 ++++++++++++++++++++
 tests/table/test_rewrite_manifests.py | 146 ++++++++++++++++++++++++++
 tests/table/test_snapshots.py         |  21 +++-
 5 files changed, 290 insertions(+), 6 deletions(-)
 create mode 100644 tests/table/test_rewrite_manifests.py

diff --git a/pyiceberg/table/maintenance.py b/pyiceberg/table/maintenance.py
index 0fcda35ae9..d0f4e0d0ea 100644
--- a/pyiceberg/table/maintenance.py
+++ b/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)
diff --git a/pyiceberg/table/snapshots.py b/pyiceberg/table/snapshots.py
index 5e9e519a01..14d15b2c66 100644
--- a/pyiceberg/table/snapshots.py
+++ b/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:
diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index 57215dca04..9ba6104415 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -1312,3 +1312,114 @@ 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. Delete manifests are kept as-is.
+    The result is committed as a `replace` snapshot.
+    """
+
+    _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._rewritten_count = 0
+        self._created_count = 0
+        self._kept_count = 0
+        self._entries_processed = 0
+
+    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]:
+        snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch or MAIN_BRANCH)
+        if snapshot is None:
+            return []
+
+        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:
+                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:
+                    # nothing to merge; keep the manifest as-is
+                    kept_manifests.append(group[0])
+                    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
+                new_manifests.append(writer.to_manifest_file())
+                self._rewritten_count += len(group)
+                self._created_count += 1
+
+        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),
+        }
+        return new_manifests + kept_manifests
+
+    def rewrites_needed(self) -> bool:
+        """Return whether the current snapshot has more than one data manifest to merge."""
+        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]
+        return len(data_manifests) > 1
diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py
new file mode 100644
index 0000000000..1e3d2b91c5
--- /dev/null
+++ b/tests/table/test_rewrite_manifests.py
@@ -0,0 +1,146 @@
+# 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.
+from pathlib import Path
+
+import pyarrow as pa
+import pytest
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.catalog.memory import InMemoryCatalog
+from pyiceberg.manifest import ManifestContent
+from pyiceberg.table import Table
+from pyiceberg.table.snapshots import Operation
+
+
+@pytest.fixture
+def catalog(tmp_path: Path) -> Catalog:
+    catalog = InMemoryCatalog("test.rewrite_manifests", warehouse=f"file://{tmp_path}")
+    catalog.create_namespace("default")
+    return catalog
+
+
+def _arrow_table(offset: int = 0) -> pa.Table:
+    return pa.table({"id": pa.array([offset + 1, offset + 2, offset + 3], type=pa.int64())})
+
+
+def _create_table_with_appends(catalog: Catalog, appends: int = 3) -> Table:
+    table = catalog.create_table("default.test_rewrite", schema=pa.schema([pa.field("id", pa.int64())]))
+    for i in range(appends):
+        table.append(_arrow_table(offset=i * 3))
+    return table
+
+
+def _data_manifests(table: Table) -> list:
+    snapshot = table.current_snapshot()
+    assert snapshot is not None
+    return [m for m in snapshot.manifests(table.io) if m.content == ManifestContent.DATA]
+
+
+def test_rewrite_manifests_merges_data_manifests(catalog: Catalog) -> None:
+    table = _create_table_with_appends(catalog, appends=3)
+    assert len(_data_manifests(table)) == 3
+    rows_before = table.scan().to_arrow().sort_by("id")
+
+    table.maintenance.rewrite_manifests().commit()
+
+    table = catalog.load_table("default.test_rewrite")
+    manifests = _data_manifests(table)
+    assert len(manifests) == 1
+    # entries are rewritten as EXISTING
+    assert manifests[0].existing_files_count == 3
+    assert manifests[0].added_files_count == 0
+
+    # data is unchanged
+    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.operation == Operation.REPLACE
+    assert snapshot.summary["manifests-created"] == "1"
+    assert snapshot.summary["manifests-replaced"] == "3"
+    assert snapshot.summary["entries-processed"] == "3"
+    # totals carry over unchanged
+    assert snapshot.summary["total-data-files"] == "3"
+    assert snapshot.summary["total-records"] == "9"
+
+
+def test_rewrite_manifests_preserves_sequence_numbers(catalog: Catalog) -> None:
+    table = _create_table_with_appends(catalog, appends=3)
+    entries_before = {
+        entry.data_file.file_path: entry.sequence_number
+        for manifest in _data_manifests(table)
+        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True)
+    }
+
+    table.maintenance.rewrite_manifests().commit()
+
+    table = catalog.load_table("default.test_rewrite")
+    manifests = _data_manifests(table)
+    entries_after = {
+        entry.data_file.file_path: entry.sequence_number
+        for manifest in manifests
+        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True)
+    }
+    assert entries_after == entries_before
+    # the merged manifest keeps the min sequence number of its entries
+    assert manifests[0].min_sequence_number == min(entries_before.values())
+
+
+def test_rewrite_manifests_single_manifest_is_noop_kept(catalog: Catalog) -> None:
+    table = _create_table_with_appends(catalog, appends=1)
+    manifest_path_before = _data_manifests(table)[0].manifest_path
+
+    table.maintenance.rewrite_manifests().commit()
+
+    table = catalog.load_table("default.test_rewrite")
+    manifests = _data_manifests(table)
+    assert len(manifests) == 1
+    # a single manifest is kept as-is, not rewritten
+    assert manifests[0].manifest_path == manifest_path_before
+
+    snapshot = table.current_snapshot()
+    assert snapshot is not None
+    assert snapshot.summary is not None
+    assert snapshot.summary["manifests-created"] == "0"
+    assert snapshot.summary["manifests-replaced"] == "0"
+
+
+def test_rewrite_manifests_respects_target_size(catalog: Catalog) -> None:
+    table = _create_table_with_appends(catalog, appends=4)
+    manifest_length = _data_manifests(table)[0].manifest_length
+
+    # allow roughly two source manifests per group
+    with table.transaction() as tx:
+        tx.set_properties({"commit.manifest.target-size-bytes": str(manifest_length * 2)})
+
+    table = catalog.load_table("default.test_rewrite")
+    table.maintenance.rewrite_manifests().commit()
+
+    table = catalog.load_table("default.test_rewrite")
+    manifests = _data_manifests(table)
+    assert len(manifests) == 2
+    assert all(m.existing_files_count == 2 for m in manifests)
+
+
+def test_rewrites_needed(catalog: Catalog) -> None:
+    table = _create_table_with_appends(catalog, appends=1)
+    assert table.maintenance.rewrite_manifests().rewrites_needed() is False
+
+    table.append(_arrow_table(offset=3))
+    table = catalog.load_table("default.test_rewrite")
+    assert table.maintenance.rewrite_manifests().rewrites_needed() is True
diff --git a/tests/table/test_snapshots.py b/tests/table/test_snapshots.py
index 5f1680ed59..9e23669cd0 100644
--- a/tests/table/test_snapshots.py
+++ b/tests/table/test_snapshots.py
@@ -399,10 +399,23 @@ def test_merge_snapshot_summaries_overwrite_summary() -> None:
     assert actual.additional_properties == expected
 
 
-def test_invalid_operation() -> None:
-    with pytest.raises(ValueError) as e:
-        update_snapshot_summaries(summary=Summary(Operation.REPLACE))
-    assert "Operation not implemented: Operation.REPLACE" in str(e.value)
+def test_replace_operation_carries_totals() -> None:
+    actual = update_snapshot_summaries(
+        summary=Summary(Operation.REPLACE),
+        previous_summary={
+            "total-data-files": "3",
+            "total-delete-files": "0",
+            "total-records": "9",
+            "total-files-size": "1234",
+            "total-position-deletes": "0",
+            "total-equality-deletes": "0",
+        },
+    )
+
+    # a replace operation does not change any of the totals
+    assert actual["total-data-files"] == "3"
+    assert actual["total-records"] == "9"
+    assert actual["total-files-size"] == "1234"
 
 
 def test_invalid_type() -> None:

From e00c63440ab6bd39116258e1ba45fe9d30b3e146 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 09:09:38 +0900
Subject: [PATCH 02/12] Address review: skip the commit when no manifests need
 merging

Committing a replace snapshot that merely re-lists the same manifests
has no value; return no updates instead, and assert in the no-op test
that the current snapshot is unchanged. Document that rewrites carry
live entries only, matching the reference implementation, and add the
Spark interop integration test (data, snapshot history, file and
manifest counts, and pre-rewrite time travel verified from Spark).
---
 pyiceberg/table/update/snapshot.py          | 25 +++++--
 tests/integration/test_rewrite_manifests.py | 74 +++++++++++++++++++++
 tests/table/test_rewrite_manifests.py       | 19 +++---
 3 files changed, 103 insertions(+), 15 deletions(-)
 create mode 100644 tests/integration/test_rewrite_manifests.py

diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index 9ba6104415..b54cd6bcc5 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -1319,10 +1319,14 @@ class RewriteManifests(_SnapshotProducer["RewriteManifests"]):
 
     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. Delete manifests are kept as-is.
-    The result is committed as a `replace` snapshot.
+    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
+
     _rewritten_count: int
     _created_count: int
     _kept_count: int
@@ -1347,6 +1351,7 @@ def __init__(
         self._created_count = 0
         self._kept_count = 0
         self._entries_processed = 0
+        self._computed_manifests = None
 
     def _deleted_entries(self) -> list[ManifestEntry]:
         return []
@@ -1378,9 +1383,13 @@ def _group_by_target_size(self, manifests: list[ManifestFile]) -> list[list[Mani
         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:
-            return []
+            self._computed_manifests = []
+            return self._computed_manifests
 
         data_manifests_by_spec: defaultdict[int, list[ManifestFile]] = defaultdict(list)
         kept_manifests: list[ManifestFile] = []
@@ -1414,7 +1423,15 @@ def _existing_manifests(self) -> list[ManifestFile]:
             "manifests-replaced": str(self._rewritten_count),
             "entries-processed": str(self._entries_processed),
         }
-        return new_manifests + kept_manifests
+        self._computed_manifests = new_manifests + kept_manifests
+        return self._computed_manifests
+
+    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 the current snapshot has more than one data manifest to merge."""
diff --git a/tests/integration/test_rewrite_manifests.py b/tests/integration/test_rewrite_manifests.py
new file mode 100644
index 0000000000..511c3bdd54
--- /dev/null
+++ b/tests/integration/test_rewrite_manifests.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"integration.{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 integration.{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 integration.{identifier}.files").collect()
+    assert len(files) == 3
+
+    # Spark sees the same manifest consolidation
+    manifests = spark.sql(f"SELECT path FROM integration.{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 integration.{identifier} VERSION AS OF {previous_snapshot_id}").collect()
+    assert sorted(row.id for row in previous_rows) == list(range(1, 10))
diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py
index 1e3d2b91c5..b3a655292b 100644
--- a/tests/table/test_rewrite_manifests.py
+++ b/tests/table/test_rewrite_manifests.py
@@ -21,7 +21,7 @@
 
 from pyiceberg.catalog import Catalog
 from pyiceberg.catalog.memory import InMemoryCatalog
-from pyiceberg.manifest import ManifestContent
+from pyiceberg.manifest import ManifestContent, ManifestFile
 from pyiceberg.table import Table
 from pyiceberg.table.snapshots import Operation
 
@@ -44,7 +44,7 @@ def _create_table_with_appends(catalog: Catalog, appends: int = 3) -> Table:
     return table
 
 
-def _data_manifests(table: Table) -> list:
+def _data_manifests(table: Table) -> list[ManifestFile]:
     snapshot = table.current_snapshot()
     assert snapshot is not None
     return [m for m in snapshot.manifests(table.io) if m.content == ManifestContent.DATA]
@@ -101,23 +101,20 @@ def test_rewrite_manifests_preserves_sequence_numbers(catalog: Catalog) -> None:
     assert manifests[0].min_sequence_number == min(entries_before.values())
 
 
-def test_rewrite_manifests_single_manifest_is_noop_kept(catalog: Catalog) -> None:
+def test_rewrite_manifests_single_manifest_is_noop(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
 
     table.maintenance.rewrite_manifests().commit()
 
     table = catalog.load_table("default.test_rewrite")
-    manifests = _data_manifests(table)
-    assert len(manifests) == 1
-    # a single manifest is kept as-is, not rewritten
-    assert manifests[0].manifest_path == manifest_path_before
-
+    # nothing to merge: no new snapshot is committed and the manifest is untouched
     snapshot = table.current_snapshot()
     assert snapshot is not None
-    assert snapshot.summary is not None
-    assert snapshot.summary["manifests-created"] == "0"
-    assert snapshot.summary["manifests-replaced"] == "0"
+    assert snapshot.snapshot_id == snapshot_before.snapshot_id
+    assert _data_manifests(table)[0].manifest_path == manifest_path_before
 
 
 def test_rewrite_manifests_respects_target_size(catalog: Catalog) -> None:

From 59d98ff21d45a245594307b11581b5a8741af305 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 09:55:52 +0900
Subject: [PATCH 03/12] Rename the interop test to avoid a duplicate module
 name

tests/table/test_rewrite_manifests.py and the integration test shared
a module name, which fails mypy collection.
---
 ...est_rewrite_manifests.py => test_rewrite_manifests_interop.py} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename tests/integration/{test_rewrite_manifests.py => test_rewrite_manifests_interop.py} (100%)

diff --git a/tests/integration/test_rewrite_manifests.py b/tests/integration/test_rewrite_manifests_interop.py
similarity index 100%
rename from tests/integration/test_rewrite_manifests.py
rename to tests/integration/test_rewrite_manifests_interop.py

From 4b586f7657817326ac3cc1cf79502cf2e4122f4f Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 10:02:13 +0900
Subject: [PATCH 04/12] Fix mypy type narrowing in sequence number preservation
 test

---
 tests/table/test_rewrite_manifests.py | 24 ++++++++++++------------
 1 file changed, 12 insertions(+), 12 deletions(-)

diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py
index b3a655292b..5d27c0afa7 100644
--- a/tests/table/test_rewrite_manifests.py
+++ b/tests/table/test_rewrite_manifests.py
@@ -79,26 +79,26 @@ def test_rewrite_manifests_merges_data_manifests(catalog: Catalog) -> None:
     assert snapshot.summary["total-records"] == "9"
 
 
+def _sequence_numbers_by_file(table: Table) -> dict[str, int]:
+    result: dict[str, int] = {}
+    for manifest in _data_manifests(table):
+        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True):
+            assert entry.sequence_number is not None
+            result[entry.data_file.file_path] = entry.sequence_number
+    return result
+
+
 def test_rewrite_manifests_preserves_sequence_numbers(catalog: Catalog) -> None:
     table = _create_table_with_appends(catalog, appends=3)
-    entries_before = {
-        entry.data_file.file_path: entry.sequence_number
-        for manifest in _data_manifests(table)
-        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True)
-    }
+    entries_before = _sequence_numbers_by_file(table)
 
     table.maintenance.rewrite_manifests().commit()
 
     table = catalog.load_table("default.test_rewrite")
-    manifests = _data_manifests(table)
-    entries_after = {
-        entry.data_file.file_path: entry.sequence_number
-        for manifest in manifests
-        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True)
-    }
+    entries_after = _sequence_numbers_by_file(table)
     assert entries_after == entries_before
     # the merged manifest keeps the min sequence number of its entries
-    assert manifests[0].min_sequence_number == min(entries_before.values())
+    assert _data_manifests(table)[0].min_sequence_number == min(entries_before.values())
 
 
 def test_rewrite_manifests_single_manifest_is_noop(catalog: Catalog) -> None:

From d799ef98e6914cf2fa2801d60f6b0566b781afaf Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 10:04:00 +0900
Subject: [PATCH 05/12] Make the target-size grouping test robust to manifest
 size variation

The target was exactly twice the first manifest's length, so a
second manifest one byte larger started its own group and the test
was flaky.
---
 tests/table/test_rewrite_manifests.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py
index 5d27c0afa7..2c1f841c37 100644
--- a/tests/table/test_rewrite_manifests.py
+++ b/tests/table/test_rewrite_manifests.py
@@ -119,11 +119,11 @@ def test_rewrite_manifests_single_manifest_is_noop(catalog: Catalog) -> None:
 
 def test_rewrite_manifests_respects_target_size(catalog: Catalog) -> None:
     table = _create_table_with_appends(catalog, appends=4)
-    manifest_length = _data_manifests(table)[0].manifest_length
+    max_manifest_length = max(m.manifest_length for m in _data_manifests(table))
 
-    # allow roughly two source manifests per group
+    # allow two source manifests per group (2x fits, 3x exceeds), robust to small size variations
     with table.transaction() as tx:
-        tx.set_properties({"commit.manifest.target-size-bytes": str(manifest_length * 2)})
+        tx.set_properties({"commit.manifest.target-size-bytes": str(int(max_manifest_length * 2.5))})
 
     table = catalog.load_table("default.test_rewrite")
     table.maintenance.rewrite_manifests().commit()

From dd052009d469e43e0253da30aa3a4ee495c45939 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 10:12:04 +0900
Subject: [PATCH 06/12] Use the default Spark catalog in the interop test,
 matching other integration tests

---
 tests/integration/test_rewrite_manifests_interop.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/tests/integration/test_rewrite_manifests_interop.py b/tests/integration/test_rewrite_manifests_interop.py
index 511c3bdd54..bb44c3bfd8 100644
--- a/tests/integration/test_rewrite_manifests_interop.py
+++ b/tests/integration/test_rewrite_manifests_interop.py
@@ -53,22 +53,22 @@ def test_spark_reads_table_after_rewrite_manifests(session_catalog: Catalog, spa
     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"integration.{identifier}").collect()
+    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 integration.{identifier}.snapshots ORDER BY committed_at").collect()
+    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 integration.{identifier}.files").collect()
+    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 integration.{identifier}.manifests").collect()
+    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 integration.{identifier} VERSION AS OF {previous_snapshot_id}").collect()
+    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))

From d46350b6e2e2843ad0449fa492edbead6e3cdaa3 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Fri, 4 Sep 2026 09:32:59 +0900
Subject: [PATCH 07/12] Docs: Document rewrite_manifests in the table
 maintenance section

---
 mkdocs/docs/api.md | 32 +++++++++++++++++++++++++++++++-
 1 file changed, 31 insertions(+), 1 deletion(-)

diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md
index 0a2a72b6f0..7376a14ae2 100644
--- a/mkdocs/docs/api.md
+++ b/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,36 @@ 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 there is only one data manifest, 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', ...}
+```
+
+<!-- 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"`:

From 928e59adcb427956b992317f5c31d141e63b9e33 Mon Sep 17 00:00:00 2001
From: hedger9487 <hedger9487@gmail.com>
Date: Thu, 3 Sep 2026 17:27:15 +0800
Subject: [PATCH 08/12] 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 85375d4a026b927fe638268514cbd617307116b8 Mon Sep 17 00:00:00 2001
From: hedger9487 <hedger9487@gmail.com>
Date: Fri, 4 Sep 2026 13:39:00 +0800
Subject: [PATCH 09/12] 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)

From a47f58f852e5cc999ecbba05ae14174314b59f4b Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Sat, 5 Sep 2026 20:37:59 +0900
Subject: [PATCH 10/12] Docs: Document the rewrite_if predicate of
 rewrite_manifests

---
 mkdocs/docs/api.md | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md
index 7376a14ae2..d1f88a4067 100644
--- a/mkdocs/docs/api.md
+++ b/mkdocs/docs/api.md
@@ -1550,6 +1550,15 @@ table.current_snapshot().summary
 #  '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"

From 1a7ab16ee1f6ac9a4da6ed4c3026274915079297 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Sat, 5 Sep 2026 21:20:59 +0900
Subject: [PATCH 11/12] Rename the rewrite_if predicate field to avoid
 shadowing _SnapshotProducer._predicate

---
 pyiceberg/table/update/snapshot.py | 16 +++++++++-------
 1 file changed, 9 insertions(+), 7 deletions(-)

diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index 1cb17ea126..a859d91c69 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -1326,7 +1326,7 @@ class RewriteManifests(_SnapshotProducer["RewriteManifests"]):
     """
 
     _computed_manifests: list[ManifestFile] | None
-    _predicate: Callable[[ManifestFile], bool] | None
+    _manifest_predicate: Callable[[ManifestFile], bool] | None
 
     _rewritten_count: int
     _created_count: int
@@ -1348,7 +1348,7 @@ 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._manifest_predicate = None
         self._rewritten_count = 0
         self._created_count = 0
         self._kept_count = 0
@@ -1367,7 +1367,7 @@ def rewrite_if(self, predicate: Callable[[ManifestFile], bool]) -> RewriteManife
         Returns:
             This RewriteManifests instance for method chaining.
         """
-        self._predicate = predicate
+        self._manifest_predicate = predicate
         return self
 
     def _deleted_entries(self) -> list[ManifestEntry]:
@@ -1411,7 +1411,9 @@ 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 and (self._predicate is None or self._predicate(manifest)):
+            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)
@@ -1419,7 +1421,7 @@ def _existing_manifests(self) -> list[ManifestFile]:
         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._predicate is None:
+                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
@@ -1463,6 +1465,6 @@ def rewrites_needed(self) -> bool:
         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)
+        if self._manifest_predicate is not None:
+            return any(self._manifest_predicate(m) for m in data_manifests)
         return len(data_manifests) > 1

From cadafef3f7e438f2d693a4c868b81c87b4f43c1f Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Sun, 6 Sep 2026 04:31:50 +0900
Subject: [PATCH 12/12] Replan the manifest rewrite on commit retry and align
 rewrites_needed with it

---
 mkdocs/docs/api.md                    |  2 +-
 pyiceberg/table/update/snapshot.py    | 36 +++++++++++++++++++++++----
 tests/table/test_commit_retry.py      | 34 +++++++++++++++++++++++++
 tests/table/test_rewrite_manifests.py | 19 ++++++++++++++
 4 files changed, 85 insertions(+), 6 deletions(-)

diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md
index d1f88a4067..fd2f5a2624 100644
--- a/mkdocs/docs/api.md
+++ b/mkdocs/docs/api.md
@@ -1535,7 +1535,7 @@ Rewrite the current snapshot's data manifests without changing any data. Live en
 table.maintenance.rewrite_manifests().commit()
 ```
 
-Nothing is committed when there is only one data manifest, so the call is safe to schedule periodically. To find out in advance whether a rewrite would do anything:
+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():
diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index a859d91c69..12bb38afb9 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -1452,6 +1452,18 @@ def _existing_manifests(self) -> list[ManifestFile]:
         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:
@@ -1460,11 +1472,25 @@ def _commit(self) -> UpdatesAndRequirements:
         return super()._commit()
 
     def rewrites_needed(self) -> bool:
-        """Return whether the current snapshot has data manifests to rewrite."""
+        """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 = [m for m in snapshot.manifests(self._io) if m.content == ManifestContent.DATA]
-        if self._manifest_predicate is not None:
-            return any(self._manifest_predicate(m) for m in data_manifests)
-        return len(data_manifests) > 1
+
+        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)
+        )
diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py
index ce5dca96aa..98d15d4f9b 100644
--- a/tests/table/test_commit_retry.py
+++ b/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")
diff --git a/tests/table/test_rewrite_manifests.py b/tests/table/test_rewrite_manifests.py
index 5e00d53d2c..8c73ad2f08 100644
--- a/tests/table/test_rewrite_manifests.py
+++ b/tests/table/test_rewrite_manifests.py
@@ -284,3 +284,22 @@ def test_rewrite_manifests_merges_live_and_fully_deleted_manifests(catalog: Cata
 
     # Verify table data is fully preserved and matches remaining live records
     assert table.scan().to_arrow().sort_by("id") == _arrow_table(offset=3)
+
+
+def test_rewrites_needed_is_false_when_every_group_is_a_single_manifest(catalog: Catalog) -> None:
+    table = _create_table_with_appends(catalog, appends=3)
+    # A tiny target size puts every manifest in its own group, so nothing can be merged
+    with table.transaction() as tx:
+        tx.set_properties({"commit.manifest.target-size-bytes": "1"})
+
+    table = catalog.load_table("default.test_rewrite")
+    assert len(_data_manifests(table)) == 3
+
+    snapshot_before = table.current_snapshot()
+    assert table.maintenance.rewrite_manifests().rewrites_needed() is False
+
+    table.maintenance.rewrite_manifests().commit()
+
+    table = catalog.load_table("default.test_rewrite")
+    assert table.current_snapshot() == snapshot_before
+    assert len(_data_manifests(table)) == 3