diff --git a/pyiceberg/table/metadata.py b/pyiceberg/table/metadata.py
index 26b6e3d3ad..33230215b1 100644
--- a/pyiceberg/table/metadata.py
+++ b/pyiceberg/table/metadata.py
@@ -68,6 +68,10 @@
 
 SUPPORTED_TABLE_FORMAT_VERSION = 2
 
+# Tolerance for clock drift when validating that last-updated-ms is not
+# before the most recent snapshot log entry. Matches the Java implementation.
+ONE_MINUTE_MS = 60_000
+
 
 def cleanup_snapshot_id(data: dict[str, Any]) -> dict[str, Any]:
     """Run before validation."""
@@ -115,6 +119,22 @@ def check_sort_orders(table_metadata: TableMetadata) -> TableMetadata:
     return table_metadata
 
 
+def check_last_updated_ms(table_metadata: TableMetadata) -> TableMetadata:
+    """Check that last-updated-ms is not before the most recent snapshot log entry.
+
+    A one minute tolerance is allowed because commits can happen concurrently
+    from different machines with small clock skew.
+    """
+    if snapshot_log := table_metadata.snapshot_log:
+        last_entry = snapshot_log[-1]
+        if table_metadata.last_updated_ms - last_entry.timestamp_ms < -ONE_MINUTE_MS:
+            raise ValidationError(
+                f"Invalid last updated timestamp {table_metadata.last_updated_ms}: "
+                f"before last snapshot log entry at {last_entry.timestamp_ms}"
+            )
+    return table_metadata
+
+
 def construct_refs(table_metadata: TableMetadata) -> TableMetadata:
     """Set the main branch if missing."""
     if table_metadata.current_snapshot_id is not None:
@@ -378,6 +398,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_last_updated_ms(self) -> TableMetadataV1:
+        return check_last_updated_ms(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 +543,10 @@ def check_sort_orders(self) -> TableMetadata:
     def construct_refs(self) -> TableMetadata:
         return construct_refs(self)
 
+    @model_validator(mode="after")
+    def check_last_updated_ms(self) -> TableMetadata:
+        return check_last_updated_ms(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 +591,10 @@ def check_sort_orders(self) -> TableMetadata:
     def construct_refs(self) -> TableMetadata:
         return construct_refs(self)
 
+    @model_validator(mode="after")
+    def check_last_updated_ms(self) -> TableMetadata:
+        return check_last_updated_ms(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."""
diff --git a/pyiceberg/table/update/__init__.py b/pyiceberg/table/update/__init__.py
index 64838b0bd6..fdb026428f 100644
--- a/pyiceberg/table/update/__init__.py
+++ b/pyiceberg/table/update/__init__.py
@@ -59,6 +59,10 @@
 
 U = TypeVar("U")
 
+# Tolerance for clock drift when validating that a new snapshot is not
+# recorded before the table's last update. Matches the Java and Rust implementations.
+ONE_MINUTE_MS = 60_000
+
 
 class UpdateTableMetadata(ABC, Generic[U]):
     _transaction: Transaction
@@ -444,6 +448,14 @@ def _(update: AddSnapshotUpdate, base_metadata: TableMetadata, context: _TableMe
             f"Cannot add snapshot with sequence number {update.snapshot.sequence_number} "
             f"older than last sequence number {base_metadata.last_sequence_number}"
         )
+    elif (
+        base_metadata.last_updated_ms is not None
+        and update.snapshot.timestamp_ms - base_metadata.last_updated_ms < -ONE_MINUTE_MS
+    ):
+        raise ValueError(
+            f"Invalid snapshot timestamp {update.snapshot.timestamp_ms}: "
+            f"before last updated timestamp {base_metadata.last_updated_ms}"
+        )
     elif base_metadata.format_version >= 3 and update.snapshot.first_row_id is None:
         raise ValueError("Cannot add snapshot without first row id")
     elif (
diff --git a/tests/conftest.py b/tests/conftest.py
index eb0b712d1e..df88574496 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -819,7 +819,7 @@ def generate_snapshot(
         "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
         "location": "s3://bucket/test/location",
         "last-sequence-number": 34,
-        "last-updated-ms": 1602638573590,
+        "last-updated-ms": max(entry["timestamp-ms"] for entry in snapshot_log),
         "last-column-id": 3,
         "current-schema-id": 1,
         "schemas": [
diff --git a/tests/table/test_init.py b/tests/table/test_init.py
index 1670e62587..585cdd8064 100644
--- a/tests/table/test_init.py
+++ b/tests/table/test_init.py
@@ -1053,6 +1053,36 @@ def test_update_metadata_add_snapshot(table_v2: Table) -> None:
     assert new_metadata.last_updated_ms == new_snapshot.timestamp_ms
 
 
+def test_update_metadata_add_snapshot_before_last_updated(table_v2: Table) -> None:
+    new_snapshot = Snapshot(
+        snapshot_id=25,
+        parent_snapshot_id=19,
+        sequence_number=200,
+        timestamp_ms=table_v2.metadata.last_updated_ms - 60_001,
+        manifest_list="s3:/a/b/c.avro",
+        summary=Summary(Operation.APPEND),
+        schema_id=3,
+    )
+
+    with pytest.raises(ValueError, match="Invalid snapshot timestamp"):
+        update_table_metadata(table_v2.metadata, (AddSnapshotUpdate(snapshot=new_snapshot),))
+
+
+def test_update_metadata_add_snapshot_within_clock_drift(table_v2: Table) -> None:
+    new_snapshot = Snapshot(
+        snapshot_id=25,
+        parent_snapshot_id=19,
+        sequence_number=200,
+        timestamp_ms=table_v2.metadata.last_updated_ms - 60_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=new_snapshot),))
+    assert new_metadata.snapshots[-1] == new_snapshot
+
+
 def test_update_metadata_set_ref_snapshot(table_v2: Table) -> None:
     update, _ = table_v2.transaction()._set_ref_snapshot(
         snapshot_id=3051729675574597004,
diff --git a/tests/table/test_metadata.py b/tests/table/test_metadata.py
index c163c90626..c4dfad5aa6 100644
--- a/tests/table/test_metadata.py
+++ b/tests/table/test_metadata.py
@@ -145,6 +145,24 @@ def test_parsing_correct_types(example_table_metadata_v2: dict[str, Any]) -> Non
     assert isinstance(table_metadata.schemas[0].fields[0].field_type, LongType)
 
 
+def test_last_updated_ms_before_snapshot_log(example_table_metadata_v2: dict[str, Any]) -> None:
+    metadata = copy(example_table_metadata_v2)
+    last_entry_ms = metadata["snapshot-log"][-1]["timestamp-ms"]
+    metadata["last-updated-ms"] = last_entry_ms - 60_001
+
+    with pytest.raises(ValidationError, match="Invalid last updated timestamp"):
+        TableMetadataV2(**metadata)
+
+
+def test_last_updated_ms_within_clock_drift(example_table_metadata_v2: dict[str, Any]) -> None:
+    metadata = copy(example_table_metadata_v2)
+    last_entry_ms = metadata["snapshot-log"][-1]["timestamp-ms"]
+    metadata["last-updated-ms"] = last_entry_ms - 60_000
+
+    table_metadata = TableMetadataV2(**metadata)
+    assert table_metadata.last_updated_ms == last_entry_ms - 60_000
+
+
 def test_updating_metadata(example_table_metadata_v2: dict[str, Any]) -> None:
     """Test creating a new TableMetadata instance that's an updated version of
     an existing TableMetadata instance"""