Add niche but no-regret tests from audit follow-up

- test_replace_table_drops_identifier_field: pairs with the existing
  preserve test; verifies that a new schema without identifier_field_ids
  clears the previous set rather than silently carrying it forward.
- test_replace_table_v2_does_not_carry_forward_void_field: v2 specs
  aren't append-only, so a dropped partition field is gone (unlike v1).
- test_replace_after_format_version_upgrade: v1 -> v2 via replace,
  then a second replace on the now-v2 table must not retrigger the
  upgrade or fail.

Author: smaheshwar-pltr

tests/catalog/test_catalog_behaviors.py

@@ -492,6 +492,23 @@ def test_replace_table_preserves_identifier_field_ids(catalog: Catalog, test_tab
     assert list(replaced.schema().identifier_field_ids) == [1]
 
 
+def test_replace_table_drops_identifier_field(catalog: Catalog, test_table_identifier: Identifier) -> None:
+    """Replacing with a schema that has no `identifier_field_ids` clears them on the table —
+    the previous identifier set is not silently carried forward."""
+    schema_with_id = Schema(
+        NestedField(field_id=1, name="id", field_type=LongType(), required=True),
+        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
+        identifier_field_ids=[1],
+    )
+    _create_simple_table(catalog, test_table_identifier, schema=schema_with_id)
+    schema_without_id = Schema(
+        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
+        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
+    )
+    replaced = catalog.replace_table(test_table_identifier, schema=schema_without_id)
+    assert list(replaced.schema().identifier_field_ids) == []
+
+
 def test_replace_table_reuses_partition_spec_id(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """An identical partition spec reuses its spec_id rather than appending a new one."""
     spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, name="id_part", transform=IdentityTransform()))
@@ -582,6 +599,36 @@ def test_replace_table_v1_carries_forward_partition_fields_as_void(catalog: Cata
     assert void_field.name == "id_part"
 
 
+def test_replace_table_v2_does_not_carry_forward_void_field(catalog: Catalog, test_table_identifier: Identifier) -> None:
+    """v2 specs are not append-only — a replace that drops a partition field does not
+    carry it forward (unlike v1). The new default spec contains only the new field(s)."""
+    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=2)
+
+    # Replace with default unpartitioned spec — the old field is gone from the default spec.
+    replaced = catalog.replace_table(test_table_identifier, schema=schema)
+    new_spec = replaced.spec()
+    assert new_spec.is_unpartitioned()
+    assert all(not isinstance(f.transform, VoidTransform) for f in new_spec.fields)
+
+
+def test_replace_after_format_version_upgrade(catalog: Catalog, test_table_identifier: Identifier) -> None:
+    """A v1 table can be upgraded to v2 via replace and then re-replaced without issue."""
+    _, schema = _create_simple_table(catalog, test_table_identifier, format_version=1)
+    upgraded = catalog.replace_table(test_table_identifier, schema=schema, properties={"format-version": "2"})
+    assert upgraded.format_version == 2
+
+    # Second replace on the now-v2 table should not re-trigger an upgrade or fail.
+    new_schema = Schema(
+        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
+        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
+        NestedField(field_id=3, name="extra", field_type=BooleanType(), required=False),
+    )
+    replaced = catalog.replace_table(test_table_identifier, schema=new_schema)
+    assert replaced.format_version == 2
+    assert {f.name for f in replaced.schema().fields} == {"id", "data", "extra"}
+
+
 def test_replace_table_raises_when_table_does_not_exist(catalog: Catalog, test_table_identifier: Identifier) -> None:
     schema = Schema(NestedField(field_id=1, name="id", field_type=LongType(), required=False))
     with pytest.raises(NoSuchTableError):