Python: Add updates, moves and deletes

python/mkdocs/docs/api.md

@@ -150,32 +150,25 @@ catalog.create_table(
 
 Add new columns through the `Transaction` or `UpdateSchema` API:
 
-Use the Transaction API:
-
 ```python
-with table.transaction() as transaction:
-    transaction.update_schema().add_column("x", IntegerType(), "doc").commit()
+with table.update_schema() as update:
+    update.add_column("x", IntegerType(), "doc")
 ```
 
-Or, without a context manager:
+Or, without a context manager by calling the `.commit()` explicitly:
 
 ```python
-transaction = table.transaction()
-transaction.update_schema().add_column("x", IntegerType(), "doc").commit()
-transaction.commit_transaction()
+table.update_schema().add_column("x", IntegerType(), "doc").commit()
 ```
 
-Or, use the UpdateSchema API directly:
+Alternatively, use the transaction API to combine changes from multiple operations:
 
 ```python
-with table.update_schema() as update:
-    update.add_column("x", IntegerType(), "doc")
-```
-
-Or, without a context manager:
+from datetime import datetime
 
-```python
-table.update_schema().add_column("x", IntegerType(), "doc").commit()
+with table.transaction() as transaction:
+    transaction.update_schema().add_column("x", IntegerType(), "doc").commit()
+    transaction.set_properties(schema_updated_at=str(datetime.now()))
 ```
 
 ### Update table properties

python/pyiceberg/schema.py

@@ -245,6 +245,21 @@ def accessor_for_field(self, field_id: int) -> Accessor:
 
         return self._lazy_id_to_accessor[field_id]
 
+    def identifier_field_names(self) -> Set[str]:
+        """Returns the names of the identifier fields.
+
+        Returns:
+            Set of names of the identifier fields
+        """
+        ids = set()
+        for field_id in self.identifier_field_ids:
+            column_name = self.find_column_name(field_id)
+            if column_name is None:
+                raise ValueError(f"Could not find identifier column id: {field_id}")
+            ids.add(column_name)
+
+        return ids
+
     def select(self, *names: str, case_sensitive: bool = True) -> Schema:
         """Return a new schema instance pruned to a subset of columns.
 
@@ -892,12 +907,6 @@ def __init__(self) -> None:
         self._field_names: List[str] = []
         self._short_field_names: List[str] = []
 
-    def before_map_key(self, key: NestedField) -> None:
-        self.before_field(key)
-
-    def after_map_key(self, key: NestedField) -> None:
-        self.after_field(key)
-
     def before_map_value(self, value: NestedField) -> None:
         if not isinstance(value.field_type, StructType):
             self._short_field_names.append(value.name)
@@ -1161,6 +1170,16 @@ def primitive(self, primitive: PrimitiveType) -> PrimitiveType:
 
 
 def prune_columns(schema: Schema, selected: Set[int], select_full_types: bool = True) -> Schema:
+    """Prunes a column by only selecting a set of field-ids.
+
+    Args:
+        schema: The schema to be pruned.
+        selected: The field-ids to be included.
+        select_full_types: Return the full struct when a subset is recorded
+
+    Returns:
+        The pruned schema.
+    """
     result = visit(schema.as_struct(), _PruneColumnsVisitor(selected, select_full_types))
     return Schema(
         *(result or StructType()).fields,