[tomron]

Closes #4202

Rationale for this change

UpdateSortOrder only asserts the default sort order id, so a stale handle can commit an AddSortOrderUpdate whose id already exists. When the orders are equal, the metadata ends up with two sort orders sharing an id, which Java's TableMetadataParser rejects. When they differ, it raises a non-retryable ValueError.

This mirrors Java's TableMetadata.Builder.addSortOrder:

• Reuse the id of an existing sort order with the same fields instead of appending a duplicate.
• Assign the next free id when the requested id is taken by a different sort order.
• Track the last added sort order in the update context so SetDefaultSortOrderUpdate(-1) resolves to it, rather than the highest id.

Are these changes tested?

Yes. New catalog behavior tests cover the stale-handle scenarios from the issue and multiple sort order changes in one transaction. There are also unit tests for update_table_metadata.

Are there any user-facing changes?

No API changes. Stale sort order updates no longer produce invalid metadata.

[rambleraptor]

Try running this test. I had AI whip up something to show an issue around reassigned IDs + wrong orders becoming the default:

def test_stale_transaction_keeps_reassigned_sort_order_as_default(
    catalog: Catalog, table_schema_nested: Schema, test_table_identifier: Identifier
) -> None:
    namespace = Catalog.namespace_from(test_table_identifier)
    catalog.create_namespace(namespace)
    table_a = catalog.create_table(test_table_identifier, table_schema_nested)
    table_b = catalog.load_table(test_table_identifier)

    with table_a.update_sort_order() as update:
        update.asc("foo", IdentityTransform())
    with table_a.update_sort_order():
        pass  # Reset to unsorted

    # Stale handle stages bar (id 1), baz (id 2), then bar again, which it sends as an explicit SetDefault(1)
    with table_b.transaction() as transaction:
        with transaction.update_sort_order() as update:
            update.asc("bar", IdentityTransform())
        with transaction.update_sort_order() as update:
            update.asc("baz", IdentityTransform())
        with transaction.update_sort_order() as update:
            update.asc("bar", IdentityTransform())

    # bar and baz are reassigned to 2 and 3, so the default must follow bar rather than foo's id 1
    metadata = catalog.load_table(test_table_identifier).metadata
    assert [sort_order.order_id for sort_order in metadata.sort_orders] == [0, 1, 2, 3]
    assert metadata.default_sort_order_id == 2
    assert metadata.sort_order().fields[0].source_id == 2

[rambleraptor]

Whoops, hit the wrong button accidentally.

[dragonknightk]

Thanks for picking this up ʕ•ᴥ•ʔ The reuse path, the taken-id fallback, and following the reassigned ids all line up with what Java does, and both the catalog behavior tests and the new transaction test cover the repro from the issue nicely.

Left a couple of nits inline, both about what the new tests would still let through — no behavior changes.

[dragonknightk]

nit: the reused ids here are all the current default, so the suite stays green if the reuse path starts writing a metadata version. Asserting the timestamp keeps it honest.

Suggested change

	assert new_metadata.default_sort_order_id == existing_sort_order.order_id
	assert new_metadata.default_sort_order_id == existing_sort_order.order_id
	assert new_metadata.last_updated_ms == table_v2.metadata.last_updated_ms

[dragonknightk]

nit: the reused ids here are always the highest one, so recording max(...) instead of the matched sort.order_id would still pass. The reuse path also has to resolve to the matched order itself when its id is no longer the highest — a writer that read metadata before a newer sort order appeared asks for the older one with the id that newer order took.

Suggested change

	assert new_metadata.default_sort_order_id == existing_sort_order.order_id + 1
	assert new_metadata.default_sort_order_id == existing_sort_order.order_id + 1
	def test_update_metadata_add_existing_sort_order_reuses_its_own_id(table_v2: Table) -> None:
	existing_sort_order = table_v2.sort_order()
	newer_sort_order = SortOrder(SortField(source_id=1, transform=IdentityTransform()), order_id=existing_sort_order.order_id + 1)
	metadata = update_table_metadata(
	table_v2.metadata,
	(AddSortOrderUpdate(sort_order=newer_sort_order), SetDefaultSortOrderUpdate(sort_order_id=-1)),
	)
	# A writer that read metadata before newer_sort_order was added picks its own next id, which newer_sort_order took.
	# Reusing existing_sort_order has to resolve to its own id, which is no longer the highest one.
	new_metadata = update_table_metadata(
	metadata,
	(
	AddSortOrderUpdate(sort_order=SortOrder(*existing_sort_order.fields, order_id=newer_sort_order.order_id + 1)),
	SetDefaultSortOrderUpdate(sort_order_id=-1),
	),
	)
	assert new_metadata.sort_orders == metadata.sort_orders
	assert new_metadata.default_sort_order_id == existing_sort_order.order_id

[tomron]

@dragonknightk, thanks for reviewing. Just pushed few fixes, please check

[dragonknightk]

@tomron
Thanks for addressing both nits! The new test confirms that the reused sort order becomes the default even when it no longer has the highest ID. That’s exactly what I was worried about. LGTM (・ω・)ゞ