Drop Catalog.replace_table shortcut, keep replace_table_transaction only

Mirrors Java, which exposes newReplaceTableTransaction only and has no
top-level Catalog.replaceTable. Callers go through the transaction
explicitly and follow up with load_table() when a fresh handle is
needed, sidestepping the StagedTable return-type wart on the shortcut.

Also collapse the docs section to the single RTAS-flow example.

Author: smaheshwar-pltr

mkdocs/docs/api.md

@@ -187,36 +187,18 @@ with catalog.create_table_transaction(identifier="docs_example.bids", schema=sch
 
 ## 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). `replace_table` redefines the table in this way; `replace_table_transaction` lets you write new data alongside this change to permit RTAS (replace-table-as-select) workflows.
-
-```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 are reused by name from the previous schema; new columns get fresh IDs above `last-column-id`.
-
-Unlike the other fields, table properties are *merged* on replace: properties you don't pass are preserved on the table. To remove a property as part of the replace, use `replace_table_transaction` and drop it explicitly within the transaction.
-
-Use `replace_table_transaction` to stage additional changes (writes, property updates, schema evolution) before committing — for example, swap the schema and write new data atomically:
+Atomically replace an existing table's schema, partition spec, sort order, location, and properties via `replace_table_transaction`. 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). Open the transaction with the new definition, stage any additional changes (writes, property updates, schema evolution), and commit — for example, an RTAS (replace-table-as-select) that swaps the schema and writes the new data atomically:
 
 ```python
 with catalog.replace_table_transaction(identifier="docs_example.bids", schema=df.schema) as txn:
     txn.append(df)
 ```
 
-To upgrade the table's format version as part of the replace, pass `format-version` in `properties`:
+Field IDs are reused by name from the previous schema; new columns get fresh IDs above `last-column-id`.
 
-```python
-catalog.replace_table(identifier="docs_example.bids", schema=new_schema, properties={"format-version": "2"})
-```
+Table properties are *merged* on replace: properties you don't pass are preserved on the table. To remove a property, drop it explicitly within the transaction.
+
+Pass `format-version` in `properties` to upgrade the table's format version as part of the replace.
 
 ## Register a table
 

pyiceberg/catalog/__init__.py

@@ -449,41 +449,6 @@ def create_table_if_not_exists(
         except TableAlreadyExistsError:
             return self.load_table(identifier)
 
-    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:
-        """Atomically replace a table's schema, spec, sort order, location, and properties.
-
-        The table UUID and history (snapshots, schemas, specs, sort orders) are preserved.
-        The current snapshot is cleared (main branch ref is removed).
-
-        Args:
-            identifier (str | Identifier): Table identifier.
-            schema (Schema): New table schema.
-            location (str | None): New table location. Defaults to the existing location.
-            partition_spec (PartitionSpec): New partition spec.
-            sort_order (SortOrder): New sort order.
-            properties (Properties): Properties to apply. Merged on top of the existing
-                table properties: keys present here override existing values; existing keys
-                not present here are preserved. To remove a property, follow up with a
-                transaction that removes it explicitly.
-
-        Returns:
-            Table: the replaced table instance.
-
-        Raises:
-            NoSuchTableError: If the table does not exist.
-        """
-        return self.replace_table_transaction(
-            identifier, schema, location, partition_spec, sort_order, properties
-        ).commit_transaction()
-
     @abstractmethod
     def replace_table_transaction(
         self,

tests/catalog/test_catalog_behaviors.py

@@ -432,7 +432,7 @@ def test_replace_transaction(catalog: Catalog, test_table_identifier: Identifier
     snapshot_log_before = list(original.metadata.snapshot_log)
     assert len(snapshot_log_before) == 1
 
-    catalog.replace_table(test_table_identifier, schema=_REPLACE_SCHEMA)
+    catalog.replace_table_transaction(test_table_identifier, schema=_REPLACE_SCHEMA).commit_transaction()
     replaced = catalog.load_table(test_table_identifier)
 
     # UUID + history preserved, current snapshot cleared, current schema swapped.
@@ -518,12 +518,13 @@ def test_complete_replace_transaction(catalog: Catalog, test_table_identifier: I
 def test_replace_transaction_requires_table_exists(catalog: Catalog, test_table_identifier: Identifier) -> None:
     schema = Schema(NestedField(field_id=1, name="id", field_type=LongType(), required=False))
     with pytest.raises(NoSuchTableError):
-        catalog.replace_table(test_table_identifier, schema=schema)
+        catalog.replace_table_transaction(test_table_identifier, schema=schema)
 
 
 def test_replace_table_reuses_schema_id_when_identical(catalog: Catalog, test_table_identifier: Identifier) -> None:
     _, base_schema = _create_simple_table(catalog, test_table_identifier)
-    replaced = catalog.replace_table(test_table_identifier, schema=base_schema)
+    catalog.replace_table_transaction(test_table_identifier, schema=base_schema).commit_transaction()
+    replaced = catalog.load_table(test_table_identifier)
     # Identical shape -> no new schema appended, current points back at id 0.
     assert [s.schema_id for s in replaced.metadata.schemas] == [0]
     assert replaced.metadata.current_schema_id == 0
@@ -537,17 +538,24 @@ def test_replace_table_reuses_partition_spec_and_sort_order_when_identical(
     sort = SortOrder(SortField(source_id=1, transform=IdentityTransform(), direction=SortDirection.ASC))
     _, schema = _create_simple_table(catalog, test_table_identifier, partition_spec=spec)
     # Introduce a sort order then replay both spec and sort — neither should append a new entry.
-    sorted_first = catalog.replace_table(test_table_identifier, schema=schema, partition_spec=spec, sort_order=sort)
+    catalog.replace_table_transaction(
+        test_table_identifier, schema=schema, partition_spec=spec, sort_order=sort
+    ).commit_transaction()
+    sorted_first = catalog.load_table(test_table_identifier)
     sorted_order_id = sorted_first.metadata.default_sort_order_id
     assert sorted_order_id != 0
 
-    replayed = catalog.replace_table(test_table_identifier, schema=schema, partition_spec=spec, sort_order=sort)
+    catalog.replace_table_transaction(
+        test_table_identifier, schema=schema, partition_spec=spec, sort_order=sort
+    ).commit_transaction()
+    replayed = catalog.load_table(test_table_identifier)
     assert [s.spec_id for s in replayed.metadata.partition_specs] == [0]
     assert replayed.metadata.default_spec_id == 0
     assert replayed.metadata.default_sort_order_id == sorted_order_id
 
     # Dropping the sort order falls back to the unsorted order_id 0 (also reused, not appended).
-    unsorted = catalog.replace_table(test_table_identifier, schema=schema, partition_spec=spec)
+    catalog.replace_table_transaction(test_table_identifier, schema=schema, partition_spec=spec).commit_transaction()
+    unsorted = catalog.load_table(test_table_identifier)
     assert unsorted.sort_order().is_unsorted
     assert unsorted.metadata.default_sort_order_id == 0
 
@@ -573,7 +581,8 @@ def test_replace_table_identifier_field_ids(catalog: Catalog, test_table_identif
             NestedField(field_id=2, name="data", field_type=StringType(), required=False),
         )
     )
-    replaced = catalog.replace_table(test_table_identifier, schema=new_schema)
+    catalog.replace_table_transaction(test_table_identifier, schema=new_schema).commit_transaction()
+    replaced = catalog.load_table(test_table_identifier)
     expected = [1] if keep_identifier else []
     assert list(replaced.schema().identifier_field_ids) == expected
 
@@ -591,7 +600,8 @@ def test_replace_table_partition_field_carry_forward(
 ) -> None:
     spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, name="id_part", transform=IdentityTransform()))
     _, schema = _create_simple_table(catalog, test_table_identifier, partition_spec=spec, format_version=format_version)
-    replaced = catalog.replace_table(test_table_identifier, schema=schema)
+    catalog.replace_table_transaction(test_table_identifier, schema=schema).commit_transaction()
+    replaced = catalog.load_table(test_table_identifier)
     new_spec = replaced.spec()
     if expect_void_carry_forward:
         void_field = next(f for f in new_spec.fields if f.field_id == 1000)
@@ -606,22 +616,26 @@ def test_replace_table_upgrades_format_version(catalog: Catalog, test_table_iden
     _, schema = _create_simple_table(catalog, test_table_identifier, format_version=1)
     assert catalog.load_table(test_table_identifier).format_version == 1
 
-    upgraded = catalog.replace_table(test_table_identifier, schema=schema, properties={"format-version": "2"})
+    catalog.replace_table_transaction(
+        test_table_identifier, schema=schema, properties={"format-version": "2"}
+    ).commit_transaction()
+    upgraded = catalog.load_table(test_table_identifier)
     assert upgraded.format_version == 2
     # `format-version` is a control input, not a persisted property.
     assert "format-version" not in upgraded.properties
 
     # A follow-up replace stays at the upgraded version.
     new_schema = Schema(*schema.fields, NestedField(field_id=3, name="extra", field_type=BooleanType(), required=False))
-    replayed = catalog.replace_table(test_table_identifier, schema=new_schema)
+    catalog.replace_table_transaction(test_table_identifier, schema=new_schema).commit_transaction()
+    replayed = catalog.load_table(test_table_identifier)
     assert replayed.format_version == 2
     assert {f.name for f in replayed.schema().fields} == {"id", "data", "extra"}
 
 
 def test_replace_table_rejects_format_version_downgrade(catalog: Catalog, test_table_identifier: Identifier) -> None:
     _, schema = _create_simple_table(catalog, test_table_identifier, format_version=2)
     with pytest.raises(ValueError, match="Cannot downgrade format-version"):
-        catalog.replace_table(test_table_identifier, schema=schema, properties={"format-version": "1"})
+        catalog.replace_table_transaction(test_table_identifier, schema=schema, properties={"format-version": "1"})
 
 
 @pytest.mark.parametrize("location_kind", ["inherit", "explicit", "trailing-slash"])
@@ -630,12 +644,14 @@ def test_replace_table_location(catalog: Catalog, test_table_identifier: Identif
     existing = catalog.load_table(test_table_identifier).metadata.location
 
     if location_kind == "inherit":
-        replaced = catalog.replace_table(test_table_identifier, schema=schema)
+        catalog.replace_table_transaction(test_table_identifier, schema=schema).commit_transaction()
+        replaced = catalog.load_table(test_table_identifier)
         assert replaced.metadata.location == existing
     else:
         bare = f"file://{tmp_path}/relocated"
         arg = bare + "/" if location_kind == "trailing-slash" else bare
-        replaced = catalog.replace_table(test_table_identifier, schema=schema, location=arg)
+        catalog.replace_table_transaction(test_table_identifier, schema=schema, location=arg).commit_transaction()
+        replaced = catalog.load_table(test_table_identifier)
         assert replaced.metadata.location == bare
 
 

tests/catalog/test_rest.py

@@ -19,7 +19,6 @@
 
 import base64
 import os
-import uuid
 from collections.abc import Callable
 from typing import Any, cast
 from unittest import mock
@@ -2981,34 +2980,3 @@ def test_replace_table_transaction_404_raises(
             identifier=("fokko", "missing"),
             schema=Schema(NestedField(field_id=1, name="id", field_type=IntegerType(), required=False)),
         )
-
-
-def test_replace_table_issues_commit_post_immediately(
-    rest_mock: Mocker,
-    example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any],
-    example_table_metadata_no_snapshot_v1_rest_json: dict[str, Any],
-) -> None:
-    """`replace_table` commits during the call; `replace_table_transaction` defers the POST until the caller commits."""
-    table_uuid = example_table_metadata_with_snapshot_v1_rest_json["metadata"]["table-uuid"]
-    example_table_metadata_no_snapshot_v1_rest_json["metadata"]["table-uuid"] = table_uuid
-    _mock_replace_endpoints(
-        rest_mock,
-        "fokko",
-        "fokko2",
-        example_table_metadata_with_snapshot_v1_rest_json,
-        example_table_metadata_no_snapshot_v1_rest_json,
-    )
-    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
-    new_schema = Schema(
-        NestedField(field_id=1, name="id", field_type=IntegerType(), required=False),
-        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
-    )
-
-    catalog.replace_table_transaction(identifier=("fokko", "fokko2"), schema=new_schema)
-    methods_after_open = [r.method for r in rest_mock.request_history]
-    assert "POST" not in methods_after_open
-
-    replaced = catalog.replace_table(identifier=("fokko", "fokko2"), schema=new_schema)
-    assert replaced.metadata.table_uuid == uuid.UUID(table_uuid)
-    methods_after_replace = [r.method for r in rest_mock.request_history]
-    assert "POST" in methods_after_replace, "replace_table must commit immediately"

tests/integration/test_catalog.py

@@ -902,7 +902,7 @@ def test_replace_table(test_catalog: Catalog, database_name: str, table_name: st
         NestedField(field_id=2, name="name", field_type=StringType(), required=False),
         NestedField(field_id=3, name="active", field_type=BooleanType(), required=False),
     )
-    test_catalog.replace_table(identifier, schema=new_schema)
+    test_catalog.replace_table_transaction(identifier, schema=new_schema).commit_transaction()
     replaced = test_catalog.load_table(identifier)
 
     assert replaced.metadata.table_uuid == original.metadata.table_uuid