Consolidate replace integration tests into single RTAS path

The DDL-only integration test only added one unique invariant beyond
the RTAS test: that current_snapshot_id is None after a replace with
no follow-up append. That invariant is already covered at the
catalog-behavior layer (test_replace_transaction) against
InMemoryCatalog + SqlCatalog. Backends don't differentiate "DDL-only
commit" from "DDL+write commit" at the protocol level, so a separate
integration variant adds little signal.

Fold the UUID-preservation assertion (which the RTAS test wasn't
checking explicitly) into the consolidated test.

Author: smaheshwar-pltr

tests/integration/test_catalog.py

@@ -44,7 +44,7 @@
 from pyiceberg.table.metadata import INITIAL_SPEC_ID
 from pyiceberg.table.sorting import INITIAL_SORT_ORDER_ID, SortField, SortOrder
 from pyiceberg.transforms import BucketTransform, DayTransform, IdentityTransform
-from pyiceberg.types import BooleanType, IntegerType, LongType, NestedField, StringType, TimestampType, UUIDType
+from pyiceberg.types import IntegerType, LongType, NestedField, TimestampType, UUIDType
 from tests.conftest import (
     clean_up,
     does_support_atomic_concurrent_updates,
@@ -877,43 +877,6 @@ def test_load_missing_table(test_catalog: Catalog, database_name: str, table_nam
         test_catalog.load_table(identifier)
 
 
-@pytest.mark.integration
-@pytest.mark.parametrize("test_catalog", CATALOGS)
-def test_replace_table(test_catalog: Catalog, database_name: str, table_name: str) -> None:
-    test_catalog.create_namespace(database_name)
-    identifier = (database_name, table_name)
-
-    original_schema = Schema(
-        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
-        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
-    )
-    original = test_catalog.create_table(identifier, schema=original_schema)
-    original.append(
-        pa.Table.from_pydict(
-            {"id": [1, 2, 3], "data": ["a", "b", "c"]},
-            schema=pa.schema([pa.field("id", pa.int64()), pa.field("data", pa.large_string())]),
-        )
-    )
-    original.refresh()
-    original_snapshot_id = original.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),
-        NestedField(field_id=3, name="active", field_type=BooleanType(), required=False),
-    )
-    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
-    assert replaced.current_snapshot() is None
-    assert any(s.snapshot_id == original_snapshot_id for s in replaced.metadata.snapshots)
-    # Time-travel back to the pre-replace snapshot returns the rows that were there before.
-    time_travel = replaced.scan(snapshot_id=original_snapshot_id).to_arrow()
-    assert time_travel.num_rows == 3
-    assert sorted(time_travel.column("id").to_pylist()) == [1, 2, 3]
-
-
 @pytest.mark.integration
 @pytest.mark.parametrize("test_catalog", CATALOGS)
 def test_replace_table_transaction(test_catalog: Catalog, database_name: str, table_name: str) -> None:
@@ -936,6 +899,7 @@ def test_replace_table_transaction(test_catalog: Catalog, database_name: str, ta
         txn.append(new_data)
 
     replaced = test_catalog.load_table(identifier)
+    assert replaced.metadata.table_uuid == original.metadata.table_uuid
     assert replaced.current_snapshot() is not None
     assert replaced.current_snapshot().snapshot_id != old_snapshot_id  # type: ignore[union-attr]
     assert any(s.snapshot_id == old_snapshot_id for s in replaced.metadata.snapshots)