Add snapshot selection to manifest inspection

mkdocs/docs/api.md

@@ -853,6 +853,20 @@ To show a table's current file manifests:
 table.inspect.manifests()
 ```
 
+To inspect a retained snapshot, pass its snapshot ID:
+
+```python
+table.inspect.manifests(snapshot_id=123456789)
+```
+
+Partition summaries use the snapshot's schema and each manifest's partition spec.
+If a snapshot has no recorded schema ID, the current schema is used. Sources dropped
+from that schema are resolved by field ID from retained schemas.
+An unknown or expired snapshot ID raises `ValueError`. Without a snapshot ID, a table
+with no snapshots returns an empty table with the manifest metadata schema.
+`all_manifests()` continues to include all retained snapshots, including repeated
+manifests, and does not accept a snapshot ID.
+
 ```python
 pyarrow.Table
 content: int8 not null

pyiceberg/table/inspect.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import itertools
+import warnings
 from collections.abc import Iterator
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
@@ -558,7 +559,18 @@ def _partition_summaries_to_rows(
             rows = []
             for i, field_summary in enumerate(partition_summaries):
                 field = spec.fields[i]
-                partition_field_type = spec.partition_type(self.tbl.schema()).fields[i].field_type
+                partition_schema = schema
+                if field.source_id not in schema.field_ids:
+                    # Retained manifests can reference partition sources dropped before this snapshot.
+                    partition_schema = next(
+                        (
+                            old_schema
+                            for old_schema in reversed(self.tbl.metadata.schemas)
+                            if field.source_id in old_schema.field_ids
+                        ),
+                        schema,
+                    )
+                partition_field_type = spec.partition_type(partition_schema).fields[i].field_type
                 lower_bound = (
                     (
                         field.transform.to_human_string(
@@ -590,6 +602,12 @@ def _partition_summaries_to_rows(
         specs = self.tbl.metadata.specs()
         manifests = []
         if snapshot:
+            schema = self.tbl.schema()
+            if snapshot.schema_id is not None:
+                if snapshot_schema := self.tbl.metadata.schema_by_id(snapshot.schema_id):
+                    schema = snapshot_schema
+                else:
+                    warnings.warn(f"Metadata does not contain schema with id: {snapshot.schema_id}", stacklevel=2)
             for manifest in snapshot.manifests(self.tbl.io):
                 is_data_file = manifest.content == ManifestContent.DATA
                 is_delete_file = manifest.content == ManifestContent.DELETES
@@ -619,15 +637,22 @@ def _partition_summaries_to_rows(
             schema=self._get_all_manifests_schema() if is_all_manifests_table else self._get_manifests_schema(),
         )
 
-    def manifests(self) -> pa.Table:
-        """Return the manifest files for the current snapshot as a PyArrow Table.
+    def manifests(self, snapshot_id: int | None = None) -> pa.Table:
+        """Return the manifest files for a snapshot as a PyArrow Table.
+
+        Args:
+            snapshot_id: Optional snapshot ID. If None, uses the current snapshot.
 
         Returns:
-            pa.Table: Manifest metadata for the current snapshot, or an empty
-                table if the table has no snapshots.
+            pa.Table: Manifest metadata for the selected snapshot, or an empty
+                table if no snapshot ID is supplied and the table has no snapshots.
                 See ``_get_manifests_schema`` for the full column list.
+
+        Raises:
+            ValueError: If the supplied snapshot ID is not found.
         """
-        return self._generate_manifests_table(self.tbl.current_snapshot())
+        snapshot = self._get_snapshot(snapshot_id) if snapshot_id is not None else self.tbl.current_snapshot()
+        return self._generate_manifests_table(snapshot)
 
     def metadata_log_entries(self) -> pa.Table:
         """Return the metadata log of the table as a PyArrow Table.

tests/table/test_inspect.py

@@ -27,9 +27,9 @@
 from pyiceberg.schema import Schema
 from pyiceberg.table.inspect import InspectTable, _readable_bound
 from pyiceberg.table.snapshots import Snapshot
-from pyiceberg.transforms import IdentityTransform
+from pyiceberg.transforms import BucketTransform, IdentityTransform
 from pyiceberg.typedef import Record
-from pyiceberg.types import NestedField, StringType
+from pyiceberg.types import DoubleType, FloatType, IntegerType, LongType, NestedField, PrimitiveType, StringType
 from tests.catalog.test_base import InMemoryCatalog
 
 
@@ -106,3 +106,148 @@ def test_inspect_manifests_preserves_empty_string_bounds(catalog: InMemoryCatalo
     partition_summary = tbl.inspect.manifests().to_pydict()["partition_summaries"][0][0]
     assert partition_summary["lower_bound"] == ""
     assert partition_summary["upper_bound"] == ""
+
+
+def test_inspect_manifests_snapshot_selection(catalog: InMemoryCatalog) -> None:
+    tbl = catalog.create_table("default.manifests", Schema(NestedField(1, "s", StringType())))
+    empty = tbl.inspect.manifests()
+    assert empty.num_rows == 0
+    assert empty.equals(tbl.inspect.manifests(snapshot_id=None))
+    assert tbl.inspect.all_manifests().num_rows == 0
+
+    data = pa.table({"s": ["first"]})
+    tbl.append(data)
+    first = tbl.current_snapshot()
+    assert first is not None
+    first_rows = tbl.inspect.manifests()
+    tbl.append(data)
+    current = tbl.current_snapshot()
+    assert current is not None
+    current_rows = tbl.inspect.manifests()
+
+    assert empty.schema == first_rows.schema == current_rows.schema
+    assert first_rows.equals(tbl.inspect.manifests(snapshot_id=first.snapshot_id))
+    assert current_rows.equals(tbl.inspect.manifests(snapshot_id=current.snapshot_id))
+    assert first_rows["path"].to_pylist() == [manifest.manifest_path for manifest in first.manifests(tbl.io)]
+    assert current_rows["path"].to_pylist() == [manifest.manifest_path for manifest in current.manifests(tbl.io)]
+    assert first_rows.num_rows == 1
+    assert current_rows.num_rows == 2
+    assert first_rows["partition_summaries"].to_pylist() == [[]]
+
+    all_rows = tbl.inspect.all_manifests()
+    assert (
+        all_rows.schema.remove(all_rows.schema.get_field_index("key_metadata")).remove(
+            all_rows.schema.get_field_index("reference_snapshot_id")
+        )
+        == empty.schema
+    )
+    for snapshot, expected in [(first, first_rows), (current, current_rows)]:
+        rows = [row for row in all_rows.to_pylist() if row.pop("reference_snapshot_id") == snapshot.snapshot_id]
+        assert all(row.pop("key_metadata") is None for row in rows)
+        assert rows == expected.to_pylist()
+    assert all_rows.num_rows == 3  # Shared manifests remain repeated for each reference snapshot.
+
+    tbl.maintenance.expire_snapshots().by_id(first.snapshot_id).commit()
+    with pytest.raises(ValueError, match=f"Cannot find snapshot with ID {first.snapshot_id}"):
+        tbl.inspect.manifests(snapshot_id=first.snapshot_id)
+
+
+@pytest.mark.parametrize("snapshot_id", [0, -1, 9223372036854775807])
+@pytest.mark.parametrize("populated", [False, True])
+def test_inspect_manifests_invalid_snapshot(catalog: InMemoryCatalog, snapshot_id: int, populated: bool) -> None:
+    tbl = catalog.create_table("default.invalid_snapshot", Schema(NestedField(1, "s", StringType())))
+    if populated:
+        tbl.append(pa.table({"s": ["value"]}))
+    with pytest.raises(ValueError, match=f"Cannot find snapshot with ID {snapshot_id}"):
+        tbl.inspect.manifests(snapshot_id=snapshot_id)
+
+
+@pytest.mark.parametrize("value", ["old", "", None])
+def test_inspect_manifests_schema_and_partition_evolution(catalog: InMemoryCatalog, value: str | None) -> None:
+    schema = Schema(NestedField(1, "s", StringType()), NestedField(2, "id", IntegerType()))
+    spec = PartitionSpec(PartitionField(1, 1000, IdentityTransform(), "s_part"))
+    tbl = catalog.create_table("default.evolved_manifests", schema, partition_spec=spec)
+    tbl.append(pa.table({"s": [value], "id": [7]}, schema=pa.schema([("s", pa.string()), ("id", pa.int32())])))
+    first = tbl.current_snapshot()
+    assert first is not None
+    first_rows = tbl.inspect.manifests()
+    assert first_rows["partition_summaries"].to_pylist() == [
+        [{"contains_null": value is None, "contains_nan": False, "lower_bound": value, "upper_bound": value}]
+    ]
+
+    with tbl.update_schema() as update:
+        update.rename_column("s", "renamed")
+    assert tbl.inspect.manifests(first.snapshot_id).equals(first_rows)
+    with tbl.update_spec() as update:
+        update.remove_field("s_part")
+        update.add_field("id", BucketTransform(8), "id_bucket")
+    with tbl.update_schema() as update:
+        update.delete_column("renamed")
+        # Reusing a name must not cause the old source ID to be resolved to the new type.
+        update.add_column("s", LongType())
+    tbl.append(pa.table({"s": [99], "id": [8]}, schema=pa.schema([("s", pa.int64()), ("id", pa.int32())])))
+    current = tbl.current_snapshot()
+    assert current is not None
+    tbl = catalog.load_table(tbl.name())
+
+    assert tbl.inspect.manifests(first.snapshot_id).equals(first_rows)
+    current_rows = tbl.inspect.manifests()
+    assert current_rows.equals(tbl.inspect.manifests(current.snapshot_id))
+    by_spec = {row["partition_spec_id"]: row for row in current_rows.to_pylist()}
+    assert by_spec[spec.spec_id] == first_rows.to_pylist()[0]
+    bucket = str(BucketTransform(8).transform(IntegerType())(8))
+    assert by_spec[tbl.spec().spec_id]["partition_summaries"] == [
+        {"contains_null": False, "contains_nan": False, "lower_bound": bucket, "upper_bound": bucket}
+    ]
+    all_rows = tbl.inspect.all_manifests().to_pylist()
+    assert len(all_rows) == 3
+    assert [row["partition_summaries"] for row in all_rows if row["path"] == first_rows["path"][0].as_py()] == [
+        first_rows["partition_summaries"][0].as_py(),
+        first_rows["partition_summaries"][0].as_py(),
+    ]
+
+
+@pytest.mark.parametrize("missing_schema_id", [None, 999])
+def test_inspect_manifests_schema_fallback(catalog: InMemoryCatalog, missing_schema_id: int | None) -> None:
+    schema = Schema(NestedField(1, "s", StringType()))
+    spec = PartitionSpec(PartitionField(1, 1000, IdentityTransform(), "s"))
+    tbl = catalog.create_table("default.legacy_manifests", schema, partition_spec=spec)
+    tbl.append(pa.table({"s": ["legacy"]}))
+    snapshot = tbl.current_snapshot()
+    assert snapshot is not None
+    expected = tbl.inspect.manifests()
+    # Emulate legacy metadata while retaining the real local manifest and data files.
+    tbl.metadata = tbl.metadata.model_copy(update={"snapshots": [snapshot.model_copy(update={"schema_id": missing_schema_id})]})
+    if missing_schema_id is None:
+        assert tbl.inspect.manifests(snapshot.snapshot_id).equals(expected)
+        assert tbl.inspect.manifests().equals(expected)
+    else:
+        with pytest.warns(UserWarning, match=f"Metadata does not contain schema with id: {missing_schema_id}"):
+            assert tbl.inspect.manifests(snapshot.snapshot_id).equals(expected)
+
+
+@pytest.mark.parametrize(
+    ("original_type", "promoted_type", "arrow_type", "value"),
+    [(IntegerType(), LongType(), pa.int32(), 7), (FloatType(), DoubleType(), pa.float32(), 1.5)],
+)
+def test_inspect_manifests_promoted_partition_source(
+    catalog: InMemoryCatalog,
+    original_type: PrimitiveType,
+    promoted_type: PrimitiveType,
+    arrow_type: pa.DataType,
+    value: int | float,
+) -> None:
+    schema = Schema(NestedField(1, "p", original_type))
+    spec = PartitionSpec(PartitionField(1, 1000, IdentityTransform(), "p"))
+    tbl = catalog.create_table("default.promoted_manifests", schema, partition_spec=spec)
+    tbl.append(pa.table({"p": [value]}, schema=pa.schema([("p", arrow_type)])))
+    first = tbl.current_snapshot()
+    assert first is not None
+    expected = tbl.inspect.manifests()
+    with tbl.update_schema() as update:
+        update.update_column("p", field_type=promoted_type)
+    tbl.append(pa.table({"p": [value]}, schema=pa.schema([("p", pa.int64() if isinstance(value, int) else pa.float64())])))
+
+    assert tbl.inspect.manifests(first.snapshot_id).equals(expected)
+    for row in tbl.inspect.manifests().to_pylist() + tbl.inspect.all_manifests().to_pylist():
+        assert row["partition_summaries"] == expected["partition_summaries"][0].as_py()