diff --git a/pyiceberg/table/update/schema.py b/pyiceberg/table/update/schema.py index 828f1e877a..f04513bb12 100644 --- a/pyiceberg/table/update/schema.py +++ b/pyiceberg/table/update/schema.py @@ -42,6 +42,7 @@ from pyiceberg.table.update import ( AddSchemaUpdate, AssertCurrentSchemaId, + AssertLastAssignedFieldId, SetCurrentSchemaUpdate, SetPropertiesUpdate, TableRequirement, @@ -635,6 +636,8 @@ def _commit(self) -> UpdatesAndRequirements: if existing_schema_id != self._schema.schema_id: requirements += (AssertCurrentSchemaId(current_schema_id=self._schema.schema_id),) if existing_schema_id is None: + last_column_id = self._transaction.table_metadata.last_column_id + requirements += (AssertLastAssignedFieldId(last_assigned_field_id=last_column_id),) updates += ( AddSchemaUpdate(schema=new_schema), SetCurrentSchemaUpdate(schema_id=-1), diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py index 1148ca8527..9540bfb08f 100644 --- a/tests/catalog/test_catalog_behaviors.py +++ b/tests/catalog/test_catalog_behaviors.py @@ -622,6 +622,34 @@ def test_concurrent_commit_table(catalog: Catalog, table_schema_nested: Schema, update.add_column(path="c", field_type=IntegerType()) +def test_concurrent_add_column_after_add_and_drop( + catalog: Catalog, table_schema_simple: Schema, random_table_identifier: Identifier +) -> None: + namespace = Catalog.namespace_from(random_table_identifier) + catalog.create_namespace(namespace) + table_a = catalog.create_table(random_table_identifier, table_schema_simple) + table_b = catalog.load_table(random_table_identifier) + + with table_a.update_schema() as update: + update.add_column(path="b", field_type=IntegerType()) + with table_a.update_schema() as update: + update.delete_column("b") + + # Dropping b makes schema 0 current again, so only the last assigned field id shows the change: + # table_b loaded it as 3 (foo, bar, baz), adding b raised it to 4 + with pytest.raises( + CommitFailedException, match="Requirement failed: last assigned field id has changed: expected 3, found 4" + ): + with table_b.update_schema() as update: + update.add_column(path="c", field_type=IntegerType()) + + # After refresh the last assigned field id is 4 (b), so c gets 5 + table_b.refresh() + with table_b.update_schema() as update: + update.add_column(path="c", field_type=IntegerType()) + assert table_b.schema().find_field("c").field_id == 5 + + def test_delete_metadata_multiple(catalog: Catalog, table_schema_nested: Schema, random_table_identifier: Identifier) -> None: namespace = Catalog.namespace_from(random_table_identifier) catalog.create_namespace(namespace)