feat(core): PBWAG — parallel BWAG wrapper

[avantgardnerio]

Summary

purely additive

PartitionedBoundedWindowAggExec wraps DataFusion's BoundedWindowAggExec and overrides its required_input_distribution from SinglePartition to UnspecifiedDistribution. It hides BWAG from tree walkers by returning only the wrapper's input from children(), so EnforceDistribution can't reinsert an SPM(K→1) beneath it.

This is extracted from #2223 — the full parallel-BWAG stack for h2o's range-window Q8 shape. #2223 has already demonstrated the shape works end-to-end and delivers a 4.7× speedup at 100M rows on a 2-pod EKS cluster. Landing the pieces incrementally keeps each PR's diff focused and reviewable.

What this operator is for

DataFusion's BoundedWindowAggExec declares Distribution::SinglePartition when no PARTITION BY is present — a correctness guard because a window frame's semantics can span rows across the whole input. If the input is already range-repartitioned (each partition is a globally-disjoint slice of the ORDER BY key + halo covering frame boundaries), BWAG can safely run per-partition and produce K correct outputs. EnforceDistribution doesn't know that, so it plants an SPM(K→1) above BWAG and collapses everything to one vcore. This wrapper is the mechanism that tells EnforceDistribution to stand down.

The wrapper doesn't (and can't) verify that its input actually is range-repartitioned. Optimizer rules which plant this operator are responsible for the invariant. Today that is zero. When we merge #2223 there will be one.

Status: temporary

This is a Ballista-side placeholder for apache/datafusion#23026 or similar. Once upstream has this functionality, this wrapper should collapse and callers target DF's BWAG directly.

What's in this PR

• ballista_core::execution_plans::partitioned_bounded_window_agg: the new operator. InputOrderMode and can_repartition are hardcoded (Sorted / false) — the only planned caller is a no-PARTITION-BY + single-Column-ORDER-BY range-window rule, so hardcoding keeps the wire format and the type small.
• BallistaPhysicalPlanNode::PartitionedBoundedWindowAgg: proto message carrying only window_expr — the rest is implicit from the caller's shape gates. Round-trip goes through DF's serialize_physical_window_expr / parse_physical_window_expr.
• Unit test per_partition_execute_running_sum_no_cross_partition_leak builds a two-partition memory source (rows 1–3 in partition 0, rows 100–102 in partition 1) with a RANGE 2.0 PRECEDING frame that spans within a partition but not across, and asserts:
  • BWAG actually aggregates (partition 0's running sums are [1, 3, 6], not [1, 2, 3]).
  • K→K partitioning doesn't leak (partition 1's first sum is 100, not 106).
  • required_input_distribution is UnspecifiedDistribution.
  • children() returns 1 element (BWAG hidden from tree walkers).

What's NOT in this PR

No in-tree caller. This is prep for the ParallelWindowRule follow-up (also extracted from #2223), which is what actually plants this operator in the plan. Landing them separately keeps each diff focused.

Test plan

• cargo test -p ballista-core --lib per_partition_execute passes locally.
• cargo fmt --all -- --check clean.
• cargo clippy --all-targets --workspace clean.
• cargo check --workspace --all-targets clean.

[avantgardnerio]

@phillipleblanc FYI

[milenkovicm]

LGTM thanks @avantgardnerio, very cool initiative