feat(scheduler): parallel bounded ROWS-frame windows via a rank-derived halo

[avantgardnerio]

What this is for

BoundedWindowAggExec requires a single input partition, so a window with no
PARTITION BY runs serially no matter how wide the cluster is. ParallelWindowRule
already fixes that for bounded RANGE frames and PrefixWindowRule for UNBOUNDED
PRECEDING. This adds the third sibling, HaloRowRule, for bounded ROWS frames:

avg(v2) OVER (ORDER BY id3 ROWS BETWEEN 100 PRECEDING AND CURRENT ROW)

On h2o window q6 at 1e7 that window is 10.03s of a 19.33s wall clock, in a stage
of exactly one task: stage 0's writer declares UnknownPartitioning, so the SPM
above the reader collapses 8 partitions to 1 and BWAG does the whole window alone.

Why the RANGE halo does not extend to ROWS

ParallelWindowRule's halo is a value delta end to end: files route by sketched
[min, max] overlap against cuts[k] +/- halo, and RangeFilterExec keeps
cuts[k-1] - halo_lo <= v < cuts[k] + halo_hi. A ROWS halo is a rank delta, and
nothing on that path expresses rank. Widening a cut by any value can land in a gap
holding no rows at all, so "100 preceding rows" is not reachable by a choice of
value width.

Where the rank bound comes from

The range shuffle sidecar index from #2364 carries one row per IPC message, with
that message's first key and its num_rows. Merging those index rows descending by
first key, across every candidate file, and accumulating num_rows until the total
reaches n gives a key: reading from it up to the cut yields a superset of the true
n predecessors. Comparison only, so strings, timestamps, multi-column and DESC
keys all work. A message straddling the cut counts as zero, since over-counting
would stop the walk short of the rows asked for.

The walk runs at two granularities:

• coarse, on the scheduler, at file granularity, over key_min / key_max /
  num_rows the runtime stats reports already carry. This fetches nothing, and it
  is what keeps the file list per consumer from growing with K.
• fine, on the consumer, at message granularity, over the index files
  RangeShuffleReaderExec already downloads. This is byte reduction, not
  correctness.

Cuts stay approximate KLL value cuts. Halo exactness rests on the index walk's exact
num_rows, not on cut exactness.

What does not change

OrderedRangeRepartitionExec, RangeFilterExec and PartitionedBoundedWindowAggExec
learn nothing about ROWS. The filter takes a resolved half-open band per partition,
which is what it already takes; the rank lives entirely in the scheduler-side walk
that produces those bands. Only the rule's frame-units gate mentions ROWS at all.

Results

h2o window q6 at 1e7 on a 2-executor cluster, target_partitions=8,
max_partitions_per_task=4, debug build. id3 is unique at that scale, so a ROWS
frame has one defined answer and DataFusion is a real row-for-row oracle; every
column below verified OK against it.

	serial	coarse walk	+ fine walk
wall clock	19.33s	15.13s	12.06s
rows read by the reader	10.00M	23.71M	10.33M
rows into the window	10.00M	23.63M	10.26M
window elapsed_compute	10.03s	25.04s	11.80s

The serial column is one task doing the whole window. The coarse column is eight,
each also reading a file's worth of halo it does not need — read amplification
2.37x, and the window pays for every extra row. The fine walk takes that to 1.03x.

Per consumer, what the second walk buys, from the reader's own log:

partition 1: cut=1389324, coarse=6,       fine=1367380    1.39M keys of halo ->  22k
partition 2: cut=2754474, coarse=1379372, fine=2724353    1.37M               ->  30k
partition 3: cut=4123520, coarse=1379372, fine=4100683    2.74M               ->  23k
partition 1: cut=6877053, coarse=5499087, fine=6842045    1.38M               ->  35k
partition 2: cut=8248583, coarse=5499087, fine=8225063    2.75M               ->  24k
partition 3: cut=9617707, coarse=8253613, fine=9581532    1.36M               ->  36k

22k-36k keys is about three 8192-row batches, against a theoretical best of the 100
rows the frame asks for.

Seven of the eight consumers tighten; the eighth is global partition 0, which has
nothing below it. That took shipping each consumer's own cut alongside the widened
one: the reader used to take it from the previous partition's upper bound, which is
the same value right up until max_partitions_per_task hands a task a slice of the
partitions and the previous one lands in a different task.

Tests

ballista/client/tests/halo_row.rs runs the standalone cluster against single-node
DataFusion, at K=4 over 48 unique keys: CURRENT ROW AND CURRENT ROW (no row from
outside a task's own range, so it isolates the plan shape), 5 PRECEDING, 20 PRECEDING (a halo reaching back past a whole partition), a FOLLOWING frame
asserting the gate leaves it on the serial path, and an EXPLAIN ANALYZE case
pinning that the rewrite fired at all.

Both walks are unit-tested on their own: rank_halo_walk_tests in runtime_stats.rs
over per-file stats, rank_walk_tests in range_shuffle/index.rs over real index
batches, both covering straddlers at the cut, key-order across files, running dry,
and a zero-row frame.

Scope

Frames reaching past the current row stay on the serial path: the walk widens a
consumer's lower edge only, and the mirror above the upper cut is not written yet.
DESC likewise, as with ParallelWindowRule.

Behaviour changes only for queries in the matched shape, and only when
ballista.planner.parallel_window.enabled is set, which defaults to false. The rule
shares that flag with ParallelWindowRule rather than adding a second knob.