in pyiceberg/table/update/schema.py [0:0]
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)
if not parent_field.field_type.is_struct:
raise ValueError(f"Cannot move fields in non-struct type: {parent_field.field_type}")
if move.op == _MoveOperation.After or move.op == _MoveOperation.Before:
if move.other_field_id is None:
raise ValueError("Expected other field when performing before/after move")
if self._id_to_parent.get(move.field_id) != self._id_to_parent.get(move.other_field_id):
raise ValueError(f"Cannot move field {move.full_name} to a different struct")
self._moves[parent_field.field_id] = self._moves.get(parent_field.field_id, []) + [move]
else:
# In the top level field
if move.op == _MoveOperation.After or move.op == _MoveOperation.Before:
if move.other_field_id is None:
raise ValueError("Expected other field when performing before/after move")
if other_struct := self._id_to_parent.get(move.other_field_id):
raise ValueError(f"Cannot move field {move.full_name} to a different struct: {other_struct}")
self._moves[TABLE_ROOT_ID] = self._moves.get(TABLE_ROOT_ID, []) + [move]