Allow adding a partition field with the name of a removed one

pyiceberg/table/update/spec.py

@@ -111,13 +111,18 @@ def add_field(
         self._transform_to_added_field[transform_key] = new_field
 
         existing_partition_field = self._name_to_field.get(new_field.name)
-        if existing_partition_field and new_field.field_id not in self._deletes:
+        if existing_partition_field and existing_partition_field.field_id not in self._deletes:
             if isinstance(existing_partition_field.transform, VoidTransform):
                 self.rename_field(
                     existing_partition_field.name, existing_partition_field.name + "_" + str(existing_partition_field.field_id)
                 )
             else:
                 raise ValueError(f"Cannot add duplicate partition field name: {existing_partition_field.name}")
+        elif existing_partition_field:
+            # The removed field becomes a void field on v1 tables, so it gives up its name
+            self._renames[existing_partition_field.name] = (
+                existing_partition_field.name + "_" + str(existing_partition_field.field_id)
+            )
 
         self._name_to_added_field[new_field.name] = new_field
         self._adds.append(new_field)

tests/table/test_init.py

@@ -97,6 +97,7 @@
 from pyiceberg.transforms import (
     BucketTransform,
     IdentityTransform,
+    VoidTransform,
 )
 from pyiceberg.typedef import Record
 from pyiceberg.types import (
@@ -1546,6 +1547,25 @@ def test_update_spec_remove_and_add_same_field_reuses_id(table: Table) -> None:
     assert txn.table_metadata.last_partition_id == 1000
 
 
+def test_update_spec_remove_and_add_field_with_same_name(table_v1: Table, table_v2: Table) -> None:
+    for table in [table_v1, table_v2]:
+        txn = table.transaction()
+        txn.update_spec().add_field("y", BucketTransform(16), "y_shard").commit()
+        txn.update_spec().remove_field("y_shard").add_field("y", BucketTransform(32), "y_shard").commit()
+
+        if table.format_version == 1:
+            assert txn.table_metadata.spec().fields == (
+                PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"),
+                PartitionField(source_id=2, field_id=1001, transform=VoidTransform(), name="y_shard_1001"),
+                PartitionField(source_id=2, field_id=1002, transform=BucketTransform(32), name="y_shard"),
+            )
+        else:
+            assert txn.table_metadata.spec().fields == (
+                PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"),
+                PartitionField(source_id=2, field_id=1002, transform=BucketTransform(32), name="y_shard"),
+            )
+
+
 def test_assert_default_spec_id(table_v2: Table) -> None:
     base_metadata = table_v2.metadata
     AssertDefaultSpecId(default_spec_id=base_metadata.default_spec_id).validate(base_metadata)