perf: Optimize prefix-group processing in PartialSortExec

[lyne7-sc]

Which issue does this PR close?

• Closes #.

Rationale for this change

PartialSortStream concatenates each incoming batch with the buffered incomplete prefix before looking for a completed prefix boundary.

When one prefix spans multiple batches, this repeatedly copies rows accumulated from earlier batches. When a batch contains multiple completed prefix groups, the completed region is also sorted using the full ordering even though the prefix ordering is already satisfied.

This adds unnecessary copying and sorting work, especially for wide batches and workloads containing multiple prefix groups per batch.

What changes are included in this PR?

• Keep the unfinished trailing prefix as a list of RecordBatch slices instead of concatenating it with every incoming batch.
• Detect completed prefix groups within each batch and compare prefix expression results directly across batch boundaries.
• Sort completed groups only by the remaining suffix ordering.
• Concatenate a single completed prefix at most once.
• For multiple completed prefixes, evaluate suffix expressions once per source batch and materialize the final output with one interleave operation.
• Preserve fetch handling, input release, empty-batch handling, and zero-column batch semantics.

What is the testing strategy for this PR?

The implementation is covered by the existing PartialSortExec unit tests and the additional SQL logic test cases in group_by.slt.

Are there any user-facing changes?

No. This is an internal performance improvement.

Benchmark

group                                 main                                   optimized
-----                                 ----                                   ---------
partial_sort/rows_per_prefix/100      3.10     12.1±0.13ms        ? ?/sec    1.00      3.9±0.05ms        ? ?/sec
partial_sort/rows_per_prefix/1000     4.17     13.2±0.07ms        ? ?/sec    1.00      3.2±0.14ms        ? ?/sec
partial_sort/rows_per_prefix/10000    1.30      2.8±0.02ms        ? ?/sec    1.00      2.1±0.07ms        ? ?/sec
partial_sort/rows_per_prefix/20000    1.41      3.0±0.22ms        ? ?/sec    1.00      2.1±0.15ms        ? ?/sec
partial_sort/rows_per_prefix/5000     3.85     12.4±0.27ms        ? ?/sec    1.00      3.2±0.03ms        ? ?/sec
partial_sort/rows_per_prefix/8192     1.56      2.8±0.05ms        ? ?/sec    1.00  1780.8±34.70µs        ? ?/sec

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.41176% with 36 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.91%. Comparing base (85d4cbb) to head (f7eb934).

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/sorts/partial_sort.rs	89.41%	19 Missing and 17 partials ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #24979    +/-   ##
========================================
  Coverage   81.91%   81.91%            
========================================
  Files        1134     1134            
  Lines      425631   425938   +307     
  Branches   425631   425938   +307     
========================================
+ Hits       348647   348923   +276     
- Misses      56304    56322    +18     
- Partials    20680    20693    +13     

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

[kosiew]

@lyne7-sc,

Thanks for working on this. The new prefix-group processing looks like a solid improvement, especially avoiding repeated concatenation for prefixes that span batches and limiting sorting work to the suffix keys.

I have one non-blocking suggestion around coverage for special prefix values. Otherwise, this looks good to me.

[kosiew]

Could we add a cross-batch regression test where the prefix key contains repeated NULL and NaN values, including nulls_first and descending sort options? It would be useful to compare the result against SortExec as well. prefix_changed_at_batch_boundary now determines boundary equality independently using make_comparator, while the in-batch groups come from evaluate_partition_ranges, so a test like this would help make sure both paths stay consistent for these special values.

[lyne7-sc]

Thanks for the suggestion! Added a cross-batch test with repeated NULL and NaN prefixes, covering ascending/descending order and NULLS FIRST/LAST, and comparing the results against SortExec.