avantgardnerio opened a new pull request, #2195: URL: https://github.com/apache/datafusion-ballista/pull/2195
## Summary Adds a standalone `PerPartitionFilterExec` operator plus its physical-plan codec. Carries `Vec<Arc<dyn PhysicalExpr>>` — one boolean predicate per input partition — and applies the k-th predicate to the k-th input stream on `execute(k)`. Not wired into any planner or rule; this is the operator surface, unit-tested in isolation. Follow-ups land the injection at range-repartition–consuming stages. ## Motivation Comes out of the adaptive-range-shuffle work (URRE / ORRE — merged in #2123, #2169). Once the scheduler picks stage-global cut points and each producer's local sub-parts are routed by sketch-overlap, the downstream consumers over-fetch: a consumer whose assigned range is `[cuts[k-1], cuts[k])` may pull an over-inclusive sub-part that straddles the cut. Those out-of-range rows need to be dropped before the consumer sees them, and the drop is per-partition (each downstream partition has a distinct predicate). Doing this as a plain filter on the whole consumer stream would require one task per predicate — K tasks for K partitions — which breaks vcore packing and defeats the whole point of the range-repartition family (a single task can own many partitions). `PerPartitionFilterExec` preserves the packing: N tasks each own a slice of partitions, and each partition applies its own predicate. ## How it plays with oversampling Oversampling (many small sub-parts relative to downstream partition count) is what makes this cheap. Under reasonable oversampling, each downstream partition's range overlaps at most its \"own\" sub-part plus its two immediate neighbors — call it ~3 files fetched per consumer. The straddling case is rare and shallow, so the filter's rejection rate stays low. ## This is the simplest thing that works The point of this PR is to unblock the end-to-end range-partitioning demo with the smallest operator surface possible. A later PR should push the filtering down into the shuffle/Flight path so the consumer never materializes rows it's about to reject: - **Where:** on the reader side of the Flight shuffle, before the batch is handed up the plan. - **How, cheap version:** the same per-partition predicate, applied to the arriving batch. Same semantics as this PR, one hop earlier. - **How, cheaper version for sorted producers:** binary-search the sorted batch on the routing column and slice the in-range prefix/suffix directly, avoiding the per-row predicate eval. Applies naturally to ORRE (ordered producers) and to any URRE stage whose upstream happens to be sorted on the routing key. That optimization is out of scope here. Landing the operator first keeps the range-partitioning wiring straightforward to review and lets us measure filter overhead on real workloads before deciding how far to push the follow-up. ## Test plan - [x] `cargo test -p ballista-core --lib per_partition_filter` — 5 operator tests + 1 codec round-trip pass locally. - [x] `cargo clippy --all-targets -p ballista-core` clean. - [x] `cargo fmt --all -- --check` clean. - [ ] CI green. 🤖 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]
