Check partition names against the current schema when adding a spec

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(

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)