fix: Compare timestamps for partitions metadata (#3603)

* fix: compare timestamps for partitions metadata last_updated_at

`InspectTable._update_partitions_map_from_manifest_entry` aggregates
per-partition stats over manifest entries, which are iterated in manifest
order rather than chronologically. The guard that keeps the most recently
committed snapshot per partition compared the stored `last_updated_snapshot_id`
(a snapshot id) against `snapshot.timestamp_ms` (a commit timestamp). Snapshot
ids are large positive 63-bit integers, so `id < timestamp_ms` is effectively
always false and the row keeps whichever entry was visited first instead of the
newest one, reporting the wrong `last_updated_at` / `last_updated_snapshot_id`
for any partition touched by more than one snapshot.

Compare the stored `last_updated_at` timestamp against the incoming
`snapshot.timestamp_ms`, matching the Java `PartitionsTable.Partition.update`
which uses `snapshotCommitTime > this.lastUpdatedAt`.

* test: reuse data file in partitions regression test

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

---------

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

pyiceberg/table/inspect.py

@@ -457,7 +457,7 @@ 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["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.inspect import InspectTable, _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,24 @@ 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_first", [False, True])
+def test_partitions_last_updated_uses_latest_snapshot_regardless_of_order(newest_first: bool) -> None:
+    # Manifest entries are visited in manifest order, which is not chronological, so the
+    # `partitions` metadata table must keep the snapshot with the highest commit timestamp
+    # per partition regardless of the order in which the entries are aggregated.
+    older = Snapshot(snapshot_id=6446744073709551000, timestamp_ms=1000, manifest_list="file:///dev/null")
+    newer = Snapshot(snapshot_id=8446744073709551111, timestamp_ms=5000, manifest_list="file:///dev/null")
+
+    data_file = DataFile.from_args(content=DataFileContent.DATA, record_count=1, file_size_in_bytes=1, partition=Record("a"))
+    data_file.spec_id = 0
+
+    inspect = InspectTable.__new__(InspectTable)
+    partitions_map: dict[tuple[str, Any], Any] = {}
+    for snapshot in [newer, older] if newest_first else [older, newer]:
+        inspect._update_partitions_map_from_manifest_entry(partitions_map, data_file, {"part": "a"}, snapshot)
+
+    (partition_row,) = partitions_map.values()
+    assert partition_row["last_updated_at"] == newer.timestamp_ms
+    assert partition_row["last_updated_snapshot_id"] == newer.snapshot_id