Use update_table_metadata in InMemory Catalog (#470)

Author: anupam-saini

tests/catalog/test_base.py

@@ -52,8 +52,9 @@
     SetCurrentSchemaUpdate,
     Table,
     TableIdentifier,
+    update_table_metadata,
 )
-from pyiceberg.table.metadata import TableMetadata, TableMetadataV1, new_table_metadata
+from pyiceberg.table.metadata import TableMetadataV1
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import EMPTY_DICT
@@ -118,36 +119,13 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
         raise NotImplementedError
 
     def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
-        new_metadata: Optional[TableMetadata] = None
-        metadata_location = ""
-        for update in table_request.updates:
-            if isinstance(update, AddSchemaUpdate):
-                add_schema_update: AddSchemaUpdate = update
-                identifier = tuple(table_request.identifier.namespace.root) + (table_request.identifier.name,)
-                table = self.__tables[identifier]
-                new_metadata = new_table_metadata(
-                    add_schema_update.schema_,
-                    table.metadata.partition_specs[0],
-                    table.sort_order(),
-                    table.location(),
-                    table.properties,
-                    table.metadata.table_uuid,
-                )
-
-                table = Table(
-                    identifier=identifier,
-                    metadata=new_metadata,
-                    metadata_location=f's3://warehouse/{"/".join(identifier)}/metadata/metadata.json',
-                    io=load_file_io(),
-                    catalog=self,
-                )
-
-                self.__tables[identifier] = table
-                metadata_location = f's3://warehouse/{"/".join(identifier)}/metadata/metadata.json'
+        identifier = tuple(table_request.identifier.namespace.root) + (table_request.identifier.name,)
+        table = self.__tables[identifier]
+        table.metadata = update_table_metadata(base_metadata=table.metadata, updates=table_request.updates)
 
         return CommitTableResponse(
-            metadata=new_metadata.model_dump() if new_metadata else {},
-            metadata_location=metadata_location if metadata_location else "",
+            metadata=table.metadata.model_dump(),
+            metadata_location=table.location(),
         )
 
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
@@ -617,8 +595,9 @@ def test_commit_table(catalog: InMemoryCatalog) -> None:
 
     # Then
     assert response.metadata.table_uuid == given_table.metadata.table_uuid
-    assert len(response.metadata.schemas) == 1
-    assert response.metadata.schemas[0] == new_schema
+    assert len(response.metadata.schemas) == 2
+    assert response.metadata.schemas[1] == new_schema
+    assert response.metadata.current_schema_id == new_schema.schema_id
 
 
 def test_add_column(catalog: InMemoryCatalog) -> None: