diff --git a/pyiceberg/table/update/__init__.py b/pyiceberg/table/update/__init__.py index aa1a09ebb2..c997363599 100644 --- a/pyiceberg/table/update/__init__.py +++ b/pyiceberg/table/update/__init__.py @@ -27,7 +27,7 @@ from pydantic import Field, field_validator, model_serializer, model_validator from pyiceberg.exceptions import CommitFailedException -from pyiceberg.partitioning import PARTITION_FIELD_ID_START, PartitionSpec +from pyiceberg.partitioning import PARTITION_FIELD_ID_START, PartitionSpec, validate_partition_name from pyiceberg.schema import Schema from pyiceberg.table.metadata import ( SUPPORTED_TABLE_FORMAT_VERSION, @@ -408,6 +408,11 @@ def _(update: AddPartitionSpecUpdate, base_metadata: TableMetadata, context: _Ta if spec.spec_id == update.spec.spec_id and spec != update.spec: raise ValueError(f"Partition spec with id {spec.spec_id} already exists: {spec}") + # The spec was built against the schema it was loaded with, which another writer may have changed since + schema = base_metadata.schema() + for field in update.spec.fields: + validate_partition_name(field.name, field.transform, field.source_id, schema, set()) + metadata_updates: dict[str, Any] = { "partition_specs": base_metadata.partition_specs + [update.spec], "last_partition_id": max( diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py index 1148ca8527..26bd9f9e41 100644 --- a/tests/catalog/test_catalog_behaviors.py +++ b/tests/catalog/test_catalog_behaviors.py @@ -46,7 +46,7 @@ from pyiceberg.table.snapshots import Operation from pyiceberg.table.sorting import NullOrder, SortDirection, SortField, SortOrder from pyiceberg.table.update import AddSchemaUpdate, SetCurrentSchemaUpdate -from pyiceberg.transforms import IdentityTransform +from pyiceberg.transforms import BucketTransform, IdentityTransform from pyiceberg.typedef import Identifier from pyiceberg.types import BooleanType, DoubleType, IntegerType, LongType, NestedField, StringType @@ -622,6 +622,25 @@ def test_concurrent_commit_table(catalog: Catalog, table_schema_nested: Schema, update.add_column(path="c", field_type=IntegerType()) +def test_concurrent_update_spec_checks_names_against_current_schema(catalog: Catalog, test_table_identifier: Identifier) -> None: + namespace = Catalog.namespace_from(test_table_identifier) + catalog.create_namespace(namespace) + table_a = catalog.create_table(test_table_identifier, Schema(NestedField(1, "x", LongType(), required=False))) + table_b = catalog.load_table(test_table_identifier) + + with table_a.update_schema() as update: + update.add_column(path="xb", field_type=LongType()) + + # table_b still has the schema without xb, so only the commit can see the conflict + with pytest.raises(ValueError, match="Cannot create partition with a name that exists in schema: xb"): + with table_b.update_spec() as update: + update.add_field("x", BucketTransform(4), "xb") + + table = catalog.load_table(test_table_identifier) + assert table.schema().column_names == ["x", "xb"] + assert table.spec().is_unpartitioned() + + 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)