Skip to content

Commit 4a89fde

Browse files
committed
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`.
1 parent 9d36e23 commit 4a89fde

2 files changed

Lines changed: 29 additions & 2 deletions

File tree

pyiceberg/table/inspect.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -352,7 +352,7 @@ def _update_partitions_map_from_manifest_entry(
352352
partition_row = partitions_map[partition_record_key]
353353

354354
if snapshot is not None:
355-
if partition_row["last_updated_at"] is None or partition_row["last_updated_snapshot_id"] < snapshot.timestamp_ms:
355+
if partition_row["last_updated_at"] is None or partition_row["last_updated_at"] < snapshot.timestamp_ms:
356356
partition_row["last_updated_at"] = snapshot.timestamp_ms
357357
partition_row["last_updated_snapshot_id"] = snapshot.snapshot_id
358358

tests/table/test_inspect.py

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,17 @@
1616
# under the License.
1717

1818
from pathlib import PosixPath
19+
from typing import Any
1920

2021
import pyarrow as pa
2122
import pytest
2223

2324
from pyiceberg.conversions import to_bytes
25+
from pyiceberg.manifest import DataFile, DataFileContent
2426
from pyiceberg.schema import Schema
25-
from pyiceberg.table.inspect import _readable_bound
27+
from pyiceberg.table.inspect import InspectTable, _readable_bound
28+
from pyiceberg.table.snapshots import Snapshot
29+
from pyiceberg.typedef import Record
2630
from pyiceberg.types import NestedField, StringType
2731
from tests.catalog.test_base import InMemoryCatalog
2832

@@ -68,3 +72,26 @@ def test_inspect_entries_and_files_render_null_bound(catalog: InMemoryCatalog) -
6872
files_metrics = tbl.inspect.files().to_pydict()["readable_metrics"][0]["s"]
6973
assert files_metrics["lower_bound"] is None
7074
assert files_metrics["upper_bound"] is None
75+
76+
77+
@pytest.mark.parametrize("newest_first", [False, True])
78+
def test_partitions_last_updated_uses_latest_snapshot_regardless_of_order(newest_first: bool) -> None:
79+
# Manifest entries are visited in manifest order, which is not chronological, so the
80+
# `partitions` metadata table must keep the snapshot with the highest commit timestamp
81+
# per partition regardless of the order in which the entries are aggregated.
82+
older = Snapshot(snapshot_id=6446744073709551000, timestamp_ms=1000, manifest_list="file:///dev/null")
83+
newer = Snapshot(snapshot_id=8446744073709551111, timestamp_ms=5000, manifest_list="file:///dev/null")
84+
85+
def _data_file() -> DataFile:
86+
file = DataFile.from_args(content=DataFileContent.DATA, record_count=1, file_size_in_bytes=1, partition=Record("a"))
87+
file.spec_id = 0
88+
return file
89+
90+
inspect = InspectTable.__new__(InspectTable)
91+
partitions_map: dict[tuple[str, Any], Any] = {}
92+
for snapshot in [newer, older] if newest_first else [older, newer]:
93+
inspect._update_partitions_map_from_manifest_entry(partitions_map, _data_file(), {"part": "a"}, snapshot)
94+
95+
(partition_row,) = partitions_map.values()
96+
assert partition_row["last_updated_at"] == newer.timestamp_ms
97+
assert partition_row["last_updated_snapshot_id"] == newer.snapshot_id

0 commit comments

Comments
 (0)