validate_partition_name function update

Author: rutb327

pyiceberg/partitioning.py

@@ -21,7 +21,7 @@
 from dataclasses import dataclass
 from datetime import date, datetime, time
 from functools import cached_property, singledispatch
-from typing import Annotated, Any, Dict, Generic, List, Optional, Tuple, TypeVar, Union
+from typing import Annotated, Any, Dict, Generic, List, Optional, Set, Tuple, TypeVar, Union
 from urllib.parse import quote_plus
 
 from pydantic import (
@@ -254,6 +254,7 @@ def validate_partition_name(
     partition_transform: Transform[Any, Any],
     source_id: int,
     schema: Schema,
+    partition_names: Set[str],
 ) -> None:
     """Validate that a partition field name doesn't conflict with schema field names."""
     try:
@@ -262,11 +263,15 @@ def validate_partition_name(
         return  # No conflict if field doesn't exist in schema
 
     if isinstance(partition_transform, (IdentityTransform, VoidTransform)):
-        # For identity transforms, allow conflict only if sourced from the same schema field
+        # For identity and void transforms, allow conflict only if sourced from the same schema field
         if schema_field.field_id != source_id:
-            raise ValueError(f"Cannot create identity partition from a different source field in the schema: {field_name}")
+            raise ValueError(f"Cannot create identity partition sourced from different field in schema: {field_name}")
     else:
         raise ValueError(f"Cannot create partition from name that exists in schema: {field_name}")
+    if not field_name:
+        raise ValueError("Undefined name")
+    if field_name in partition_names:
+        raise ValueError(f"Partition name has to be unique: {field_name}")
 
 
 def assign_fresh_partition_spec_ids(spec: PartitionSpec, old_schema: Schema, fresh_schema: Schema) -> PartitionSpec:
@@ -279,7 +284,7 @@ def assign_fresh_partition_spec_ids(spec: PartitionSpec, old_schema: Schema, fre
         if fresh_field is None:
             raise ValueError(f"Could not find field in fresh schema: {original_column_name}")
 
-        validate_partition_name(field.name, field.transform, fresh_field.field_id, fresh_schema)
+        validate_partition_name(field.name, field.transform, fresh_field.field_id, fresh_schema, set())
 
         partition_fields.append(
             PartitionField(

pyiceberg/table/update/schema.py

@@ -658,14 +658,13 @@ def _apply(self) -> Schema:
 
         # Check the field-ids
         new_schema = Schema(*struct.fields)
-        if self._transaction is not None:
-            from pyiceberg.partitioning import validate_partition_name
+        from pyiceberg.partitioning import validate_partition_name
 
-            for spec in self._transaction.table_metadata.partition_specs:
-                for partition_field in spec.fields:
-                    validate_partition_name(
-                        partition_field.name, partition_field.transform, partition_field.source_id, new_schema
-                    )
+        for spec in self._transaction.table_metadata.partition_specs:
+            for partition_field in spec.fields:
+                validate_partition_name(
+                    partition_field.name, partition_field.transform, partition_field.source_id, new_schema, set()
+                )
         field_ids = set()
         for name in self._identifier_field_names:
             try:

pyiceberg/table/update/spec.py

@@ -179,11 +179,7 @@ def _check_and_add_partition_name(
         ) -> None:
             from pyiceberg.partitioning import validate_partition_name
 
-            validate_partition_name(name, transform, source_id, schema)
-            if not name:
-                raise ValueError("Undefined name")
-            if name in partition_names:
-                raise ValueError(f"Partition name has to be unique: {name}")
+            validate_partition_name(name, transform, source_id, schema, partition_names)
             partition_names.add(name)
 
         def _add_new_field(

tests/integration/test_partition_evolution.py

@@ -15,13 +15,12 @@
 # specific language governing permissions and limitations
 # under the License.
 # pylint:disable=redefined-outer-name
-from typing import Optional
 
 import pytest
 
 from pyiceberg.catalog import Catalog
 from pyiceberg.exceptions import NoSuchTableError
-from pyiceberg.partitioning import PartitionField, PartitionSpec
+from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table import Table
 from pyiceberg.transforms import (
@@ -65,7 +64,7 @@ def _table_v2(catalog: Catalog) -> Table:
 
 
 def _create_table_with_schema(
-    catalog: Catalog, schema: Schema, format_version: str, partition_spec: Optional[PartitionSpec] = None
+    catalog: Catalog, schema: Schema, format_version: str, partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC
 ) -> Table:
     tbl_name = "default.test_schema_evolution"
     try:
@@ -77,7 +76,9 @@ def _create_table_with_schema(
         return catalog.create_table(
             identifier=tbl_name, schema=schema, partition_spec=partition_spec, properties={"format-version": format_version}
         )
-    return catalog.create_table(identifier=tbl_name, schema=schema, properties={"format-version": format_version})
+    return catalog.create_table(
+        identifier=tbl_name, schema=schema, partition_spec=partition_spec, properties={"format-version": format_version}
+    )
 
 
 @pytest.mark.integration
@@ -590,11 +591,9 @@ def test_partition_schema_field_name_conflict(catalog: Catalog) -> None:
     with pytest.raises(ValueError, match="Cannot create partition from name that exists in schema: id"):
         table.update_spec().add_field("event_ts", DayTransform(), "id").commit()
 
-    with pytest.raises(
-        ValueError, match="Cannot create identity partition from a different source field in the schema: another_ts"
-    ):
+    with pytest.raises(ValueError, match="Cannot create identity partition sourced from different field in schema: another_ts"):
         table.update_spec().add_field("event_ts", IdentityTransform(), "another_ts").commit()
-    with pytest.raises(ValueError, match="Cannot create identity partition from a different source field in the schema: str"):
+    with pytest.raises(ValueError, match="Cannot create identity partition sourced from different field in schema: str"):
         table.update_spec().add_field("id", IdentityTransform(), "str").commit()
 
     table.update_spec().add_field("id", IdentityTransform(), "id").commit()
@@ -640,18 +639,14 @@ def test_schema_evolution_partition_conflict(catalog: Catalog) -> None:
 
     with pytest.raises(ValueError, match="Cannot create partition from name that exists in schema: event_year"):
         table.update_schema().add_column("event_year", StringType()).commit()
-    with pytest.raises(
-        ValueError, match="Cannot create identity partition from a different source field in the schema: first_name"
-    ):
+    with pytest.raises(ValueError, match="Cannot create identity partition sourced from different field in schema: first_name"):
         table.update_schema().add_column("first_name", StringType()).commit()
 
     table.update_schema().add_column("other_field", StringType()).commit()
 
     with pytest.raises(ValueError, match="Cannot create partition from name that exists in schema: event_year"):
         table.update_schema().rename_column("other_field", "event_year").commit()
-    with pytest.raises(
-        ValueError, match="Cannot create identity partition from a different source field in the schema: first_name"
-    ):
+    with pytest.raises(ValueError, match="Cannot create identity partition sourced from different field in schema: first_name"):
         table.update_schema().rename_column("other_field", "first_name").commit()
 
     table.update_schema().rename_column("other_field", "valid_name").commit()