From 88a6d777285ae4b3527f69b1a749a999063fa837 Mon Sep 17 00:00:00 2001
From: Sanjay Santhanam <51058514+Sanjays2402@users.noreply.github.com>
Date: Sun, 19 Jul 2026 00:05:19 -0700
Subject: [PATCH] Resolve staged renames when moving columns in UpdateSchema

Fixes #2599. rename_column stages the new name in self._updates, but
move_first/move_before/move_after resolved the target via the original
schema, so moving a column by its new name in the same update context
raised 'Cannot move missing column'. _find_for_move now checks staged
renames first, matching the Java implementation.
---
 pyiceberg/table/update/schema.py  |  9 ++++
 tests/table/test_update_schema.py | 73 +++++++++++++++++++++++++++++++
 2 files changed, 82 insertions(+)
 create mode 100644 tests/table/test_update_schema.py

diff --git a/pyiceberg/table/update/schema.py b/pyiceberg/table/update/schema.py
index 828f1e877a..1b0de068cb 100644
--- a/pyiceberg/table/update/schema.py
+++ b/pyiceberg/table/update/schema.py
@@ -503,6 +503,15 @@ def update_column(
         return self
 
     def _find_for_move(self, name: str) -> int | None:
+        # Resolve staged renames first so a column can be moved by its new name
+        # within the same update context (matches the Java implementation).
+        for field_id, updated in self._updates.items():
+            if self._case_sensitive:
+                if updated.name == name:
+                    return field_id
+            elif updated.name.lower() == name.lower():
+                return field_id
+
         try:
             return self._schema.find_field(name, self._case_sensitive).field_id
         except ValueError:
diff --git a/tests/table/test_update_schema.py b/tests/table/test_update_schema.py
new file mode 100644
index 0000000000..162c3330b3
--- /dev/null
+++ b/tests/table/test_update_schema.py
@@ -0,0 +1,73 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from pathlib import Path
+
+import pytest
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.catalog.memory import InMemoryCatalog
+from pyiceberg.schema import Schema
+from pyiceberg.types import IntegerType, NestedField, StringType
+
+
+@pytest.fixture
+def catalog(tmp_path: Path) -> Catalog:
+    catalog = InMemoryCatalog("test", warehouse=str(tmp_path), uri=f"sqlite:///{tmp_path}/catalog.db")
+    catalog.create_namespace("default")
+    return catalog
+
+
+TEST_SCHEMA = Schema(
+    NestedField(field_id=1, name="id", field_type=IntegerType(), required=True),
+    NestedField(field_id=2, name="some_column", field_type=StringType(), required=False),
+    NestedField(field_id=3, name="other_column", field_type=StringType(), required=False),
+)
+
+
+def test_rename_then_move_first_by_new_name(catalog: Catalog) -> None:
+    """Renaming a column and then moving it by its new name in the same
+    update context should succeed (matches the Java implementation).
+
+    Regression test for https://github.com/apache/iceberg-python/issues/2599.
+    """
+    tbl = catalog.create_table("default.test_rename_move_first", TEST_SCHEMA)
+
+    with tbl.update_schema() as update:
+        update.rename_column("some_column", "renamed_column")
+        update.move_first("renamed_column")
+
+    assert [field.name for field in tbl.schema().fields] == ["renamed_column", "id", "other_column"]
+
+
+def test_rename_then_move_before_by_new_name(catalog: Catalog) -> None:
+    tbl = catalog.create_table("default.test_rename_move_before", TEST_SCHEMA)
+
+    with tbl.update_schema() as update:
+        update.rename_column("some_column", "renamed_column")
+        update.move_before("renamed_column", "id")
+
+    assert [field.name for field in tbl.schema().fields] == ["renamed_column", "id", "other_column"]
+
+
+def test_rename_then_move_after_by_new_name(catalog: Catalog) -> None:
+    tbl = catalog.create_table("default.test_rename_move_after", TEST_SCHEMA)
+
+    with tbl.update_schema() as update:
+        update.rename_column("some_column", "renamed_column")
+        update.move_after("renamed_column", "other_column")
+
+    assert [field.name for field in tbl.schema().fields] == ["id", "other_column", "renamed_column"]