fix: duplicate groups after spilling in legacy hash aggregation with a fallback group key (type not having dedicated impl)

[rluvaton]

Found by the new aggregate fuzz tests:

• test: add aggregate fuzz tests for all shapes #24881

Claude:

Which issue does this PR close?

N/A

Rationale for this change

GROUP BY on a single nested column (Struct, Map) returns duplicate groups when the legacy GroupedHashAggregateStream spills: the same key comes out as several rows, with the aggregate values split between them. It needs the legacy stream (datafusion.execution.enable_migration_aggregate = false), a single nested group key, and enough memory pressure to spill in a Final or Single stage. Results are silently wrong rather than an error.

The migrated streams are not affected: after spilling they hand the merged input to OrderedFinalAggregateStream, which builds a fresh group values collector with GroupOrdering::Full.

After spilling, the legacy stream re-aggregates the merged spill files with GroupOrderingFull, which requires group ids in first-seen order along the sorted input. The stream recreates its group values collector for that phase to guarantee the order, but only when there is more than one group column, assuming a single column always uses a sequential single-column collector. A single nested column has no specialized single-column collector and is served by GroupValuesColumn through a row-backed column, whose vectorized interning assigns new ids out of input order under hash collisions. In a merged batch of 28 sorted rows the ids came out as 0 to 4, then 9 to 13, then 5 to 8. GroupOrderingFull then treated a group that was still arriving as complete and emitted it, and the next batch reopened it as a new group.

What changes are included in this PR?

GroupedHashAggregateStream now always recreates the group values collector when it switches to merging spill files, instead of only for multi-column keys.

What is the testing strategy for this PR?

New integration test memory_limit::nested_key_spill_keeps_groups_unique: a 200k-row table grouped by a struct of a list and an integer, with null and empty lists, null numbers and null structs mixed in, aggregated with six aggregates including count(distinct) under an 8 MB FairSpillPool with a 64-row batch size, compared against the same query with unlimited memory. It runs both the legacy stream and the migrated streams. Without the fix the legacy run fails deterministically with 72 rows instead of 71, keys split into two rows whose counts add up to the reference. The migrated run passes with and without the fix and is kept as a regression guard.

Are there any user-facing changes?

No API changes. Queries that hit this path now return correct results.

🤖 Generated with Claude Code

[codecov-commenter]

Codecov Report

❌ Patch coverage is 0% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 81.82%. Comparing base (4048898) to head (1c2097c).

Files with missing lines	Patch %	Lines
...hysical-plan/src/aggregates/grouped_hash_stream.rs	0.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24889      +/-   ##
==========================================
+ Coverage   81.80%   81.82%   +0.02%     
==========================================
  Files        1130     1130              
  Lines      417754   417752       -2     
  Branches   417754   417752       -2     
==========================================
+ Hits       341754   341845      +91     
+ Misses      55875    55771     -104     
- Partials    20125    20136      +11     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Weijun-H]

LGTM