Reject duplicate partition field names

pyiceberg/partitioning.py

@@ -300,6 +300,11 @@ def validate_partition_name(
     partition_names: set[str],
 ) -> None:
     """Validate that a partition field name doesn't conflict with schema field names."""
+    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}")
+
     try:
         schema_field = schema.find_field(field_name)
     except ValueError:
@@ -311,14 +316,11 @@ def validate_partition_name(
             raise ValueError(f"Cannot create identity partition sourced from different field in schema: {field_name}")
     else:
         raise ValueError(f"Cannot create partition with a 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:
     partition_fields = []
+    partition_names: set[str] = set()
     for pos, field in enumerate(spec.fields):
         original_column_name = old_schema.find_column_name(field.source_id)
         if original_column_name is None:
@@ -327,7 +329,8 @@ 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, set())
+        validate_partition_name(field.name, field.transform, fresh_field.field_id, fresh_schema, partition_names)
+        partition_names.add(field.name)
 
         partition_fields.append(
             PartitionField(

tests/table/test_init.py

@@ -1522,6 +1522,22 @@ def test_last_assigned_partition_id(table_v2: Table) -> None:
         AssertLastAssignedPartitionId(last_assigned_partition_id=1).validate(base_metadata)
 
 
+def test_update_spec_rejects_duplicate_partition_name(table_v2: Table) -> None:
+    txn = table_v2.transaction()
+    txn.update_spec().add_field("y", BucketTransform(16), "y_bucket").commit()
+
+    with pytest.raises(ValueError, match="Partition name has to be unique: y_bucket"):
+        txn.update_spec().rename_field("x", "y_bucket").commit()
+
+    with pytest.raises(ValueError, match="Partition name has to be unique: z_bucket"):
+        txn.update_spec().rename_field("y_bucket", "z_bucket").add_field("z", BucketTransform(16), "z_bucket").commit()
+
+
+def test_update_spec_rejects_empty_partition_name(table_v2: Table) -> None:
+    with pytest.raises(ValueError, match="Undefined name"):
+        table_v2.transaction().update_spec().add_field("z", BucketTransform(16), "").commit()
+
+
 @pytest.mark.parametrize("table", [lf("table_v2"), lf("table_v3")])
 def test_update_spec_reuses_removed_partition_field_id(table: Table) -> None:
     txn = table.transaction()

tests/table/test_metadata.py

@@ -880,6 +880,27 @@ def test_new_table_metadata_format_v2_with_v3_schema_fails(field_type: Primitive
         )
 
 
+def test_new_table_metadata_with_duplicate_partition_name_fails() -> None:
+    schema = Schema(
+        NestedField(field_id=10, name="foo", field_type=StringType(), required=False),
+        NestedField(field_id=22, name="bar", field_type=IntegerType(), required=False),
+    )
+
+    partition_spec = PartitionSpec(
+        PartitionField(source_id=10, field_id=1000, transform=IdentityTransform(), name="part"),
+        PartitionField(source_id=22, field_id=1001, transform=IdentityTransform(), name="part"),
+    )
+
+    with pytest.raises(ValueError, match="Partition name has to be unique: part"):
+        new_table_metadata(
+            schema=schema,
+            partition_spec=partition_spec,
+            sort_order=SortOrder(),
+            location="s3://some_v1_location/",
+            properties={"format-version": "2"},
+        )
+
+
 def test_encrypted_key_minimal() -> None:
     # Mirrors Java's TestEncryptedKeyParser, where the key metadata is base64 of b"key"
     key = EncryptedKey.model_validate_json('{"key-id": "a", "encrypted-key-metadata": "a2V5"}')