Support atomic table replacement transactions

mkdocs/docs/api.md

@@ -185,6 +185,26 @@ with catalog.create_table_transaction(identifier="docs_example.bids", schema=sch
     txn.set_properties(test_a="test_aa", test_b="test_b", test_c="test_c")
 ```
 
+## Replace a table
+
+Use `replace_table_transaction` to replace an existing table's definition and data in one atomic commit:
+
+```python
+with catalog.replace_table_transaction(identifier="docs_example.bids", schema=df.schema) as txn:
+    txn.append(df)
+```
+
+The existing table remains visible until the transaction commits. The replacement can use an incompatible
+schema, such as changing a column from an integer to a string. Field IDs are reused by name, while new
+fields receive IDs above the table's last assigned field ID.
+
+The table UUID, historical schemas, snapshots, and other branches and tags are preserved. Replacement data
+starts a new snapshot lineage. Committing without appending data leaves an empty table with no current snapshot.
+
+By default, the replacement is unpartitioned and unsorted. Supply `partition_spec` and `sort_order` to set
+the new layout. Existing properties are retained and merged with `properties`, and the existing location is
+retained unless `location` is supplied.
+
 ## Register a table
 
 To register a table using existing metadata:

pyiceberg/avro/resolver.py

@@ -147,7 +147,7 @@ def struct(self, struct: StructType, field_results: list[Writer]) -> Writer:
         return StructWriter(tuple((pos, result) for pos, result in enumerate(field_results)))
 
     def field(self, field: NestedField, field_result: Writer) -> Writer:
-        return field_result if field.required else OptionWriter(field_result)
+        return field_result if field.required or isinstance(field.field_type, UnknownType) else OptionWriter(field_result)
 
     def list(self, list_type: ListType, element_result: Writer) -> Writer:
         return ListWriter(element_result)
@@ -302,7 +302,9 @@ def struct(self, file_schema: StructType, record_struct: IcebergType | None, fil
         return StructWriter(field_writers=tuple(results))
 
     def field(self, file_field: NestedField, record_type: IcebergType | None, field_writer: Writer) -> Writer:
-        return field_writer if file_field.required else OptionWriter(field_writer)
+        return (
+            field_writer if file_field.required or isinstance(file_field.field_type, UnknownType) else OptionWriter(field_writer)
+        )
 
     def list(self, file_list_type: ListType, file_list: IcebergType | None, element_writer: Writer) -> Writer:
         return ListWriter(element_writer if file_list_type.element_required else OptionWriter(element_writer))

pyiceberg/catalog/__init__.py

@@ -43,12 +43,13 @@
 from pyiceberg.io import FileIO, load_file_io
 from pyiceberg.manifest import ManifestFile
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
-from pyiceberg.schema import Schema
+from pyiceberg.schema import Schema, assign_fresh_schema_ids
 from pyiceberg.serializers import ToOutputFile
 from pyiceberg.table import (
     DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE,
     CommitTableResponse,
     CreateTableTransaction,
+    ReplaceTableTransaction,
     StagedTable,
     Table,
     TableProperties,
@@ -416,6 +417,31 @@ def create_table_transaction(
             CreateTableTransaction: createTableTransaction instance.
         """
 
+    def replace_table_transaction(
+        self,
+        identifier: str | Identifier,
+        schema: Schema | pa.Schema,
+        location: str | None = None,
+        partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+        sort_order: SortOrder = UNSORTED_SORT_ORDER,
+        properties: Properties = EMPTY_DICT,
+    ) -> ReplaceTableTransaction:
+        """Replace an existing table atomically, preserving its UUID and history.
+
+        The replacement starts empty, with the supplied schema, partition spec, and
+        sort order. Properties are merged with existing properties. The location
+        defaults to the existing table location. Append data to the transaction
+        before committing to replace both the table definition and its contents.
+
+        Raises:
+            NoSuchTableError: If the table does not exist.
+        """
+        table = self.load_table(identifier)
+        iceberg_schema = self._convert_schema_if_needed(schema, table.metadata.format_version)
+        if not isinstance(schema, Schema):
+            iceberg_schema = assign_fresh_schema_ids(iceberg_schema)
+        return ReplaceTableTransaction(table, iceberg_schema, location, partition_spec, sort_order, properties)
+
     def create_table_if_not_exists(
         self,
         identifier: str | Identifier,

pyiceberg/io/pyarrow.py

@@ -153,7 +153,7 @@
 from pyiceberg.table.locations import load_location_provider
 from pyiceberg.table.metadata import TableMetadata
 from pyiceberg.table.name_mapping import NameMapping, apply_name_mapping
-from pyiceberg.transforms import IdentityTransform, TruncateTransform
+from pyiceberg.transforms import IdentityTransform, TruncateTransform, VoidTransform
 from pyiceberg.typedef import EMPTY_DICT, Properties, Record, TableVersion
 from pyiceberg.types import (
     BinaryType,
@@ -3169,6 +3169,9 @@ def _determine_partitions(spec: PartitionSpec, schema: Schema, arrow_table: pa.T
     partition_fields = [f"_partition_{field.name}" for field in spec.fields]
 
     for partition, name in zip(spec.fields, partition_fields, strict=True):
+        if isinstance(partition.transform, VoidTransform):
+            arrow_table = arrow_table.append_column(name, pa.nulls(arrow_table.num_rows, type=pa.int32()))
+            continue
         source_field = schema.find_field(partition.source_id)
         full_field_name = schema.find_column_name(partition.source_id)
         if full_field_name is None:

pyiceberg/partitioning.py

@@ -490,6 +490,8 @@ def partition_record_value(partition_field: PartitionField, value: Any, schema:
     Then the corresponding PartitionField's transform is applied to return
     the final partition record value.
     """
+    if isinstance(partition_field.transform, VoidTransform):
+        return None
     iceberg_type = schema.find_field(name_or_id=partition_field.source_id).field_type
     return _to_partition_representation(iceberg_type, value)
 

pyiceberg/schema.py

@@ -1310,27 +1310,35 @@ def build_position_accessors(schema_or_type: Schema | IcebergType) -> dict[int,
     return visit(schema_or_type, _BuildPositionAccessors())
 
 
-def assign_fresh_schema_ids(schema_or_type: Schema | IcebergType, next_id: Callable[[], int] | None = None) -> Schema:
-    """Traverses the schema, and sets new IDs."""
-    return pre_order_visit(schema_or_type, _SetFreshIDs(next_id_func=next_id))
+def assign_fresh_schema_ids(
+    schema_or_type: Schema | IcebergType,
+    next_id: Callable[[], int] | None = None,
+    base_schema: Schema | None = None,
+) -> Schema:
+    """Assign fresh IDs, reusing IDs by name from a base schema when supplied."""
+    return pre_order_visit(schema_or_type, _SetFreshIDs(next_id_func=next_id, base_schema=base_schema))
 
 
 class _SetFreshIDs(PreOrderSchemaVisitor[IcebergType]):
     """Traverses the schema and assigns monotonically increasing ids."""
 
     old_id_to_new_id: dict[int, int]
 
-    def __init__(self, next_id_func: Callable[[], int] | None = None) -> None:
+    def __init__(self, next_id_func: Callable[[], int] | None = None, base_schema: Schema | None = None) -> None:
         self.old_id_to_new_id = {}
-        counter = itertools.count(1)
+        self._base_name_to_id = index_by_name(base_schema) if base_schema is not None else {}
+        self._id_to_name: dict[int, str] = {}
+        counter = itertools.count(base_schema.highest_field_id + 1 if base_schema is not None else 1)
         self.next_id_func = next_id_func if next_id_func is not None else lambda: next(counter)
 
     def _get_and_increment(self, current_id: int) -> int:
-        new_id = self.next_id_func()
+        name = self._id_to_name.get(current_id)
+        new_id = self._base_name_to_id[name] if name is not None and name in self._base_name_to_id else self.next_id_func()
         self.old_id_to_new_id[current_id] = new_id
         return new_id
 
     def schema(self, schema: Schema, struct_result: Callable[[], StructType]) -> Schema:
+        self._id_to_name = index_name_by_id(schema) if self._base_name_to_id else {}
         return Schema(
             *struct_result().fields,
             identifier_field_ids=[self.old_id_to_new_id[field_id] for field_id in schema.identifier_field_ids],
@@ -1340,15 +1348,7 @@ def struct(self, struct: StructType, field_results: builtins.list[Callable[[], I
         new_ids = [self._get_and_increment(field.field_id) for field in struct.fields]
         new_fields = []
         for field_id, field, field_type in zip(new_ids, struct.fields, field_results, strict=True):
-            new_fields.append(
-                NestedField(
-                    field_id=field_id,
-                    name=field.name,
-                    field_type=field_type(),
-                    required=field.required,
-                    doc=field.doc,
-                )
-            )
+            new_fields.append(field.model_copy(update={"field_id": field_id, "field_type": field_type()}))
         return StructType(*new_fields)
 
     def field(self, field: NestedField, field_result: Callable[[], IcebergType]) -> IcebergType:

pyiceberg/table/__init__.py

@@ -26,7 +26,7 @@
 from collections.abc import Callable, Iterable, Iterator
 from dataclasses import dataclass
 from functools import cached_property
-from itertools import chain
+from itertools import chain, count
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, TypeVar
 
@@ -52,8 +52,15 @@
     ManifestEntryStatus,
     ManifestFile,
 )
-from pyiceberg.partitioning import PARTITION_FIELD_ID_START, UNPARTITIONED_PARTITION_SPEC, PartitionKey, PartitionSpec
-from pyiceberg.schema import Schema
+from pyiceberg.partitioning import (
+    PARTITION_FIELD_ID_START,
+    UNPARTITIONED_PARTITION_SPEC,
+    PartitionField,
+    PartitionKey,
+    PartitionSpec,
+    assign_fresh_partition_spec_ids,
+)
+from pyiceberg.schema import Schema, assign_fresh_schema_ids
 from pyiceberg.table.delete_file_index import DeleteFileIndex
 from pyiceberg.table.inspect import InspectTable
 from pyiceberg.table.locations import LocationProvider, load_location_provider
@@ -69,16 +76,22 @@
     is_ancestor_of,
     is_parent_ancestor_of,
 )
-from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
+from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder, assign_fresh_sort_order_ids
 from pyiceberg.table.update import (
     AddPartitionSpecUpdate,
     AddSchemaUpdate,
     AddSortOrderUpdate,
     AssertCreate,
+    AssertCurrentSchemaId,
+    AssertDefaultSortOrderId,
+    AssertDefaultSpecId,
+    AssertLastAssignedFieldId,
+    AssertLastAssignedPartitionId,
     AssertRefSnapshotId,
     AssertTableUUID,
     AssignUUIDUpdate,
     RemovePropertiesUpdate,
+    RemoveSnapshotRefUpdate,
     SetCurrentSchemaUpdate,
     SetDefaultSortOrderUpdate,
     SetDefaultSpecUpdate,
@@ -96,7 +109,7 @@
 from pyiceberg.table.update.sorting import UpdateSortOrder
 from pyiceberg.table.update.spec import UpdateSpec
 from pyiceberg.table.update.statistics import UpdateStatistics
-from pyiceberg.transforms import IdentityTransform
+from pyiceberg.transforms import IdentityTransform, VoidTransform
 from pyiceberg.typedef import (
     EMPTY_DICT,
     IcebergBaseModel,
@@ -1233,6 +1246,118 @@ def _rebuild_snapshot_updates(self) -> None:
             self._stage(updates, requirements)
 
 
+class ReplaceTableTransaction(Transaction):
+    """A transaction that atomically replaces a table while retaining its history."""
+
+    def __init__(
+        self,
+        table: Table,
+        schema: Schema,
+        location: str | None = None,
+        partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+        sort_order: SortOrder = UNSORTED_SORT_ORDER,
+        properties: Properties = EMPTY_DICT,
+    ):
+        super().__init__(table)
+        base = table.metadata
+        schema.check_format_version_compatibility(int(properties.get(TableProperties.FORMAT_VERSION, base.format_version)))
+        next_id = count(base.last_column_id + 1)
+        fresh_schema = assign_fresh_schema_ids(schema, lambda: next(next_id), base.schema())
+        existing_schema = next((old for old in base.schemas if old == fresh_schema), None)
+        updates: tuple[TableUpdate, ...] = (RemoveSnapshotRefUpdate(ref_name=MAIN_BRANCH),)
+        if existing_schema is None:
+            fresh_schema = fresh_schema.model_copy(update={"schema_id": max(old.schema_id for old in base.schemas) + 1})
+            updates += (AddSchemaUpdate(schema_=fresh_schema),)
+        else:
+            fresh_schema = existing_schema
+        updates += (SetCurrentSchemaUpdate(schema_id=fresh_schema.schema_id),)
+
+        fresh_spec = assign_fresh_partition_spec_ids(partition_spec, schema, fresh_schema)
+        fresh_spec = self._replacement_spec(fresh_spec)
+        fresh_spec.check_compatible(fresh_schema)
+        if fresh_spec.spec_id not in base.specs():
+            updates += (AddPartitionSpecUpdate(spec=fresh_spec),)
+        updates += (SetDefaultSpecUpdate(spec_id=fresh_spec.spec_id),)
+
+        fresh_order = assign_fresh_sort_order_ids(sort_order, schema, fresh_schema)
+        fresh_order.check_compatible(fresh_schema)
+        existing_order = next((old for old in base.sort_orders if old.fields == fresh_order.fields), None)
+        if existing_order is None:
+            if not fresh_order.is_unsorted:
+                fresh_order = fresh_order.model_copy(
+                    update={"order_id": max((old.order_id for old in base.sort_orders), default=0) + 1}
+                )
+            updates += (AddSortOrderUpdate(sort_order=fresh_order),)
+        else:
+            fresh_order = existing_order
+        updates += (SetDefaultSortOrderUpdate(sort_order_id=fresh_order.order_id),)
+
+        persisted_properties = dict(properties)
+        if format_version := persisted_properties.pop(TableProperties.FORMAT_VERSION, None):
+            updates += (UpgradeFormatVersionUpdate(format_version=int(format_version)),)
+        if location is not None:
+            updates += (SetLocationUpdate(location=location.rstrip("/")),)
+        updates += (SetPropertiesUpdate(updates=persisted_properties),)
+        self._stage(
+            updates,
+            (
+                AssertLastAssignedFieldId(last_assigned_field_id=base.last_column_id),
+                AssertCurrentSchemaId(current_schema_id=base.current_schema_id),
+                AssertLastAssignedPartitionId(last_assigned_partition_id=base.last_partition_id),
+                AssertDefaultSpecId(default_spec_id=base.default_spec_id),
+                AssertDefaultSortOrderId(default_sort_order_id=base.default_sort_order_id),
+                AssertRefSnapshotId(ref=MAIN_BRANCH, snapshot_id=base.current_snapshot_id),
+            ),
+        )
+
+    def _replacement_spec(self, spec: PartitionSpec) -> PartitionSpec:
+        base = self._table.metadata
+        next_id = count((base.last_partition_id or PARTITION_FIELD_ID_START - 1) + 1)
+        fields: list[PartitionField] = []
+        if base.format_version == 1:
+            remaining = {(field.source_id, str(field.transform)): field for field in spec.fields}
+            new_names = {field.name for field in spec.fields}
+            for old in base.spec().fields:
+                if replacement := remaining.pop((old.source_id, str(old.transform)), None):
+                    fields.append(replacement.model_copy(update={"field_id": old.field_id}))
+                else:
+                    fields.append(
+                        old.model_copy(
+                            update={
+                                "transform": VoidTransform(),
+                                "name": f"{old.name}_{old.field_id}" if old.name in new_names else old.name,
+                            }
+                        )
+                    )
+            fields.extend(field.model_copy(update={"field_id": next(next_id)}) for field in remaining.values())
+        else:
+            historical_ids: dict[tuple[int, str], int] = {}
+            for old_spec in base.partition_specs:
+                for field in old_spec.fields:
+                    key = (field.source_id, str(field.transform))
+                    historical_ids[key] = max(historical_ids.get(key, 0), field.field_id)
+            for field in spec.fields:
+                key = (field.source_id, str(field.transform))
+                if key not in historical_ids:
+                    historical_ids[key] = next(next_id)
+                fields.append(field.model_copy(update={"field_id": historical_ids[key]}))
+        existing_spec = next((old for old in base.partition_specs if old.fields == tuple(fields)), None)
+        return existing_spec or PartitionSpec(*fields, spec_id=max(old.spec_id for old in base.partition_specs) + 1)
+
+    def _stage(self, updates: tuple[TableUpdate, ...], requirements: tuple[TableRequirement, ...] = ()) -> Transaction:
+        # Main is cleared in staged metadata, so guard the published head instead.
+        replaces_main = any(isinstance(req, AssertRefSnapshotId) and req.ref == MAIN_BRANCH for req in requirements)
+        super()._stage(
+            updates,
+            tuple(req for req in requirements if not (isinstance(req, AssertRefSnapshotId) and req.ref == MAIN_BRANCH)),
+        )
+        if replaces_main and not any(
+            isinstance(req, AssertRefSnapshotId) and req.ref == MAIN_BRANCH for req in self._requirements
+        ):
+            self._requirements += (AssertRefSnapshotId(ref=MAIN_BRANCH, snapshot_id=self._table.metadata.current_snapshot_id),)
+        return self
+
+
 class CreateTableTransaction(Transaction):
     """A transaction that involves the creation of a new table."""
 

pyiceberg/utils/schema_conversion.py

@@ -540,7 +540,7 @@ def field(self, field: NestedField, field_result: AvroType) -> AvroType:
         result = {
             "name": sanitized_name,
             FIELD_ID_PROP: field.field_id,
-            "type": field_result if field.required else ["null", field_result],
+            "type": field_result if field.required or field_result == "null" else ["null", field_result],
         }
 
         if original_name != sanitized_name:

tests/avro/test_file.py

@@ -56,6 +56,7 @@
     TimestamptzNanoType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 from pyiceberg.utils.schema_conversion import AvroSchemaConversion
@@ -571,3 +572,19 @@ def test_avro_output_file_tell_includes_buffered_records() -> None:
             out.write_block([manifest_entry(0)])
 
             assert out.tell() > after_header
+
+
+@pytest.mark.parametrize("with_record_schema", [False, True])
+def test_write_optional_unknown_field(with_record_schema: bool) -> None:
+    schema = Schema(NestedField(1, "removed_partition", UnknownType()), NestedField(2, "value", LongType()))
+    with TemporaryDirectory() as tmpdir:
+        path = tmpdir + "/null.avro"
+        with avro.AvroOutputFile[Record](
+            output_file=PyArrowFileIO().new_output(path),
+            file_schema=schema,
+            record_schema=schema if with_record_schema else None,
+            schema_name="null_field",
+        ) as out:
+            out.write_block([Record(None, 42)])
+        with open(path, "rb") as source:
+            assert list(reader(source)) == [{"removed_partition": None, "value": 42}]