Bug Fix: Use historical partition field name

pyiceberg/table/update/spec.py

@@ -275,12 +275,12 @@ def _partition_field(self, transform_key: Tuple[int, Transform[Any, Any]], name:
             historical_fields = []
             for spec in self._transaction.table_metadata.specs().values():
                 for field in spec.fields:
-                    historical_fields.append((field.source_id, field.field_id, repr(field.transform), field.name))
+                    historical_fields.append(field)
 
-            for field_key in historical_fields:
-                if field_key[0] == source_id and field_key[2] == repr(transform):
-                    if name is None or field_key[3] == name:
-                        return PartitionField(source_id, field_key[1], transform, name)
+            for field in historical_fields:
+                if field.source_id == source_id and repr(field.transform) == repr(transform):
+                    if name is None or field.name == name:
+                        return PartitionField(source_id, field.field_id, transform, field.name)
 
         new_field_id = self._new_field_id()
         if name is None:

tests/integration/test_partition_evolution.py

@@ -207,6 +207,38 @@ def test_remove_identity_v2(catalog: Catalog) -> None:
     assert table_v2.spec() == PartitionSpec(spec_id=0)
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
+def test_remove_and_add_identity(catalog: Catalog) -> None:
+    table = _table(catalog)
+    table.update_spec().add_identity("id").commit()
+    table.update_spec().remove_field("id").commit()
+    table.update_spec().add_identity("id").commit()
+
+    assert len(table.specs()) == 4
+    assert table.spec().spec_id == 3
+    assert table.spec() == PartitionSpec(
+        PartitionField(source_id=1, field_id=1000, transform=VoidTransform(), name="id_1000"),
+        PartitionField(source_id=1, field_id=1001, transform=IdentityTransform(), name="id"),
+        spec_id=3,
+    )
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
+def test_remove_and_add_identity_v2(catalog: Catalog) -> None:
+    table_v2 = _table_v2(catalog)
+    table_v2.update_spec().add_identity("id").commit()
+    table_v2.update_spec().remove_field("id").commit()
+    table_v2.update_spec().add_identity("id").commit()
+
+    assert len(table_v2.specs()) == 2
+    assert table_v2.spec().spec_id == 1
+    assert table_v2.spec() == PartitionSpec(
+        PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="id"), spec_id=1
+    )
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
 def test_remove_bucket(catalog: Catalog) -> None: