Test coverage: persisted props, list/map fresh ids, sort-order reuse

- Pin that 'format-version' is not in replaced.metadata.properties.
- Add list/map element_id/key_id/value_id reuse coverage for
  assign_fresh_schema_ids_for_replace.
- Extend sort-order reuse test to cover reusing a non-zero sorted
  order_id from history.
- Drop the stale HEAD (view-exists) reference in
  test_replace_table_issues_commit_post_immediately.

Author: smaheshwar-pltr

tests/catalog/test_catalog_behaviors.py

@@ -520,21 +520,26 @@ def test_replace_table_reuses_partition_spec_id(catalog: Catalog, test_table_ide
 
 def test_replace_table_with_sort_order_changes(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """Replace can change the sort order. The new sort order is appended to the history and
-    becomes the default; a follow-up replace back to unsorted reuses the unsorted order_id
-    rather than appending a duplicate."""
+    becomes the default; a follow-up replace back to a sort already in history reuses its
+    order_id rather than appending a duplicate."""
     _, schema = _create_simple_table(catalog, test_table_identifier)
     sort = SortOrder(SortField(source_id=1, transform=IdentityTransform(), direction=SortDirection.ASC))
 
     # unsorted → sorted: a new order is added and becomes the default.
     sorted_table = catalog.replace_table(test_table_identifier, schema=schema, sort_order=sort)
     assert sorted_table.sort_order().fields == sort.fields
-    assert sorted_table.metadata.default_sort_order_id != 0
+    sorted_order_id = sorted_table.metadata.default_sort_order_id
+    assert sorted_order_id != 0
 
     # sorted → unsorted: reuses the unsorted order_id 0 from history.
     unsorted_table = catalog.replace_table(test_table_identifier, schema=schema)
     assert unsorted_table.sort_order().is_unsorted
     assert unsorted_table.metadata.default_sort_order_id == 0
 
+    # unsorted → original sorted form: reuses the existing sorted order_id from history.
+    replayed = catalog.replace_table(test_table_identifier, schema=schema, sort_order=sort)
+    assert replayed.metadata.default_sort_order_id == sorted_order_id
+
 
 def test_replace_table_inherits_existing_location(catalog: Catalog, test_table_identifier: Identifier) -> None:
     """`location=None` keeps the existing table's location."""
@@ -574,6 +579,8 @@ def test_replace_table_upgrades_format_version(catalog: Catalog, test_table_iden
     assert catalog.load_table(test_table_identifier).format_version == 1
     replaced = catalog.replace_table(test_table_identifier, schema=schema, properties={"format-version": "2"})
     assert replaced.format_version == 2
+    # `format-version` is a control input, not a regular property — must not leak into persisted properties.
+    assert "format-version" not in replaced.properties
 
 
 def test_replace_table_rejects_format_version_downgrade(catalog: Catalog, test_table_identifier: Identifier) -> None:

tests/catalog/test_rest.py

@@ -3017,8 +3017,8 @@ def test_replace_table_issues_commit_post_immediately(
         NestedField(field_id=2, name="data", field_type=StringType(), required=False),
     )
 
-    # Baseline: opening the transaction alone makes the HEAD (view-exists) + GET (load) calls
-    # but no POST until the caller commits.
+    # Baseline: opening the transaction alone makes only the load GET call; no POST is
+    # issued until the caller commits.
     catalog.replace_table_transaction(identifier=("fokko", "fokko2"), schema=new_schema)
     methods_after_open = [r.method for r in rest_mock.request_history]
     assert "POST" not in methods_after_open

tests/test_schema.py

@@ -1888,3 +1888,42 @@ def test_assign_fresh_schema_ids_for_replace_with_nested_struct() -> None:
     assert loc_fields[1].field_id == 4  # location.lon reused
     assert loc_fields[2].field_id == 5  # location.alt is new
     assert last_col_id == 5
+
+
+def test_assign_fresh_schema_ids_for_replace_with_list_and_map() -> None:
+    """`element_id`, `key_id`, and `value_id` are reused by name path (e.g. `tags.element`, `m.key`, `m.value`)."""
+    base_schema = Schema(
+        NestedField(
+            field_id=1,
+            name="tags",
+            field_type=ListType(element_id=2, element_type=StringType(), element_required=False),
+            required=False,
+        ),
+        NestedField(
+            field_id=3,
+            name="m",
+            field_type=MapType(key_id=4, key_type=StringType(), value_id=5, value_type=IntegerType(), value_required=False),
+            required=False,
+        ),
+    )
+    new_schema = Schema(
+        NestedField(
+            field_id=10,
+            name="tags",
+            field_type=ListType(element_id=20, element_type=StringType(), element_required=False),
+            required=False,
+        ),
+        NestedField(
+            field_id=30,
+            name="m",
+            field_type=MapType(key_id=40, key_type=StringType(), value_id=50, value_type=IntegerType(), value_required=False),
+            required=False,
+        ),
+    )
+    fresh, last_col_id = assign_fresh_schema_ids_for_replace(new_schema, base_schema, 5)
+    assert fresh.fields[0].field_id == 1
+    assert fresh.fields[0].field_type.element_id == 2
+    assert fresh.fields[1].field_id == 3
+    assert fresh.fields[1].field_type.key_id == 4
+    assert fresh.fields[1].field_type.value_id == 5
+    assert last_col_id == 5