avantgardnerio opened a new pull request, #2253: URL: https://github.com/apache/datafusion-ballista/pull/2253
## Summary `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 the first slice 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. Callers are responsible for the invariant. Wiring this wrapper below arbitrary inputs will silently produce wrong window values — the module doc calls this out. ## Status: temporary This is a Ballista-side placeholder for [apache/datafusion#23026](https://github.com/apache/datafusion/pull/23026) ("Parallel bounded RANGE-frame window functions without PARTITION BY"). Once that lands upstream and Ballista bumps its DF pin past it, 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 - [x] `cargo test -p ballista-core --lib per_partition_execute` passes locally. - [x] `cargo fmt --all -- --check` clean. - [x] `cargo clippy --all-targets --workspace` clean. - [x] `cargo check --workspace --all-targets` clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
