perf(scheduler): merge a task's sorted partitions before the shuffle write

[Dandandan]

What

A sorted passthrough shuffle stage is read back by an ordering-preserving merge.
This does that merge inside the producing task instead, with the per-partition
TopK's row limit: each task writes one file of at most n rows instead of one
file per partition holding n each. The consumer then merges T × n rows from
T sources rather than P × n from P.

The rewrite runs on the per-task plan — in the stage plan the writer would take
its partitioning from the merge and the stage would collapse to one task. Tasks
report output partition 0, which file_id already disambiguates.

Taking the limit is sound because a per-partition TopK(n) only exists when a
global limit of n sits above the consuming merge, so any row in the global top
n that a task holds is in that task's own top n.

Only sound when the consumer merges a partition's locations in order, so the AQE
adapter makes the call and marks the writer — build_reader plants
RangeShuffleReaderExec for that same exchange under the same conditions. The
static planner never marks; it plants the arrival-order reader, and concatenating
separately sorted files is not sorted.
ballista.shuffle.merge_ordered_passthrough (default true) turns it off.

Measured

TPC-H SF10, --partitions 64, 2 executors × 4 vcores. All 22 queries verified
against single-process DataFusion.

Rows the consumer's ordered reader merges — each result is exactly T × n
(16 tasks × the query's limit):

query	off	on
q2	4,670	1,600	−66%
q3	497	160	−68%
q10	1,100	320	−71%
q21	4,010	1,600	−60%
q18	624	624	limit not binding

Passthrough shuffle files across those queries: 709 → 469, i.e. P → T per stage.

Wall clock does not change for TPC-H (at SF=10 at least)

[milenkovicm]

Thanks @Dandandan one question regarding planning logic

[milenkovicm]

I wonder could this be moved to AQE planner rule instead of having planning logic at this point ?

[avantgardnerio]

When I ran this with h2o q8 with parallel windows enabled, it hung. PTAL

Kicking off CI again, since it should trigger there too.