Point to the original name when moving a column renamed in the same update

mkdocs/docs/api.md

@@ -1242,6 +1242,8 @@ with table.update_schema() as update:
     update.rename_column(("details", "confirmed_by"), "processed_by")
 ```
 
+Within the same `update_schema()`, other operations still refer to a renamed column by its original name, for example `update.move_first("retries")` after the rename above.
+
 ### Move column
 
 Move order of fields:

pyiceberg/table/update/schema.py

@@ -510,6 +510,19 @@ def _find_for_move(self, name: str) -> int | None:
 
         return self._added_name_to_id.get(name)
 
+    def _missing_for_move(self, name: str) -> str:
+        """Describe a name that cannot be moved, pointing to the original name if it is a pending rename."""
+        for field_id, update in self._updates.items():
+            field = self._schema.find_field(field_id)
+            if update.name == field.name:
+                continue
+            parent = self._id_to_parent.get(field_id)
+            new_name = f"{parent}.{update.name}" if parent else update.name
+            if new_name == name or (not self._case_sensitive and new_name.lower() == name.lower()):
+                original_name = self._schema.find_column_name(field_id)
+                return f"{name} (renamed from {original_name} in this update, refer to it as {original_name})"
+        return name
+
     def _move(self, move: _Move) -> None:
         if parent_name := self._id_to_parent.get(move.field_id):
             parent_field = self._schema.find_field(parent_name, case_sensitive=self._case_sensitive)
@@ -549,7 +562,7 @@ def move_first(self, path: str | tuple[str, ...]) -> UpdateSchema:
         field_id = self._find_for_move(full_name)
 
         if field_id is None:
-            raise ValueError(f"Cannot move missing column: {full_name}")
+            raise ValueError(f"Cannot move missing column: {self._missing_for_move(full_name)}")
 
         self._move(_Move(field_id=field_id, full_name=full_name, op=_MoveOperation.First))
 
@@ -568,7 +581,7 @@ def move_before(self, path: str | tuple[str, ...], before_path: str | tuple[str,
         field_id = self._find_for_move(full_name)
 
         if field_id is None:
-            raise ValueError(f"Cannot move missing column: {full_name}")
+            raise ValueError(f"Cannot move missing column: {self._missing_for_move(full_name)}")
 
         before_full_name = (
             ".".join(
@@ -580,7 +593,7 @@ def move_before(self, path: str | tuple[str, ...], before_path: str | tuple[str,
         before_field_id = self._find_for_move(before_full_name)
 
         if before_field_id is None:
-            raise ValueError(f"Cannot move {full_name} before missing column: {before_full_name}")
+            raise ValueError(f"Cannot move {full_name} before missing column: {self._missing_for_move(before_full_name)}")
 
         if field_id == before_field_id:
             raise ValueError(f"Cannot move {full_name} before itself")
@@ -603,13 +616,13 @@ def move_after(self, path: str | tuple[str, ...], after_name: str | tuple[str, .
         field_id = self._find_for_move(full_name)
 
         if field_id is None:
-            raise ValueError(f"Cannot move missing column: {full_name}")
+            raise ValueError(f"Cannot move missing column: {self._missing_for_move(full_name)}")
 
         after_path = ".".join(after_name) if isinstance(after_name, tuple) else after_name
         after_field_id = self._find_for_move(after_path)
 
         if after_field_id is None:
-            raise ValueError(f"Cannot move {full_name} after missing column: {after_path}")
+            raise ValueError(f"Cannot move {full_name} after missing column: {self._missing_for_move(after_path)}")
 
         if field_id == after_field_id:
             raise ValueError(f"Cannot move {full_name} after itself")

tests/test_schema.py

@@ -1844,3 +1844,18 @@ def test_check_schema_compatible_optional_map_field_present() -> None:
     )
     # Should not raise - schemas match
     _check_schema_compatible(requested_schema, provided_schema)
+
+
+def test_move_renamed_column_by_new_name_points_to_original_name(table_v2: Table) -> None:
+    update = UpdateSchema(transaction=Transaction(table_v2)).rename_column("y", "y2")
+
+    with pytest.raises(ValueError, match=r"Cannot move missing column: y2 \(renamed from y in this update, refer to it as y\)"):
+        update.move_first("y2")
+    with pytest.raises(ValueError, match=r"before missing column: y2 \(renamed from y in this update, refer to it as y\)"):
+        update.move_before("x", "y2")
+    with pytest.raises(ValueError, match=r"after missing column: y2 \(renamed from y in this update, refer to it as y\)"):
+        update.move_after("x", "y2")
+    with pytest.raises(ValueError, match=r"^Cannot move missing column: unknown$"):
+        update.move_first("unknown")
+
+    assert [field.name for field in update.move_first("y")._apply().fields] == ["y2", "x", "z"]