Clean up replace test suite

- test_replace_table_location split into three named tests
  (_inherits_existing_location, _uses_explicit_location,
  _strips_trailing_slash_from_location) — kills the in-body branching
  on the parametrize id.

- _run_complete_replace returns a @DataClass _ReplaceFixture instead
  of a 5-tuple; callers name what they care about (fx.replaced etc.)
  instead of writing _, replaced, _, _, _.

- snapshot_log assertion tightened from unordered membership to
  ordered prefix equality.

- _rejects_format_version_downgrade / _rejects_non_numeric_format_version
  / _rejects_empty_location merged into one parametrized
  test_replace_table_rejects_invalid_inputs.

- test_replace_table_upgrades_format_version split: the follow-up
  "stays at v2" half is now
  test_replace_table_keeps_upgraded_format_version_on_subsequent_replace.

- concurrent-conflict tests now assert that last_column_id /
  last_partition_id did NOT advance after the failed commit, proving
  the conflict was a true no-op.

- multi-suffix void test: 4-field schema trimmed to 3, columns
  renamed to name their role (current_source / collide_on_name /
  collide_on_fallback).

Author: smaheshwar-pltr

tests/catalog/test_catalog_behaviors.py

@@ -20,6 +20,7 @@
 """
 
 import os
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -439,9 +440,10 @@ def test_replace_transaction(catalog: Catalog, test_table_identifier: Identifier
     assert replaced.metadata.table_uuid == original.metadata.table_uuid
     assert replaced.metadata.current_snapshot_id is None
     assert {f.name for f in replaced.schema().fields} == {"id", "data", "extra"}
-    # Old snapshot kept by identity (not just count), and snapshot_log entries from before survive.
+    # Old snapshot kept by identity (not just count), and snapshot_log entries from before survive
+    # in order at the front of the log.
     assert any(s.snapshot_id == old_snapshot_id for s in replaced.metadata.snapshots)
-    assert all(entry in replaced.metadata.snapshot_log for entry in snapshot_log_before)
+    assert replaced.metadata.snapshot_log[: len(snapshot_log_before)] == snapshot_log_before
     # Old schema is still in the schemas list alongside the new one.
     schema_ids = sorted(s.schema_id for s in replaced.metadata.schemas)
     assert schema_ids == [0, 1]
@@ -450,14 +452,20 @@ def test_replace_transaction(catalog: Catalog, test_table_identifier: Identifier
     assert replaced.scan(snapshot_id=old_snapshot_id).to_arrow().equals(_simple_data())
 
 
-def _run_complete_replace(
-    catalog: Catalog, identifier: Identifier, tmp_path: Path
-) -> tuple[Table, Table, SortOrder, pa.Table, int]:
-    """Set up a table, run a full-six-args RTAS replace, and return the handles needed for assertions.
+@dataclass
+class _ReplaceFixture:
+    """State produced by `_run_complete_replace`: the table before/after the replace plus
+    the inputs needed to assert on the result."""
 
-    Returns:
-        (original, replaced, new_sort, original_data, old_snapshot_id)
-    """
+    original: Table
+    replaced: Table
+    new_sort: SortOrder
+    original_data: pa.Table
+    old_snapshot_id: int
+
+
+def _run_complete_replace(catalog: Catalog, identifier: Identifier, tmp_path: Path) -> _ReplaceFixture:
+    """Set up a table, run a full-six-args RTAS replace, and return the handles needed for assertions."""
     _create_simple_table(catalog, identifier, properties={"keep": "yes", "override": "old"})
     catalog.load_table(identifier).append(_simple_data())
     original = catalog.load_table(identifier)
@@ -487,52 +495,58 @@ def _run_complete_replace(
     ) as txn:
         txn.append(new_data)
 
-    return original, catalog.load_table(identifier), new_sort, original_data, old_snapshot_id
+    return _ReplaceFixture(
+        original=original,
+        replaced=catalog.load_table(identifier),
+        new_sort=new_sort,
+        original_data=original_data,
+        old_snapshot_id=old_snapshot_id,
+    )
 
 
 def test_complete_replace_transaction_applies_new_schema_spec_and_sort(
     catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path
 ) -> None:
-    original, replaced, new_sort, _, _ = _run_complete_replace(catalog, test_table_identifier, tmp_path)
+    fx = _run_complete_replace(catalog, test_table_identifier, tmp_path)
     # Identity invariants.
-    assert replaced.metadata.table_uuid == original.metadata.table_uuid
-    assert replaced.metadata.location == f"file://{tmp_path}/replaced"
+    assert fx.replaced.metadata.table_uuid == fx.original.metadata.table_uuid
+    assert fx.replaced.metadata.location == f"file://{tmp_path}/replaced"
     # New schema / spec / sort applied; old entries retained in history.
-    assert {f.name for f in replaced.schema().fields} == {"id", "data", "extra"}
-    assert sorted(s.schema_id for s in replaced.metadata.schemas) == [0, 1]
-    assert replaced.spec().fields[0].source_id == 1
-    assert isinstance(replaced.spec().fields[0].transform, IdentityTransform)
-    assert {s.spec_id for s in replaced.metadata.partition_specs} == {0, 1}
-    assert replaced.sort_order().fields == new_sort.fields
-    assert {s.order_id for s in replaced.metadata.sort_orders} == {0, replaced.metadata.default_sort_order_id}
+    assert {f.name for f in fx.replaced.schema().fields} == {"id", "data", "extra"}
+    assert sorted(s.schema_id for s in fx.replaced.metadata.schemas) == [0, 1]
+    assert fx.replaced.spec().fields[0].source_id == 1
+    assert isinstance(fx.replaced.spec().fields[0].transform, IdentityTransform)
+    assert {s.spec_id for s in fx.replaced.metadata.partition_specs} == {0, 1}
+    assert fx.replaced.sort_order().fields == fx.new_sort.fields
+    assert {s.order_id for s in fx.replaced.metadata.sort_orders} == {0, fx.replaced.metadata.default_sort_order_id}
 
 
 def test_complete_replace_transaction_merges_properties(
     catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path
 ) -> None:
-    _, replaced, _, _, _ = _run_complete_replace(catalog, test_table_identifier, tmp_path)
+    fx = _run_complete_replace(catalog, test_table_identifier, tmp_path)
     # `keep` is preserved, `override` is updated, `added` is new, and `format-version` does not leak.
-    assert replaced.properties["keep"] == "yes"
-    assert replaced.properties["override"] == "new"
-    assert replaced.properties["added"] == "v"
-    assert "format-version" not in replaced.properties
+    assert fx.replaced.properties["keep"] == "yes"
+    assert fx.replaced.properties["override"] == "new"
+    assert fx.replaced.properties["added"] == "v"
+    assert "format-version" not in fx.replaced.properties
 
 
 def test_complete_replace_transaction_rtas_preserves_old_snapshot(
     catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path
 ) -> None:
-    _, replaced, _, original_data, old_snapshot_id = _run_complete_replace(catalog, test_table_identifier, tmp_path)
+    fx = _run_complete_replace(catalog, test_table_identifier, tmp_path)
     # New snapshot exists, has no parent (fresh start), old snapshot is still in the snapshot list.
-    new_snapshot = replaced.current_snapshot()
+    new_snapshot = fx.replaced.current_snapshot()
     assert new_snapshot is not None
-    assert new_snapshot.snapshot_id != old_snapshot_id
+    assert new_snapshot.snapshot_id != fx.old_snapshot_id
     assert new_snapshot.parent_snapshot_id is None
-    assert any(s.snapshot_id == old_snapshot_id for s in replaced.metadata.snapshots)
-    assert replaced.scan().to_arrow().num_rows == 2
+    assert any(s.snapshot_id == fx.old_snapshot_id for s in fx.replaced.metadata.snapshots)
+    assert fx.replaced.scan().to_arrow().num_rows == 2
     # Time-travel back to before the replace returns the original rows from the old schema.
-    time_travel = replaced.scan(snapshot_id=old_snapshot_id).to_arrow()
-    assert time_travel.num_rows == original_data.num_rows
-    assert time_travel.column("id").to_pylist() == original_data.column("id").to_pylist()
+    time_travel = fx.replaced.scan(snapshot_id=fx.old_snapshot_id).to_arrow()
+    assert time_travel.num_rows == fx.original_data.num_rows
+    assert time_travel.column("id").to_pylist() == fx.original_data.column("id").to_pylist()
 
 
 def test_replace_transaction_requires_table_exists(catalog: Catalog, test_table_identifier: Identifier) -> None:
@@ -644,47 +658,62 @@ def test_replace_table_upgrades_format_version(catalog: Catalog, test_table_iden
     # `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.
+
+def test_replace_table_keeps_upgraded_format_version_on_subsequent_replace(
+    catalog: Catalog, test_table_identifier: Identifier
+) -> None:
+    _, schema = _create_simple_table(catalog, test_table_identifier, format_version=1)
+    catalog.replace_table_transaction(
+        test_table_identifier, schema=schema, properties={"format-version": "2"}
+    ).commit_transaction()
     new_schema = Schema(*schema.fields, NestedField(field_id=3, name="extra", field_type=BooleanType(), required=False))
     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:
+@pytest.mark.parametrize(
+    "properties, location, expected_match",
+    [
+        pytest.param({"format-version": "1"}, None, "Cannot downgrade format-version", id="format-version-downgrade"),
+        pytest.param({"format-version": "two"}, None, "Invalid format-version property", id="non-numeric-format-version"),
+        pytest.param({}, "/", "location must not be empty", id="empty-location-after-rstrip"),
+    ],
+)
+def test_replace_table_rejects_invalid_inputs(
+    catalog: Catalog,
+    test_table_identifier: Identifier,
+    properties: dict[str, str],
+    location: str | None,
+    expected_match: str,
+) -> None:
     _, schema = _create_simple_table(catalog, test_table_identifier, format_version=2)
-    with pytest.raises(ValueError, match="Cannot downgrade format-version"):
-        catalog.replace_table_transaction(test_table_identifier, schema=schema, properties={"format-version": "1"})
+    with pytest.raises(ValueError, match=expected_match):
+        catalog.replace_table_transaction(test_table_identifier, schema=schema, properties=properties, location=location)
 
 
-def test_replace_table_rejects_non_numeric_format_version(catalog: Catalog, test_table_identifier: Identifier) -> None:
+def test_replace_table_inherits_existing_location(catalog: Catalog, test_table_identifier: Identifier) -> None:
     _, schema = _create_simple_table(catalog, test_table_identifier)
-    with pytest.raises(ValueError, match="Invalid format-version property"):
-        catalog.replace_table_transaction(test_table_identifier, schema=schema, properties={"format-version": "two"})
+    existing = catalog.load_table(test_table_identifier).metadata.location
+    catalog.replace_table_transaction(test_table_identifier, schema=schema).commit_transaction()
+    assert catalog.load_table(test_table_identifier).metadata.location == existing
 
 
-def test_replace_table_rejects_empty_location(catalog: Catalog, test_table_identifier: Identifier) -> None:
+def test_replace_table_uses_explicit_location(catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path) -> None:
     _, schema = _create_simple_table(catalog, test_table_identifier)
-    with pytest.raises(ValueError, match="location must not be empty"):
-        catalog.replace_table_transaction(test_table_identifier, schema=schema, location="/")
+    new_location = f"file://{tmp_path}/relocated"
+    catalog.replace_table_transaction(test_table_identifier, schema=schema, location=new_location).commit_transaction()
+    assert catalog.load_table(test_table_identifier).metadata.location == new_location
 
 
-@pytest.mark.parametrize("location_kind", ["inherit", "explicit", "trailing-slash"])
-def test_replace_table_location(catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path, location_kind: str) -> None:
+def test_replace_table_strips_trailing_slash_from_location(
+    catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path
+) -> None:
     _, schema = _create_simple_table(catalog, test_table_identifier)
-    existing = catalog.load_table(test_table_identifier).metadata.location
-
-    if location_kind == "inherit":
-        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
-        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
+    bare = f"file://{tmp_path}/relocated"
+    catalog.replace_table_transaction(test_table_identifier, schema=schema, location=bare + "/").commit_transaction()
+    assert catalog.load_table(test_table_identifier).metadata.location == bare
 
 
 def test_replace_table_transaction_rolls_back_on_failure(catalog: Catalog, test_table_identifier: Identifier) -> None:
@@ -712,8 +741,11 @@ def test_concurrent_replace_transaction_schema_conflict(catalog: Catalog, test_t
     txn_b = catalog.replace_table_transaction(test_table_identifier, schema=_REPLACE_SCHEMA)
 
     txn_a.commit_transaction()
+    after_a = catalog.load_table(test_table_identifier).metadata
     with pytest.raises(CommitFailedException, match="last assigned field id"):
         txn_b.commit_transaction()
+    # The failed commit must be a true no-op: no metadata advanced past where `txn_a` left things.
+    assert catalog.load_table(test_table_identifier).metadata.last_column_id == after_a.last_column_id
 
 
 def test_concurrent_replace_transaction_partition_spec_conflict(catalog: Catalog, test_table_identifier: Identifier) -> None:
@@ -723,8 +755,11 @@ def test_concurrent_replace_transaction_partition_spec_conflict(catalog: Catalog
     txn_b = catalog.replace_table_transaction(test_table_identifier, schema=schema, partition_spec=new_spec)
 
     txn_a.commit_transaction()
+    after_a = catalog.load_table(test_table_identifier).metadata
     with pytest.raises(CommitFailedException, match="last assigned partition id"):
         txn_b.commit_transaction()
+    # The failed commit must be a true no-op: no metadata advanced past where `txn_a` left things.
+    assert catalog.load_table(test_table_identifier).metadata.last_partition_id == after_a.last_partition_id
 
 
 # Rename table tests

tests/table/test_partitioning.py

@@ -434,15 +434,16 @@ def test_assign_fresh_partition_spec_ids_for_replace_v1_keeps_field_preserves_id
 
 def test_assign_fresh_partition_spec_ids_for_replace_v1_void_name_uses_multi_suffix_loop() -> None:
     """When `name` and `name_<field_id>` are both already used, append `_2`, `_3`, ... until unique."""
+    # Three columns, one role each: source for the current (about-to-be-voided) partition,
+    # source for the new partition that collides on the void's preferred name, and source for
+    # the new partition that collides on the void's first fallback name.
     schema = Schema(
-        NestedField(field_id=1, name="foo", field_type=IntegerType(), required=False),
-        NestedField(field_id=2, name="bar", field_type=IntegerType(), required=False),
-        NestedField(field_id=3, name="baz", field_type=IntegerType(), required=False),
-        NestedField(field_id=4, name="qux", field_type=IntegerType(), required=False),
+        NestedField(field_id=1, name="current_source", field_type=IntegerType(), required=False),
+        NestedField(field_id=2, name="collide_on_name", field_type=IntegerType(), required=False),
+        NestedField(field_id=3, name="collide_on_fallback", field_type=IntegerType(), required=False),
     )
-    # Current v1 spec partitions source=1 by bucket(4) at field_id=1000, named "p" — note
-    # the partition NAME ("p") differs from the source COLUMN NAME ("foo") which is allowed
-    # for non-identity transforms.
+    # Current v1 spec partitions source=1 by bucket(4) at field_id=1000, named "p" — for
+    # non-identity transforms the partition NAME doesn't have to match the source column name.
     current_spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=BucketTransform(4), name="p"), spec_id=0)
     # New spec has two partition fields named "p" and "p_1000" — colliding with both the
     # void's preferred name and its first fallback. Both are on different sources, so they