Core: Select inspect.partitions snapshot metadata by commit time

pyiceberg/table/inspect.py

@@ -352,7 +352,8 @@ def _update_partitions_map_from_manifest_entry(
         partition_row = partitions_map[partition_record_key]
 
         if snapshot is not None:
-            if partition_row["last_updated_at"] is None or partition_row["last_updated_snapshot_id"] < snapshot.timestamp_ms:
+            if partition_row["last_updated_at"] is None or partition_row["last_updated_at"] < snapshot.timestamp_ms:
+                partition_row["spec_id"] = file.spec_id
                 partition_row["last_updated_at"] = snapshot.timestamp_ms
                 partition_row["last_updated_snapshot_id"] = snapshot.snapshot_id
 

tests/table/test_inspect.py

@@ -16,13 +16,17 @@
 # under the License.
 
 from pathlib import PosixPath
+from typing import Any
 
 import pyarrow as pa
 import pytest
 
 from pyiceberg.conversions import to_bytes
+from pyiceberg.manifest import DataFile, DataFileContent
 from pyiceberg.schema import Schema
 from pyiceberg.table.inspect import _readable_bound
+from pyiceberg.table.snapshots import Snapshot
+from pyiceberg.typedef import Record
 from pyiceberg.types import NestedField, StringType
 from tests.catalog.test_base import InMemoryCatalog
 
@@ -68,3 +72,42 @@ def test_inspect_entries_and_files_render_null_bound(catalog: InMemoryCatalog) -
     files_metrics = tbl.inspect.files().to_pydict()["readable_metrics"][0]["s"]
     assert files_metrics["lower_bound"] is None
     assert files_metrics["upper_bound"] is None
+
+
+@pytest.mark.parametrize("newest_entry_first", [False, True])
+def test_partitions_uses_latest_snapshot_metadata(catalog: InMemoryCatalog, newest_entry_first: bool) -> None:
+    table = catalog.create_table("default.snapshot_info", Schema(NestedField(1, "s", StringType(), required=False)))
+
+    def data_file(spec_id: int) -> DataFile:
+        file = DataFile.from_args(
+            content=DataFileContent.DATA,
+            partition=Record("a"),
+            record_count=1,
+            file_size_in_bytes=1,
+        )
+        file.spec_id = spec_id
+        return file
+
+    older_snapshot = Snapshot(
+        snapshot_id=9_000_000_000_000_000,
+        timestamp_ms=1_000,
+        manifest_list="older.avro",
+    )
+    newer_snapshot = Snapshot(
+        snapshot_id=8_000_000_000_000_000,
+        timestamp_ms=2_000,
+        manifest_list="newer.avro",
+    )
+    partitions_map: dict[tuple[str, Any], Any] = {}
+
+    updates = [(data_file(spec_id=1), older_snapshot), (data_file(spec_id=2), newer_snapshot)]
+    if newest_entry_first:
+        updates.reverse()
+
+    for file, snapshot in updates:
+        table.inspect._update_partitions_map_from_manifest_entry(partitions_map, file, {"part": "a"}, snapshot)
+
+    partition_row = next(iter(partitions_map.values()))
+    assert partition_row["spec_id"] == 2
+    assert partition_row["last_updated_at"] == newer_snapshot.timestamp_ms
+    assert partition_row["last_updated_snapshot_id"] == newer_snapshot.snapshot_id