Extend replace_table to all catalogs, harden semantics

- Move replace_table_transaction implementations to MetastoreCatalog and
  RestCatalog; keep stub + shared _replace_staged_table helper on Catalog
  so out-of-tree subclasses don't break.
- Emit UpgradeFormatVersionUpdate when properties bump format-version,
  matching Java's TableMetadata.buildReplacement.
- Always emit SetCurrentSchema / SetDefaultPartitionSpec /
  SetDefaultSortOrder, mirroring RESTSessionCatalog.replaceTransaction.
- Handle v1 partition specs by carrying forward removed fields as
  VoidTransform (v1 specs are append-only).
- Reject view collisions before replacing.
- Assign fresh schema_id / spec_id / order_id on Add* updates so
  AddSchemaUpdate / AddPartitionSpecUpdate / AddSortOrderUpdate produce
  uniquely-keyed entries.
- Tests: behaviour now covered in tests/catalog/test_catalog_behaviors.py
  parametrized across InMemoryCatalog + SqlCatalog. test_rest.py keeps
  only REST wire-specific cases (payload shape, 404, view collision,
  immediate commit). One end-to-end smoke test remains in
  tests/integration/test_rest_catalog.py.
- Add docs section in mkdocs/docs/api.md.

Author: smaheshwar-pltr

mkdocs/docs/api.md

@@ -185,6 +185,43 @@ 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
+
+Atomically replace an existing table's schema, partition spec, sort order, location, and properties. The table UUID and history (snapshots, schemas, specs, sort orders, metadata log) are preserved; the current snapshot is cleared (the `main` branch ref is removed). This is the analog of Spark/Trino's `CREATE OR REPLACE TABLE` for the table-metadata side, and supports RTAS-style workflows when combined with subsequent writes.
+
+```python
+from pyiceberg.schema import Schema
+from pyiceberg.types import NestedField, LongType, StringType, BooleanType
+
+new_schema = Schema(
+    NestedField(field_id=1, name="datetime", field_type=LongType(), required=False),
+    NestedField(field_id=2, name="symbol", field_type=StringType(), required=False),
+    NestedField(field_id=3, name="active", field_type=BooleanType(), required=False),
+)
+catalog.replace_table(
+    identifier="docs_example.bids",
+    schema=new_schema,
+)
+```
+
+Field IDs from columns whose names appear in the previous schema are reused, so existing data files remain readable when the new schema is a compatible superset. New columns get fresh IDs above `last-column-id`.
+
+Use `replace_table_transaction` to stage additional changes (writes, property updates, schema evolution) before committing — the equivalent of `CREATE OR REPLACE TABLE AS SELECT`:
+
+```python
+with catalog.replace_table_transaction(identifier="docs_example.bids", schema=new_schema) as txn:
+    with txn.update_snapshot().fast_append() as snap:
+        for data_file in _dataframe_to_data_files(table_metadata=txn.table_metadata, df=df, io=txn._table.io):
+            snap.append_data_file(data_file)
+    txn.set_properties(write_replaced_at="2026-04-19T00:00:00Z")
+```
+
+To upgrade the table's format version as part of the replace, pass `format-version` in `properties`:
+
+```python
+catalog.replace_table(identifier="docs_example.bids", schema=new_schema, properties={"format-version": "2"})
+```
+
 ## Register a table
 
 To register a table using existing metadata:

pyiceberg/catalog/__init__.py

@@ -42,8 +42,12 @@
 )
 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.partitioning import (
+    UNPARTITIONED_PARTITION_SPEC,
+    PartitionSpec,
+    assign_fresh_partition_spec_ids_for_replace,
+)
+from pyiceberg.schema import Schema, assign_fresh_schema_ids_for_replace
 from pyiceberg.serializers import ToOutputFile
 from pyiceberg.table import (
     DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE,
@@ -56,7 +60,7 @@
 )
 from pyiceberg.table.locations import load_location_provider
 from pyiceberg.table.metadata import TableMetadata, TableMetadataV1, new_table_metadata
-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 (
     TableRequirement,
     TableUpdate,
@@ -324,6 +328,20 @@ def delete_data_files(io: FileIO, manifests_to_delete: list[ManifestFile]) -> No
                 deleted_files[path] = True
 
 
+def _raise_if_view_exists(catalog: Catalog, identifier: str | Identifier) -> None:
+    """Raise TableAlreadyExistsError if a view exists at the same identifier.
+
+    Mirrors Java's `RESTSessionCatalog.replaceTransaction()` precondition. Catalogs that
+    don't support views raise `NotImplementedError` from `view_exists` — treat as "no view".
+    """
+    try:
+        view_collision = catalog.view_exists(identifier)
+    except NotImplementedError:
+        view_collision = False
+    if view_collision:
+        raise TableAlreadyExistsError(f"View with same name already exists: {identifier}")
+
+
 def _import_catalog(name: str, catalog_impl: str, properties: Properties) -> Catalog | None:
     try:
         path_parts = catalog_impl.split(".")
@@ -445,7 +463,6 @@ def create_table_if_not_exists(
         except TableAlreadyExistsError:
             return self.load_table(identifier)
 
-    @abstractmethod
     def replace_table(
         self,
         identifier: str | Identifier,
@@ -473,9 +490,12 @@ def replace_table(
 
         Raises:
             NoSuchTableError: If the table does not exist.
+            TableAlreadyExistsError: If a view exists at the same identifier.
         """
+        return self.replace_table_transaction(
+            identifier, schema, location, partition_spec, sort_order, properties
+        ).commit_transaction()
 
-    @abstractmethod
     def replace_table_transaction(
         self,
         identifier: str | Identifier,
@@ -503,7 +523,63 @@ def replace_table_transaction(
 
         Raises:
             NoSuchTableError: If the table does not exist.
+            TableAlreadyExistsError: If a view exists at the same identifier.
         """
+        raise NotImplementedError("replace_table_transaction is not supported for this catalog type")
+
+    def _replace_staged_table(
+        self,
+        identifier: str | Identifier,
+        schema: Schema | pa.Schema,
+        location: str | None,
+        partition_spec: PartitionSpec,
+        sort_order: SortOrder,
+        properties: Properties,
+    ) -> tuple[StagedTable, Schema, PartitionSpec, SortOrder, str]:
+        """Load the existing table and build fresh schema/spec/sort-order for replacement.
+
+        Mirrors the bookkeeping in `TableMetadata.buildReplacement` (iceberg-java):
+        - reuses existing field IDs by name (current schema)
+        - reuses partition field IDs by `(source, transform)` across all specs (v2+),
+          or carries forward the current spec with `VoidTransform`s (v1)
+        - reassigns sort field IDs against the fresh schema
+        - resolves `location` to the existing table's location when omitted
+
+        Returns:
+            A tuple `(staged_table, fresh_schema, fresh_partition_spec, fresh_sort_order, resolved_location)`.
+        """
+        existing_table = self.load_table(identifier)
+        existing_metadata = existing_table.metadata
+
+        resolved_format_version = int(properties.get(TableProperties.FORMAT_VERSION, existing_metadata.format_version))  # type: ignore
+        iceberg_schema = self._convert_schema_if_needed(schema, resolved_format_version)
+
+        fresh_schema, _ = assign_fresh_schema_ids_for_replace(
+            iceberg_schema, existing_metadata.schema(), existing_metadata.last_column_id
+        )
+
+        fresh_partition_spec, _ = assign_fresh_partition_spec_ids_for_replace(
+            partition_spec,
+            iceberg_schema,
+            fresh_schema,
+            existing_metadata.partition_specs,
+            existing_metadata.last_partition_id,
+            format_version=existing_metadata.format_version,
+            current_spec=existing_metadata.spec(),
+        )
+
+        fresh_sort_order = assign_fresh_sort_order_ids(sort_order, iceberg_schema, fresh_schema)
+
+        resolved_location = location.rstrip("/") if location else existing_metadata.location
+
+        staged_table = StagedTable(
+            identifier=existing_table.name(),
+            metadata=existing_metadata,
+            metadata_location=existing_table.metadata_location,
+            io=existing_table.io,
+            catalog=self,
+        )
+        return staged_table, fresh_schema, fresh_partition_spec, fresh_sort_order, resolved_location
 
     @abstractmethod
     def load_table(self, identifier: str | Identifier) -> Table:
@@ -985,18 +1061,6 @@ def create_table_transaction(
             self._create_staged_table(identifier, schema, location, partition_spec, sort_order, properties)
         )
 
-    @override
-    def replace_table(
-        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,
-    ) -> Table:
-        raise NotImplementedError("replace_table is not yet supported for this catalog type")
-
     @override
     def replace_table_transaction(
         self,
@@ -1007,7 +1071,18 @@ def replace_table_transaction(
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
     ) -> ReplaceTableTransaction:
-        raise NotImplementedError("replace_table_transaction is not yet supported for this catalog type")
+        _raise_if_view_exists(self, identifier)
+        staged_table, fresh_schema, fresh_spec, fresh_sort_order, resolved_location = self._replace_staged_table(
+            identifier, schema, location, partition_spec, sort_order, properties
+        )
+        return ReplaceTableTransaction(
+            table=staged_table,
+            new_schema=fresh_schema,
+            new_spec=fresh_spec,
+            new_sort_order=fresh_sort_order,
+            new_location=resolved_location,
+            new_properties=properties,
+        )
 
     @override
     def table_exists(self, identifier: str | Identifier) -> bool:

pyiceberg/catalog/noop.py

@@ -28,7 +28,6 @@
 from pyiceberg.table import (
     CommitTableResponse,
     CreateTableTransaction,
-    ReplaceTableTransaction,
     Table,
 )
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
@@ -69,30 +68,6 @@ def create_table_transaction(
     ) -> CreateTableTransaction:
         raise NotImplementedError
 
-    @override
-    def replace_table(
-        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,
-    ) -> Table:
-        raise NotImplementedError
-
-    @override
-    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:
-        raise NotImplementedError
-
     @override
     def load_table(self, identifier: str | Identifier) -> Table:
         raise NotImplementedError

pyiceberg/catalog/rest/__init__.py

@@ -30,7 +30,15 @@
 from typing_extensions import override
 
 from pyiceberg import __version__
-from pyiceberg.catalog import BOTOCORE_SESSION, TOKEN, URI, WAREHOUSE_LOCATION, Catalog, PropertiesUpdateSummary
+from pyiceberg.catalog import (
+    BOTOCORE_SESSION,
+    TOKEN,
+    URI,
+    WAREHOUSE_LOCATION,
+    Catalog,
+    PropertiesUpdateSummary,
+    _raise_if_view_exists,
+)
 from pyiceberg.catalog.rest.auth import AUTH_MANAGER, AuthManager, AuthManagerAdapter, AuthManagerFactory, LegacyOAuth2AuthManager
 from pyiceberg.catalog.rest.response import _handle_non_200_response
 from pyiceberg.catalog.rest.scan_planning import (
@@ -72,9 +80,8 @@
     UNPARTITIONED_PARTITION_SPEC,
     PartitionSpec,
     assign_fresh_partition_spec_ids,
-    assign_fresh_partition_spec_ids_for_replace,
 )
-from pyiceberg.schema import Schema, assign_fresh_schema_ids, assign_fresh_schema_ids_for_replace
+from pyiceberg.schema import Schema, assign_fresh_schema_ids
 from pyiceberg.table import (
     CommitTableRequest,
     CommitTableResponse,
@@ -990,43 +997,14 @@ def replace_table_transaction(
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
     ) -> ReplaceTableTransaction:
-        existing_table = self.load_table(identifier)
-        existing_metadata = existing_table.metadata
-
-        iceberg_schema = self._convert_schema_if_needed(
-            schema,
-            int(properties.get(TableProperties.FORMAT_VERSION, existing_metadata.format_version)),  # type: ignore
-        )
-
-        # Assign fresh schema IDs, reusing IDs from the existing schema by field name
-        fresh_schema, _ = assign_fresh_schema_ids_for_replace(
-            iceberg_schema, existing_metadata.schema(), existing_metadata.last_column_id
-        )
-
-        # Assign fresh partition spec IDs, reusing IDs from existing specs
-        fresh_partition_spec, _ = assign_fresh_partition_spec_ids_for_replace(
-            partition_spec, iceberg_schema, fresh_schema, existing_metadata.partition_specs, existing_metadata.last_partition_id
+        _raise_if_view_exists(self, identifier)
+        staged_table, fresh_schema, fresh_spec, fresh_sort_order, resolved_location = self._replace_staged_table(
+            identifier, schema, location, partition_spec, sort_order, properties
         )
-
-        # Assign fresh sort order IDs
-        fresh_sort_order = assign_fresh_sort_order_ids(sort_order, iceberg_schema, fresh_schema)
-
-        # Use existing location if not specified
-        resolved_location = location.rstrip("/") if location else existing_metadata.location
-
-        # Create a StagedTable from the existing table
-        staged_table = StagedTable(
-            identifier=existing_table.name(),
-            metadata=existing_metadata,
-            metadata_location=existing_table.metadata_location,
-            io=existing_table.io,
-            catalog=self,
-        )
-
         return ReplaceTableTransaction(
             table=staged_table,
             new_schema=fresh_schema,
-            new_spec=fresh_partition_spec,
+            new_spec=fresh_spec,
             new_sort_order=fresh_sort_order,
             new_location=resolved_location,
             new_properties=properties,