feat: add AssertMetadataLocation requirement for table metadata validation

Author: louiszhuang

pyiceberg/catalog/rest/__init__.py

@@ -83,7 +83,7 @@
 from pyiceberg.table.locations import load_location_provider
 from pyiceberg.table.metadata import TableMetadata, new_table_metadata
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder, assign_fresh_sort_order_ids
-from pyiceberg.table.update import SetTableMetadataLocationUpdate, TableRequirement, TableUpdate
+from pyiceberg.table.update import AssertMetadataLocation, SetTableMetadataLocationUpdate, TableRequirement, TableUpdate
 from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel, Identifier, Properties
 from pyiceberg.types import transform_dict_value_to_str
 from pyiceberg.utils.deprecated import deprecation_message
@@ -1203,6 +1203,7 @@ def commit_table(
                 io=updated_staged_table.io,
                 metadata_path=updated_staged_table.metadata_location,
             )
+            requirements = (AssertMetadataLocation(metadata_location=table.metadata_location), *requirements)
             updates = (SetTableMetadataLocationUpdate(metadata_location=updated_staged_table.metadata_location), *updates)
         table_identifier = TableIdentifier(namespace=identifier[:-1], name=identifier[-1])
         table_request = CommitTableRequest(identifier=table_identifier, requirements=requirements, updates=updates)

pyiceberg/table/update/__init__.py

@@ -911,6 +911,22 @@ def validate(self, base_metadata: TableMetadata | None) -> None:
             )
 
 
+class AssertMetadataLocation(ValidatableTableRequirement):
+    """The table's metadata location must match the requirement's `metadata-location`."""
+
+    type: Literal["assert-table-metadata-location"] = Field(default="assert-table-metadata-location")
+    metadata_location: str = Field(..., alias="metadata-location")
+
+    def validate(self, base_metadata: TableMetadata | None) -> None:
+        if base_metadata is None:
+            raise CommitFailedException("Requirement failed: current table metadata is missing")
+        elif self.metadata_location != base_metadata.metadata_location:
+            raise CommitFailedException(
+                f"Requirement failed: metadata location has changed: "
+                f"expected {self.metadata_location}, found {base_metadata.metadata_location}"
+            )
+
+
 TableRequirement = Annotated[
     AssertCreate
     | AssertTableUUID
@@ -919,7 +935,8 @@ def validate(self, base_metadata: TableMetadata | None) -> None:
     | AssertCurrentSchemaId
     | AssertLastAssignedPartitionId
     | AssertDefaultSpecId
-    | AssertDefaultSortOrderId,
+    | AssertDefaultSortOrderId
+    | AssertMetadataLocation,
     Field(discriminator="type"),
 ]