fix(table): validate snapshot-log timestamps in metadata parsing

Author: mrutunjay-kinagi

pyiceberg/table/metadata.py

@@ -67,6 +67,7 @@
 DEFAULT_SCHEMA_ID = 0
 
 SUPPORTED_TABLE_FORMAT_VERSION = 2
+ONE_MINUTE_MS = 60_000
 
 
 def cleanup_snapshot_id(data: dict[str, Any]) -> dict[str, Any]:
@@ -125,6 +126,29 @@ def construct_refs(table_metadata: TableMetadata) -> TableMetadata:
     return table_metadata
 
 
+def check_snapshot_timestamps(table_metadata: TableMetadata) -> TableMetadata:
+    """Validate snapshot and snapshot-log timestamps with small clock skew tolerance."""
+    last_snapshot_log_entry: SnapshotLogEntry | None = None
+    for snapshot_log_entry in table_metadata.snapshot_log:
+        if (
+            last_snapshot_log_entry is not None
+            and snapshot_log_entry.timestamp_ms - last_snapshot_log_entry.timestamp_ms < -ONE_MINUTE_MS
+        ):
+            raise ValidationError("[BUG] Expected sorted snapshot log entries.")
+        last_snapshot_log_entry = snapshot_log_entry
+
+    if (
+        last_snapshot_log_entry is not None
+        and table_metadata.last_updated_ms - last_snapshot_log_entry.timestamp_ms < -ONE_MINUTE_MS
+    ):
+        raise ValidationError(
+            f"Invalid update timestamp {table_metadata.last_updated_ms}: "
+            f"before last snapshot log entry at {last_snapshot_log_entry.timestamp_ms}"
+        )
+
+    return table_metadata
+
+
 class TableMetadataCommonFields(IcebergBaseModel):
     """Metadata for an Iceberg table as specified in the Apache Iceberg spec.
 
@@ -378,6 +402,10 @@ def cleanup_snapshot_id(cls, data: dict[str, Any]) -> dict[str, Any]:
     def construct_refs(self) -> TableMetadataV1:
         return construct_refs(self)
 
+    @model_validator(mode="after")
+    def check_snapshot_timestamps(self) -> TableMetadata:
+        return check_snapshot_timestamps(self)
+
     @model_validator(mode="before")
     def set_v2_compatible_defaults(cls, data: dict[str, Any]) -> dict[str, Any]:
         """Set default values to be compatible with the format v2.
@@ -519,6 +547,10 @@ def check_sort_orders(self) -> TableMetadata:
     def construct_refs(self) -> TableMetadata:
         return construct_refs(self)
 
+    @model_validator(mode="after")
+    def check_snapshot_timestamps(self) -> TableMetadata:
+        return check_snapshot_timestamps(self)
+
     format_version: Literal[2] = Field(alias="format-version", default=2)
     """An integer version number for the format. Implementations must throw
     an exception if a table’s version is higher than the supported version."""
@@ -563,6 +595,10 @@ def check_sort_orders(self) -> TableMetadata:
     def construct_refs(self) -> TableMetadata:
         return construct_refs(self)
 
+    @model_validator(mode="after")
+    def check_snapshot_timestamps(self) -> TableMetadata:
+        return check_snapshot_timestamps(self)
+
     format_version: Literal[3] = Field(alias="format-version", default=3)
     """An integer version number for the format. Implementations must throw
     an exception if a table’s version is higher than the supported version."""

tests/table/test_metadata.py

@@ -18,7 +18,7 @@
 
 import io
 import json
-from copy import copy
+from copy import copy, deepcopy
 from typing import Any
 from unittest.mock import MagicMock, patch
 from uuid import UUID
@@ -415,6 +415,26 @@ def test_sort_order_unsorted() -> None:
     assert len(table_metadata.sort_orders) == 0
 
 
+def test_snapshot_log_entries_are_sorted_with_tolerance(example_table_metadata_v2: dict[str, Any]) -> None:
+    table_metadata = deepcopy(example_table_metadata_v2)
+    table_metadata["snapshot-log"][1]["timestamp-ms"] = table_metadata["snapshot-log"][0]["timestamp-ms"] - 60_001
+
+    with pytest.raises(ValidationError) as exc_info:
+        TableMetadataUtil.parse_raw(json.dumps(table_metadata))
+
+    assert "Expected sorted snapshot log entries" in str(exc_info.value)
+
+
+def test_last_updated_ms_not_before_last_snapshot_log_entry(example_table_metadata_v2: dict[str, Any]) -> None:
+    table_metadata = deepcopy(example_table_metadata_v2)
+    table_metadata["last-updated-ms"] = table_metadata["snapshot-log"][-1]["timestamp-ms"] - 60_001
+
+    with pytest.raises(ValidationError) as exc_info:
+        TableMetadataUtil.parse_raw(json.dumps(table_metadata))
+
+    assert "before last snapshot log entry at" in str(exc_info.value)
+
+
 def test_invalid_partition_spec() -> None:
     table_metadata_spec_not_found = {
         "format-version": 2,