perf: use columnar state for ordered ARRAY_AGG and STRING_AGG

[lyne7-sc]

Which issue does this PR close?

• Related to Very high memory usage for unnest + group by #20788.

Rationale for this change

Ordered ARRAY_AGG and STRING_AGG currently have high memory and sorting overhead for large inputs.

This is relevant to #20788, where removing ORDER BY from several ARRAY_AGG expressions avoids the reported memory explosion. This PR improves the ordered ARRAY_AGG and STRING_AGG accumulators.

What changes are included in this PR?

This PR changes the ordered aggregate accumulator to:

• Retain payload values as Arrow arrays.
• Store compact references to payload rows.
• Encode ordering keys with RowConverter.
• Reuse already sorted input and partial-state runs.
• Sort remaining entries and merge the sorted runs.
• Materialize the result with Arrow interleave or a zero-copy slice.

The accumulator is shared by ordered ARRAY_AGG and STRING_AGG.

Are these changes tested?

Yes. Unit and SQL logic tests cover ordering, partial-state merging, NULL handling, and type preservation.

Are there any user-facing changes?

No. This is an internal performance optimization

Benchmarks

Benchmark coverage is added in #24391.

Benchmark	Main runtime	Optimized runtime	Runtime improvement	Main peak memory	Optimized peak memory	Memory reduction
Ordered ARRAY_AGG	897.8 ms	85.4 ms	10.5x	404.9 MB	165.1 MB	59.2%
Two ordered ARRAY_AGG expressions	2048.2 ms	150.8 ms	13.6x	795.3 MB	247.8 MB	68.8%
Ordered STRING_AGG	895.1 ms	96.3 ms	9.3x	414.6 MB	158.0 MB	61.9%

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.40486% with 116 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.19%. Comparing base (bbb552f) to head (bc03e0c).
⚠️ Report is 37 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/functions-aggregate/src/array_agg.rs	83.40%	30 Missing and 86 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24392      +/-   ##
==========================================
+ Coverage   81.14%   81.19%   +0.04%     
==========================================
  Files        1110     1110              
  Lines      386137   389248    +3111     
  Branches   386137   389248    +3111     
==========================================
+ Hits       313347   316047    +2700     
- Misses      54324    54544     +220     
- Partials    18466    18657     +191     

☔ 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.

[jayzhan211]

I think overall looks good to me

[jayzhan211]

I'm not sure about this casting — should it happen inside the accumulator, or should the types already be correct before they enter it, like at the logical layer, so we don't need to cast again here?

[lyne7-sc]

I think this could ideally be handled at a common physical-expression boundary rather than by each accumulator.

However, self.value_type comes from the declared aggregate argument type, while the concrete ArrayRef is produced later during physical expression evaluation. It seems that exact type equality is not currently validated at that boundary, as also noted in #24022. #24029 handles such differences by normalizing the final array in ScalarValue::new_list.

Since this columnar implementation bypasses ScalarValue::new_list and interleave expects consistent source types, normalizing compatible inputs here seems like a reasonable way to preserve the existing behavior.

The accumulator-side nested-nullability case is exercised by ordered_aggregate_nested_nullability_mismatch_issue_24022. The test expects the result to use the accumulator's declared nullable type even when the concrete input field is non-nullable.

[jayzhan211]

If the array isn't filtered much — barely anything changes, or only 1-2 items get removed — would copying the whole array every time still cause high overhead?

[lyne7-sc]

Yes, I had the same concern. When I tried removing this copy locally and ran does_not_over_account_memory_ordered, the accumulator's reported size increased from 2295 to 7243 bytes.

This makes me concerned that removing it could introduce a memory regression in cases where retained ArrayRefs keep much larger backing buffers alive, especially given the related memory explosion reported in #20788 with the previous ordered ARRAY_AGG implementation.

On the performance side, I also compared the two variants using the benchmark, and the difference was relatively small for this workload.

It may still be worth exploring a conditional policy, although there does not seem to be a simple way to determine whether an arbitrary nested arrow array is already compact.