From c8f9ed85440ee9bf08f7d75c8b7ccaadae9c4ccf Mon Sep 17 00:00:00 2001 From: Rahul-cs50 Date: Mon, 2 Nov 2026 23:02:31 +0530 Subject: [PATCH] Check partition names against the current schema when adding a spec UpdateSpec checks new partition names against the schema it was loaded with. If another writer added a column with the same name in between, the commit still went through and wrote metadata Java refuses to load. Check the names again when the spec is applied to the current metadata. Closes #4198 --- pyiceberg/table/update/__init__.py | 7 ++++++- tests/catalog/test_catalog_behaviors.py | 21 ++++++++++++++++++++- 2 files changed, 26 insertions(+), 2 deletions(-) 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)