fix(spec): reuse the highest matching partition field id, not an arbitrary one

[xanderbailey]

reuse_partition_field_ids collects (source_id, transform) -> field_id into a HashMap with no merge function, so when several specs give one pair different field ids the last one visited wins. partition_specs is itself a HashMap, so the winner changes from run to run and the same input metadata yields different output.

Keep the highest id of each pair, as Iceberg Java does:

Collectors.toMap(
    field -> Pair.of(field.sourceId(), field.transform().toString()),
    PartitionField::fieldId,
    Math::max)

Divergent ids for one pair arise from tables written before the reuse rule existed, by writers that do not apply it, or from a caller passing an explicit field_id that conflicts with an equivalent existing field — reuse here only fills in an unset field_id, and nothing rejects a conflicting explicit one.

What changes are included in this PR?

The reuse map is now built by highest_equivalent_field_ids, which keeps the highest field id per (source_ids, transform) pair instead of letting the last spec visited win.

Are these changes tested?

Two unit tests in table_metadata_builder.rs: test_highest_equivalent_field_ids_is_independent_of_field_order checks the helper gives the same map for a field order and its reverse, and test_partition_field_id_reuse_picks_the_highest_of_several_candidates builds a spec on metadata where four specs give identity(x) four different ids and asserts the highest is reused.

AI Disclosure

Core code change made by me, tests written with Claude.

[mbutrovich]

Thanks @xanderbailey for making the partition field id reuse deterministic.

Could you fill in the template sections at the bottom of the description, in particular the testing and AI disclosure sections that the contribution guidelines ask for?

[mbutrovich]

Could we keep a comment here that says why the highest id wins? Without it, the entry loop reads like a longer way to write the old collect(), and someone could simplify it back. #3057 has a version of this comment that says why.

Suggested change

	let mut equivalent_field_ids = HashMap::new();
	// Build a map of (source_id, transform) -> field_id from existing specs. Keep the highest
	// id of each pair so the result doesn't depend on the iteration order of `partition_specs`.
	let mut equivalent_field_ids = HashMap::new();

[mbutrovich]

There are four existing specs here, so "either" doesn't fit.

Suggested change

	// Two fields, so this is not compatible with either existing spec and is added as a new
	// one rather than resolving to an existing spec id.
	// Two fields, so this is not compatible with any existing spec and is added as a new
	// one rather than resolving to an existing spec id.

[mbutrovich]

Does this test catch a regression on every run? With the old collect(), the result depends on which spec the HashMap visits last, so the test passes whenever spec 3 comes last. When I put the old code back and ran this test 40 times, it passed 11 times.

What do you think about deserializing the metadata inside a loop? Each serde_json::from_value builds partition_specs with a new RandomState, so each iteration visits the specs in a different order. Something like this, with metadata_json holding the json! value from above:

for _ in 0..32 {
    let metadata: TableMetadata = serde_json::from_value(metadata_json.clone()).unwrap();
    let result = metadata
        .into_builder(Some(
            "s3://bucket/test/location/metadata/v1.json".to_string(),
        ))
        .add_partition_spec(new_spec.clone())
        .unwrap()
        .build()
        .unwrap();
    // same assertions as now
}

With that change, the test failed in 40 of 40 runs against the old code.

[xanderbailey]

This is a nice idea, I've riffed on this, rather than the loop I've extracted out highest_equivalent_field_ids with a test to show that the result is the same regardless of the input order. I this this encodes the behaviour a little better than just relying on a loop I thnink, WDYT?

[mbutrovich]

I've extracted out highest_equivalent_field_ids with a test to show that the result is the same regardless of the input order. I this this encodes the behaviour a little better than just relying on a loop I thnink, WDYT?

I like this better too. When I put the old last-wins collect() back into highest_equivalent_field_ids, test_highest_equivalent_field_ids_is_independent_of_field_order fails on every run, and the reversed iterator covers first-wins as well.

test_partition_field_id_reuse_picks_the_highest_of_several_candidates still depends on the HashMap order. Against the last-wins collect() it passed 13 of 40 runs, so it would only sometimes catch reuse_partition_field_ids going back to an inline collect() that skips the helper. What do you think about wrapping its deserialize-and-build in the loop from my first comment, so the end-to-end test is deterministic as well?

[xanderbailey]

Is the new test not enough to document and prevent regressions? It would seem very deliberate to remove that test? I personally don't love the idea of tests looping to try and force non-determinism?

[mbutrovich]

Is the new test not enough to document and prevent regressions? It would seem very deliberate to remove that test? I personally don't love the idea of tests looping to try and force non-determinism?

Fair point, let's drop the loop. It only makes a false pass less likely, so calling it deterministic was wrong on my part. test_highest_equivalent_field_ids_is_independent_of_field_order is the deterministic check, and keeping the end-to-end test as is works for me.

[xanderbailey]

@laskoviymishka if you have a moment to review that would be awesome! As always, thanks in advance!

[mbutrovich]

Thanks @xanderbailey for the updates. The code change looks right to me, and my remaining questions are about the PR description.

The description says divergent ids for one pair "cannot be produced by this crate". As I read reuse_partition_field_ids, reuse only applies when the caller leaves field_id unset (L857). On a v2 table whose spec 0 has identity(id) with field id 1000, add_partition_spec accepts a new spec with identity(id) given an explicit field_id of 1005, and the table ends up with both ids for the same pair. Could you reword that sentence? The spec says partition field IDs must be reused for an equivalent field, so should we also file an issue to decide whether add_partition_spec should reject an explicit id that conflicts with an existing equivalent field?

Could you also fill in the testing section (where the tests are is enough) and either link an issue on the "Closes" line or remove it?

[xanderbailey]

@mbutrovich @laskoviymishka I think this should be good for another look now, thank you both!