Address review feedback (round 2)

- Drop view-collision pre-flight check; deferring to a follow-up PR
  that ports the check across create/replace/rename in one pass.
- Simplify RTAS tests + docs to use txn.append(df) instead of the
  verbose update_snapshot().fast_append() + _dataframe_to_data_files
  pattern.
- Drop parallel new_schema construction in RTAS tests; pass df.schema
  directly (replace_table* accepts Schema | pa.Schema).
- Reword docs intro per reviewer suggestion.
- Run formatter / linter; trailing whitespace and trailing newline fix.
- mypy: cast format_version to TableVersion; extract inner with into
  helper to silence unreachable warning under pytest.raises.

Author: smaheshwar-pltr

mkdocs/docs/api.md

@@ -187,41 +187,28 @@ 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). Use this when you want to redefine the table's metadata; pair it with `replace_table_transaction` to atomically write new data alongside the metadata change (RTAS-style).
+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,
-)
+catalog.replace_table(identifier="docs_example.bids", schema=df.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`.
+Where `df` is a PyArrow table (or `Schema`) carrying the new column set. 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`.
 
 Properties passed to `replace_table` are **merged** with the existing table properties (your values override; existing keys you don't pass are preserved). To remove a property as part of the replace, use `replace_table_transaction` and remove 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:
 
 ```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)
+with catalog.replace_table_transaction(identifier="docs_example.bids", schema=df.schema) as txn:
+    txn.append(df)
     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"})
+catalog.replace_table(identifier="docs_example.bids", schema=df.schema, properties={"format-version": "2"})
 ```
 
 ## Register a table

pyiceberg/catalog/__init__.py

@@ -328,20 +328,6 @@ 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.
-
-    Catalogs that don't support views raise `NotImplementedError` from `view_exists` —
-    treat as "no view" in that case.
-    """
-    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(".")
@@ -564,7 +550,7 @@ def _replace_staged_table(
         resolved_format_version = (
             int(requested_format_version) if requested_format_version is not None else existing_metadata.format_version
         )
-        iceberg_schema = self._convert_schema_if_needed(schema, resolved_format_version)
+        iceberg_schema = self._convert_schema_if_needed(schema, cast(TableVersion, resolved_format_version))
 
         fresh_schema, _ = assign_fresh_schema_ids_for_replace(
             iceberg_schema, existing_metadata.schema(), existing_metadata.last_column_id
@@ -1083,7 +1069,6 @@ def replace_table_transaction(
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
     ) -> ReplaceTableTransaction:
-        _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
         )

pyiceberg/catalog/rest/__init__.py

@@ -30,15 +30,7 @@
 from typing_extensions import override
 
 from pyiceberg import __version__
-from pyiceberg.catalog import (
-    BOTOCORE_SESSION,
-    TOKEN,
-    URI,
-    WAREHOUSE_LOCATION,
-    Catalog,
-    PropertiesUpdateSummary,
-    _raise_if_view_exists,
-)
+from pyiceberg.catalog import BOTOCORE_SESSION, TOKEN, URI, WAREHOUSE_LOCATION, Catalog, PropertiesUpdateSummary
 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 (
@@ -977,7 +969,6 @@ def replace_table_transaction(
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
     ) -> ReplaceTableTransaction:
-        _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
         )

tests/catalog/test_catalog_behaviors.py

@@ -419,9 +419,7 @@ def _simple_data(num_rows: int = 2) -> pa.Table:
     )
 
 
-def test_replace_table_preserves_uuid_and_clears_current_snapshot(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_preserves_uuid_and_clears_current_snapshot(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """Replace preserves table_uuid and snapshot history, but clears the main ref."""
     _create_simple_table(catalog, test_table_identifier)
     original = catalog.load_table(test_table_identifier)
@@ -553,9 +551,7 @@ def test_replace_table_merges_properties_with_overrides_and_additions(
     """Properties are merged onto existing: new values override, untouched keys are preserved."""
     schema = Schema(NestedField(field_id=1, name="id", field_type=LongType(), required=False))
     _create_simple_table(catalog, test_table_identifier, schema=schema, properties={"keep": "yes", "override": "old"})
-    replaced = catalog.replace_table(
-        test_table_identifier, schema=schema, properties={"override": "new", "new_key": "v"}
-    )
+    replaced = catalog.replace_table(test_table_identifier, schema=schema, properties={"override": "new", "new_key": "v"})
     assert replaced.properties["keep"] == "yes"
     assert replaced.properties["override"] == "new"
     assert replaced.properties["new_key"] == "v"
@@ -569,19 +565,15 @@ def test_replace_table_upgrades_format_version(catalog: Catalog, test_table_iden
     assert replaced.format_version == 2
 
 
-def test_replace_table_rejects_format_version_downgrade(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_rejects_format_version_downgrade(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """A `format-version` lower than the existing one must be rejected to avoid silently
     running the schema-conversion path with the wrong semantics."""
     _, 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"})
 
 
-def test_replace_table_v1_carries_forward_partition_fields_as_void(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_v1_carries_forward_partition_fields_as_void(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """v1 specs are append-only; dropped partition fields must be carried forward as VoidTransform."""
     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=1)
@@ -602,9 +594,7 @@ def test_replace_table_raises_when_table_does_not_exist(catalog: Catalog, test_t
         catalog.replace_table(test_table_identifier, schema=schema)
 
 
-def test_replace_table_transaction_can_stage_additional_changes(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_transaction_can_stage_additional_changes(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """The transaction context lets callers stage extra updates (e.g. properties) before commit."""
     _, schema = _create_simple_table(catalog, test_table_identifier)
     with catalog.replace_table_transaction(test_table_identifier, schema=schema) as txn:
@@ -613,9 +603,7 @@ def test_replace_table_transaction_can_stage_additional_changes(
     assert replaced.properties.get("staged") == "yes"
 
 
-def test_replace_table_transaction_with_write_atomic_rtas(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_transaction_with_write_atomic_rtas(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """RTAS (Replace Table As Select): replace + write new data in one transaction.
 
     The new schema and new data must land atomically. After commit:
@@ -626,18 +614,12 @@ def test_replace_table_transaction_with_write_atomic_rtas(
     catalog.load_table(test_table_identifier).append(_simple_data(num_rows=1))
     old_snapshot_id = catalog.load_table(test_table_identifier).current_snapshot().snapshot_id  # type: ignore[union-attr]
 
-    new_schema = Schema(
-        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
-        NestedField(field_id=2, name="name", field_type=StringType(), required=False),
-    )
     new_data = pa.Table.from_pydict(
         {"id": [10, 20], "name": ["alice", "bob"]},
         schema=pa.schema([pa.field("id", pa.int64()), pa.field("name", pa.large_string())]),
     )
-    with catalog.replace_table_transaction(test_table_identifier, 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=new_data, io=txn._table.io):
-                snap.append_data_file(data_file)
+    with catalog.replace_table_transaction(test_table_identifier, schema=new_data.schema) as txn:
+        txn.append(new_data)
 
     replaced = catalog.load_table(test_table_identifier)
     new_snapshot = replaced.current_snapshot()
@@ -649,9 +631,7 @@ def test_replace_table_transaction_with_write_atomic_rtas(
     assert replaced.scan().to_arrow().num_rows == 2
 
 
-def test_replace_table_transaction_rolls_back_on_failure(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_transaction_rolls_back_on_failure(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """If the body of the transaction raises, no metadata change is committed.
 
     The table's UUID, schemas, current snapshot, and current schema id must all be unchanged."""
@@ -664,10 +644,14 @@ def test_replace_table_transaction_rolls_back_on_failure(
         NestedField(field_id=2, name="data", field_type=StringType(), required=False),
         NestedField(field_id=3, name="extra", field_type=BooleanType(), required=False),
     )
-    with pytest.raises(RuntimeError, match="simulated failure inside replace transaction"):
+
+    def run_failing_replace() -> None:
         with catalog.replace_table_transaction(test_table_identifier, schema=new_schema):
             raise RuntimeError("simulated failure inside replace transaction")
 
+    with pytest.raises(RuntimeError, match="simulated failure inside replace transaction"):
+        run_failing_replace()
+
     after = catalog.load_table(test_table_identifier).metadata
     assert after.table_uuid == before.table_uuid
     assert after.current_snapshot_id == before.current_snapshot_id
@@ -696,9 +680,7 @@ def test_concurrent_replace_table(catalog: Catalog, test_table_identifier: Ident
         txn_b.commit_transaction()
 
 
-def test_replace_table_allows_subsequent_append(
-    catalog: Catalog, test_table_identifier: Identifier
-) -> None:
+def test_replace_table_allows_subsequent_append(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """After `replace_table` clears the current snapshot, a separate `append` produces a new
     snapshot containing only the post-replace data — the pre-replace rows are not visible."""
     _, schema = _create_simple_table(catalog, test_table_identifier)

tests/catalog/test_rest.py

@@ -2906,7 +2906,7 @@ def test_load_table_without_storage_credentials(
 # Catalog-agnostic behavior (UUID preservation, snapshot clearing, schema/spec/sort-order reuse,
 # format-version upgrade, etc.) is covered in tests/catalog/test_catalog_behaviors.py against
 # both InMemoryCatalog and SqlCatalog. The tests below cover REST-specific concerns: the wire
-# payload sent to the server, 404 handling, and the view-collision pre-flight check.
+# payload sent to the server and 404 handling.
 
 
 def _mock_replace_endpoints(
@@ -2915,13 +2915,7 @@ def _mock_replace_endpoints(
     table: str,
     load_response: dict[str, Any],
     commit_response: dict[str, Any],
-    view_exists: bool = False,
 ) -> None:
-    rest_mock.head(
-        f"{TEST_URI}v1/namespaces/{namespace}/views/{table}",
-        status_code=204 if view_exists else 404,
-        request_headers=TEST_HEADERS,
-    )
     rest_mock.get(
         f"{TEST_URI}v1/namespaces/{namespace}/tables/{table}",
         json=load_response,
@@ -3009,28 +3003,6 @@ def test_replace_table_transaction_404_raises(
         )
 
 
-def test_replace_table_transaction_rejects_view_collision(
-    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:
-    """A view at the same identifier must fail replace."""
-    _mock_replace_endpoints(
-        rest_mock,
-        "fokko",
-        "fokko2",
-        example_table_metadata_with_snapshot_v1_rest_json,
-        example_table_metadata_no_snapshot_v1_rest_json,
-        view_exists=True,
-    )
-    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
-    with pytest.raises(TableAlreadyExistsError, match="View with same name already exists"):
-        catalog.replace_table_transaction(
-            identifier=("fokko", "fokko2"),
-            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],
@@ -3064,5 +3036,3 @@ def test_replace_table_issues_commit_post_immediately(
     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"
-
-