Core: Keep mapped names of existing fields when a column is added

pyiceberg/table/name_mapping.py

@@ -186,53 +186,21 @@ def __init__(self, updates: dict[int, NestedField], adds: dict[int, list[NestedF
         self._updates = updates
         self._adds = adds
 
-    @staticmethod
-    def _remove_reassigned_names(field: MappedField, assignments: dict[str, int]) -> MappedField | None:
-        removed_names = set()
-        for name in field.names:
-            if (assigned_id := assignments.get(name)) and assigned_id != field.field_id:
-                removed_names.add(name)
-
-        remaining_names = [f for f in field.names if f not in removed_names]
-        if remaining_names:
-            return MappedField(field_id=field.field_id, names=remaining_names, fields=field.fields)
-        else:
-            return None
-
     def _add_new_fields(self, mapped_fields: list[MappedField], parent_id: int) -> list[MappedField]:
         if fields_to_add := self._adds.get(parent_id):
-            fields: list[MappedField] = []
-            new_fields: list[MappedField] = []
-
-            for add in fields_to_add:
-                new_fields.append(
-                    MappedField(field_id=add.field_id, names=[add.name], fields=visit(add.field_type, _CreateMapping()))
-                )
-
-            reassignments = {f.name: f.field_id for f in fields_to_add}
-            fields = [
-                updated_field
-                for field in mapped_fields
-                if (updated_field := self._remove_reassigned_names(field, reassignments)) is not None
-            ] + new_fields
-            return fields
+            new_fields = [
+                MappedField(field_id=add.field_id, names=[add.name], fields=visit(add.field_type, _CreateMapping()))
+                for add in fields_to_add
+            ]
+            return mapped_fields + new_fields
         else:
             return mapped_fields
 
     def mapping(self, nm: NameMapping, field_results: list[MappedField]) -> list[MappedField]:
         return self._add_new_fields(field_results, -1)
 
     def fields(self, struct: list[MappedField], field_results: list[MappedField]) -> list[MappedField]:
-        reassignments: dict[str, int] = {
-            update.name: update.field_id
-            for f in field_results
-            if f.field_id is not None and (update := self._updates.get(f.field_id))
-        }
-        return [
-            updated_field
-            for field in field_results
-            if (updated_field := self._remove_reassigned_names(field, reassignments)) is not None
-        ]
+        return field_results
 
     def field(self, field: MappedField, field_result: list[MappedField]) -> MappedField:
         if field.field_id is None:

tests/catalog/test_catalog_behaviors.py

@@ -26,6 +26,7 @@
 from typing import Any
 
 import pyarrow as pa
+import pyarrow.parquet as pq
 import pytest
 from pydantic_core import ValidationError
 from pytest_lazy_fixtures import lf
@@ -338,6 +339,26 @@ def test_write_pyarrow_schema(catalog: Catalog, test_table_identifier: Identifie
     table.append(pyarrow_table)
 
 
+def test_add_column_keeps_values_of_renamed_column_without_field_ids(
+    catalog_with_warehouse: Catalog,
+    tmp_path: Path,
+) -> None:
+    imported = tmp_path / "imported.parquet"
+    pq.write_table(pa.table({"a": [1, 3]}), imported)
+
+    catalog_with_warehouse.create_namespace("default")
+    table = catalog_with_warehouse.create_table(("default", "imported"), Schema(NestedField(1, "a", LongType())))
+    table.add_files([f"file://{imported}"])
+
+    with table.update_schema() as update:
+        update.rename_column("a", "b")
+
+    with table.update_schema() as update:
+        update.add_column("a", LongType())
+
+    assert table.scan().to_arrow().to_pydict() == {"b": [1, 3], "a": [None, None]}
+
+
 @pytest.mark.parametrize(
     "schema,expected",
     [

tests/table/test_name_mapping.py

@@ -345,6 +345,7 @@ def test_update_mapping(table_name_mapping_nested: NameMapping) -> None:
                 field_id=15,
                 names=["person"],
                 fields=[
+                    MappedField(field_id=16, names=["name"]),
                     MappedField(field_id=17, names=["age"]),
                     MappedField(field_id=19, names=["name"]),
                     MappedField(field_id=20, names=["add_20"]),
@@ -356,6 +357,16 @@ def test_update_mapping(table_name_mapping_nested: NameMapping) -> None:
     assert update_mapping(table_name_mapping_nested, updates, adds) == expected
 
 
+def test_update_mapping_keeps_name_of_existing_field(table_name_mapping_nested: NameMapping) -> None:
+    updates = {1: NestedField(1, "foo_update", StringType(), True)}
+    adds = {-1: [NestedField(18, "foo", StringType(), True)]}
+
+    updated = update_mapping(table_name_mapping_nested, updates, adds)
+
+    assert updated.root[0] == MappedField(field_id=1, names=["foo", "foo_update"])
+    assert updated.root[-1] == MappedField(field_id=18, names=["foo"])
+
+
 def test_mapping_using_by_visitor(table_schema_nested: Schema, table_name_mapping_nested: NameMapping) -> None:
     schema_without_ids = Schema(
         NestedField(field_id=0, name="foo", field_type=StringType(), required=False),