fix

Author: KazuhitoT

pyiceberg/table/update/schema.py

@@ -146,9 +146,13 @@ def _reset_state(self) -> None:
         """
         self._init_state_from_metadata(self._provided_schema)
 
-        # Refresh name mapping from the latest table metadata to avoid overwriting concurrent changes
+        # Refresh name mapping from the latest table metadata to avoid overwriting concurrent changes.
         if self._name_mapping is not None:
-            self._name_mapping = self._transaction.table_metadata.name_mapping()
+            refreshed_mapping = self._transaction.table_metadata.name_mapping()
+
+            # Check refreshed_mapping because if another transaction removes schema.name-mapping.default, refreshed_mapping may become null.
+            if refreshed_mapping is not None:
+                self._name_mapping = refreshed_mapping
 
         for operation in self._operations:
             op_type = operation[0]

pyiceberg/table/update/snapshot.py

@@ -1109,10 +1109,11 @@ def _reset_state(self) -> None:
         """Clear accumulated updates for retry.
 
         The _snapshot_ids_to_expire are user-provided and preserved.
-        The _updates are cleared so _commit() can rebuild them with
+        The _updates and _requirements are cleared so _commit() can rebuild them with
         refreshed protected snapshot IDs.
         """
         self._updates = ()
+        self._requirements = ()
 
     def _commit(self) -> UpdatesAndRequirements:
         """

pyiceberg/table/update/spec.py

@@ -153,7 +153,8 @@ def add_field(
         transform: str | Transform[Any, Any],
         partition_field_name: str | None = None,
     ) -> UpdateSpec:
-        transform = parse_transform(transform)
+        if isinstance(transform, str):
+            transform = parse_transform(transform)
         self._operations.append(("add", source_column_name, transform, partition_field_name))
         self._do_add_field(source_column_name, transform, partition_field_name)
         return self