partition-schema name conflict validation when renaming column

Author: rutb327

pyiceberg/table/update/schema.py

@@ -193,7 +193,7 @@ def add_column(
         parent_id: int = TABLE_ROOT_ID
 
         # Check for conflicts with partition field names
-        if len(parent) == 0 and self._transaction is not None:
+        if self._transaction is not None:
             for spec in self._transaction.table_metadata.partition_specs:
                 for field in spec.fields:
                     if field.name == name:
@@ -320,6 +320,20 @@ def rename_column(self, path_from: Union[str, Tuple[str, ...]], new_name: str) -
         if field_from.field_id in self._deletes:
             raise ValueError(f"Cannot rename a column that will be deleted: {path_from}")
 
+        if self._transaction is not None:
+            for spec in self._transaction.table_metadata.partition_specs:
+                for field in spec.fields:
+                    if field.name == new_name:
+                        from pyiceberg.transforms import IdentityTransform, VoidTransform
+
+                        if isinstance(field.transform, (IdentityTransform, VoidTransform)):
+                            # For identity transforms, allow conflict only if partition field sources from the renamed field
+                            if field.source_id != field_from.field_id:
+                                raise ValueError(f"Cannot rename column to name that conflicts with partition field: {new_name}")
+                        else:
+                            # For non-identity transforms, never allow conflicts
+                            raise ValueError(f"Cannot rename column to name that conflicts with partition field: {new_name}")
+
         if updated := self._updates.get(field_from.field_id):
             self._updates[field_from.field_id] = NestedField(
                 field_id=updated.field_id,

tests/integration/test_partition_evolution.py

@@ -638,8 +638,14 @@ def test_schema_evolution_partition_conflict(catalog: Catalog) -> None:
 
     with pytest.raises(ValueError, match="Cannot add column with name that conflicts with partition field: event_year"):
         table.update_schema().add_column("event_year", StringType()).commit()
-
     with pytest.raises(ValueError, match="Cannot add column with name that conflicts with partition field: 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 rename column to name that conflicts with partition field: event_year"):
+        table.update_schema().rename_column("other_field", "event_year").commit()
+    with pytest.raises(ValueError, match="Cannot rename column to name that conflicts with partition field: first_name"):
+        table.update_schema().rename_column("other_field", "first_name").commit()
+
+    table.update_schema().rename_column("other_field", "valid_name").commit()