Close test coverage gaps and split mega-test

Coverage gaps surfaced by branch coverage of pyiceberg/partitioning.py:

- v1 keep-field branch (preserves field_id when current and new specs
  share a (source, transform) key) had no helper-level test.
- v1 _unique_void_name multi-suffix loop (when name and name_<field_id>
  are both taken) had no test.
- v2 "prefer highest field_id" tiebreak (same key across multiple
  historical specs) had no test.

Also split test_complete_replace_transaction into three focused tests
behind a shared _run_complete_replace helper: applies-new-spec-and-sort,
merges-properties, and rtas-preserves-old-snapshot. The umbrella test
asserted a dozen unrelated invariants in one body; splitting gives
actionable failure messages.

Tightens test_replace_table_transaction_wire_payload to hardcode the
expected action list rather than testing-of-tests via len(set(...)).

Author: smaheshwar-pltr

tests/catalog/test_catalog_behaviors.py

@@ -41,7 +41,7 @@
 from pyiceberg.io.pyarrow import _dataframe_to_data_files, schema_to_pyarrow
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
-from pyiceberg.table import TableProperties
+from pyiceberg.table import Table, TableProperties
 from pyiceberg.table.snapshots import Operation
 from pyiceberg.table.sorting import NullOrder, SortDirection, SortField, SortOrder
 from pyiceberg.table.update import AddSchemaUpdate, SetCurrentSchemaUpdate
@@ -450,10 +450,17 @@ def test_replace_transaction(catalog: Catalog, test_table_identifier: Identifier
     assert replaced.scan(snapshot_id=old_snapshot_id).to_arrow().equals(_simple_data())
 
 
-def test_complete_replace_transaction(catalog: Catalog, test_table_identifier: Identifier, tmp_path: Path) -> None:
-    _create_simple_table(catalog, test_table_identifier, properties={"keep": "yes", "override": "old"})
-    catalog.load_table(test_table_identifier).append(_simple_data())
-    original = catalog.load_table(test_table_identifier)
+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.
+
+    Returns:
+        (original, replaced, new_sort, original_data, old_snapshot_id)
+    """
+    _create_simple_table(catalog, identifier, properties={"keep": "yes", "override": "old"})
+    catalog.load_table(identifier).append(_simple_data())
+    original = catalog.load_table(identifier)
     old_snapshot_id = original.current_snapshot().snapshot_id  # type: ignore[union-attr]
     original_data = original.scan().to_arrow()
 
@@ -471,7 +478,7 @@ def test_complete_replace_transaction(catalog: Catalog, test_table_identifier: I
     )
 
     with catalog.replace_table_transaction(
-        test_table_identifier,
+        identifier,
         schema=new_schema,
         partition_spec=new_spec,
         sort_order=new_sort,
@@ -480,12 +487,16 @@ def test_complete_replace_transaction(catalog: Catalog, test_table_identifier: I
     ) as txn:
         txn.append(new_data)
 
-    replaced = catalog.load_table(test_table_identifier)
+    return original, catalog.load_table(identifier), new_sort, original_data, 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)
     # Identity invariants.
     assert replaced.metadata.table_uuid == original.metadata.table_uuid
-    assert replaced.metadata.location == new_location
-
+    assert 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]
@@ -495,14 +506,23 @@ def test_complete_replace_transaction(catalog: Catalog, test_table_identifier: I
     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}
 
-    # Property merge: kept, overridden, added — and `format-version` does not leak.
+
+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)
+    # `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
 
-    # RTAS atomicity: new snapshot exists, has no parent (fresh start), old snapshot is still
-    # in the snapshot list, and time-travel reads return the original rows.
+
+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)
+    # New snapshot exists, has no parent (fresh start), old snapshot is still in the snapshot list.
     new_snapshot = replaced.current_snapshot()
     assert new_snapshot is not None
     assert new_snapshot.snapshot_id != old_snapshot_id

tests/catalog/test_rest.py

@@ -2953,7 +2953,13 @@ def test_replace_table_transaction_wire_payload(
     ]
 
     actions = [u["action"] for u in request["updates"]]
-    assert len(actions) == len(set(actions)), f"duplicate actions in request: {actions}"
+    assert sorted(actions) == [
+        "add-schema",
+        "remove-snapshot-ref",
+        "set-current-schema",
+        "set-default-sort-order",
+        "set-default-spec",
+    ]
     updates_by_action = {u["action"]: u for u in request["updates"]}
 
     assert updates_by_action["remove-snapshot-ref"] == {"action": "remove-snapshot-ref", "ref-name": "main"}

tests/table/test_partitioning.py

@@ -405,3 +405,76 @@ def test_assign_fresh_partition_spec_ids_for_replace_v1_renames_void_on_name_col
     void_field = next(f for f in fresh_spec.fields if isinstance(f.transform, VoidTransform))
     assert void_field.name != "data", "void name must not collide with active partition name"
     assert void_field.name == "data_1000"
+
+
+def test_assign_fresh_partition_spec_ids_for_replace_v1_keeps_field_preserves_id() -> None:
+    """v1 carry-forward: when a current-spec field is also in the new spec, its field_id is preserved."""
+    schema = Schema(
+        NestedField(field_id=1, name="id", field_type=IntegerType(), required=False),
+        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
+    )
+    current_spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="id"), spec_id=0)
+    # New spec keeps the same (source, transform) on "id" — should reuse field_id=1000, no void emitted.
+    new_spec = PartitionSpec(PartitionField(source_id=1, field_id=999, transform=IdentityTransform(), name="id"))
+    fresh_spec, new_last_pid = assign_fresh_partition_spec_ids_for_replace(
+        new_spec,
+        schema,
+        schema,
+        existing_specs=[current_spec],
+        last_partition_id=1000,
+        format_version=1,
+        current_spec=current_spec,
+    )
+    assert [f.field_id for f in fresh_spec.fields] == [1000]
+    assert fresh_spec.fields[0].name == "id"
+    assert isinstance(fresh_spec.fields[0].transform, IdentityTransform)
+    assert not any(isinstance(f.transform, VoidTransform) for f in fresh_spec.fields)
+    assert new_last_pid == 1000
+
+
+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."""
+    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),
+    )
+    # 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_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
+    # do not match the current (source=1, bucket[4]) key and the current field becomes void.
+    new_spec = PartitionSpec(
+        PartitionField(source_id=2, field_id=997, transform=BucketTransform(4), name="p"),
+        PartitionField(source_id=3, field_id=998, transform=BucketTransform(4), name="p_1000"),
+    )
+    fresh_spec, _ = assign_fresh_partition_spec_ids_for_replace(
+        new_spec,
+        schema,
+        schema,
+        existing_specs=[current_spec],
+        last_partition_id=1000,
+        format_version=1,
+        current_spec=current_spec,
+    )
+    void_field = next(f for f in fresh_spec.fields if isinstance(f.transform, VoidTransform))
+    assert void_field.name == "p_1000_2"
+
+
+def test_assign_fresh_partition_spec_ids_for_replace_v2_prefers_highest_field_id_for_repeated_key() -> None:
+    """v2: when the same (source_id, transform) appears across multiple specs, the highest field_id wins."""
+    # Two historical specs both partition by (source_id=1, IdentityTransform), with different field_ids.
+    existing_specs = [
+        PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="id"), spec_id=0),
+        PartitionSpec(PartitionField(source_id=1, field_id=1002, transform=IdentityTransform(), name="id_v2"), spec_id=1),
+    ]
+    # New spec uses the same (source, transform) — should reuse the highest historical field_id (1002).
+    new_spec = PartitionSpec(PartitionField(source_id=1, field_id=999, transform=IdentityTransform(), name="id"))
+    fresh_spec, new_last_pid = assign_fresh_partition_spec_ids_for_replace(
+        new_spec, _REPLACE_SCHEMA_FOR_PARTITION, _REPLACE_SCHEMA_FOR_PARTITION, existing_specs, last_partition_id=1002
+    )
+    assert fresh_spec.fields[0].field_id == 1002
+    assert new_last_pid == 1002