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).

Author: moomindani

pyiceberg/table/update/snapshot.py

@@ -1138,10 +1138,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
@@ -1166,6 +1170,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 []
@@ -1197,9 +1202,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] = []
@@ -1233,7 +1242,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."""

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))

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: