From 450e78ce41f7c6fcb4397628276229545c2bc8ce Mon Sep 17 00:00:00 2001
From: Yuang Gao <yuang@apache.org>
Date: Mon, 2 Nov 2026 19:45:05 -0800
Subject: [PATCH] fix: keep last_updated_ms from preceding the latest
 metadata-log entry

---
 pyiceberg/table/update/__init__.py |  4 ++--
 tests/table/test_init.py           | 21 +++++++++++++++++++--
 2 files changed, 21 insertions(+), 4 deletions(-)

diff --git a/pyiceberg/table/update/__init__.py b/pyiceberg/table/update/__init__.py
index aa1a09ebb2..e48e969702 100644
--- a/pyiceberg/table/update/__init__.py
+++ b/pyiceberg/table/update/__init__.py
@@ -746,11 +746,11 @@ def update_table_metadata(
     for update in updates:
         new_metadata = _apply_table_update(update, new_metadata, context)
 
-    # Update last_updated_ms if it was not updated by update operations
+    # last_updated_ms must not precede the base's, which becomes the latest metadata-log entry
     if context.has_changes():
         if metadata_location:
             new_metadata = _update_table_metadata_log(new_metadata, metadata_location, base_metadata.last_updated_ms)
-        if base_metadata.last_updated_ms == new_metadata.last_updated_ms:
+        if new_metadata.last_updated_ms <= base_metadata.last_updated_ms:
             new_metadata = new_metadata.model_copy(update={"last_updated_ms": datetime_to_millis(datetime.now().astimezone())})
 
     # Check correctness of partition spec, and sort order
diff --git a/tests/table/test_init.py b/tests/table/test_init.py
index ff7ecd7c37..48040d339b 100644
--- a/tests/table/test_init.py
+++ b/tests/table/test_init.py
@@ -1252,7 +1252,7 @@ def test_update_metadata_with_multiple_updates(table_v1: Table) -> None:
         snapshot_id=25,
         parent_snapshot_id=19,
         sequence_number=200,
-        timestamp_ms=1602638573590,
+        timestamp_ms=1602638593590,
         manifest_list="s3:/a/b/c.avro",
         summary=Summary(Operation.APPEND),
         schema_id=3,
@@ -1295,7 +1295,7 @@ def test_update_metadata_with_multiple_updates(table_v1: Table) -> None:
     assert len(new_metadata.snapshot_log) == 1
     assert new_metadata.snapshot_log[0].snapshot_id == 25
     assert new_metadata.current_snapshot_id == 25
-    assert new_metadata.last_updated_ms == 1602638573590
+    assert new_metadata.last_updated_ms == 1602638593590
     assert new_metadata.refs["main"] == SnapshotRef(
         snapshot_id=25,
         snapshot_ref_type="branch",
@@ -1683,6 +1683,23 @@ def test_update_metadata_log(table_v2: Table) -> None:
     assert len(new_metadata.metadata_log) == 2
 
 
+def test_update_metadata_last_updated_ms_not_before_metadata_log(table_v2: Table) -> None:
+    stale_snapshot = Snapshot(
+        snapshot_id=25,
+        parent_snapshot_id=19,
+        sequence_number=200,
+        timestamp_ms=table_v2.metadata.last_updated_ms - 61_000,
+        manifest_list="s3:/a/b/c.avro",
+        summary=Summary(Operation.APPEND),
+        schema_id=3,
+    )
+
+    new_metadata = update_table_metadata(
+        table_v2.metadata, (AddSnapshotUpdate(snapshot=stale_snapshot),), False, table_v2.metadata_location
+    )
+    assert new_metadata.last_updated_ms > new_metadata.metadata_log[-1].timestamp_ms
+
+
 def test_update_metadata_log_overflow(table_v2: Table) -> None:
     metadata_log = [
         MetadataLogEntry(