adriangb opened a new pull request, #24601: URL: https://github.com/apache/datafusion/pull/24601
## Which issue does this PR close? - Related to #18856 - Informs https://github.com/datafusion-contrib/datafusion-distributed/pull/634 Does not close #18856: `PushedDown::No` still conflates "I will not use this filter" with "I will use it, but not for exact row-level filtering". This PR only stops that ambiguity from forcing a runtime decision. ## Rationale for this change `HashJoinExec` decides whether to compute a dynamic filter inside `execute()`, by walking the probe subtree looking for a node that holds the filter expression: ```rust // Only compute a dynamic filter when the probe subtree contains a consumer. let enable_dynamic_filter_pushdown = ... .map(|id| plan_contains_expression_id(&self.right, id)) ``` Whether a consumer exists is a planning-time property. Deciding it at execution time breaks any consumer that rewrites the plan after optimization. The concrete case is a distributed planner splitting the optimized plan into stages: ``` worker 1 HashJoinExec (dynamic filter) NetworkShuffleExec worker 2 DataSourceExec (consumes the dynamic filter) ``` At execution time on worker 1 the probe subtree ends at the network boundary, so the traversal finds nothing and the filter is silently never produced — even though the pushdown had found a consumer while the plan was still whole. Working around this requires the shuffle node to hold "anchor" references to filters it never evaluates, purely so the traversal sees them. The check itself is well motivated (#17527: skip build-side bounds accumulation when nothing will read the result). Its placement in `execute()` is a leftover from #19546, which implemented it as `Arc::strong_count`, a signal only meaningful once the whole plan is assembled. Since #24018 replaced refcounting with `expression_id` + `apply_expressions`, that constraint is gone — and `AggregateExec` already makes the same decision at planning time. ## What changes are included in this PR? - `HashJoinExec::handle_child_pushdown_result` runs the consumer check and only attaches the dynamic filter if the probe subtree contains a consumer, mirroring `AggregateExec::handle_child_pushdown_result`. - `HashJoinExec::execute` reduces to `self.dynamic_filter.is_some()`. - Documents the resulting contract on `HashJoinExec::with_dynamic_filter_expr`: holding a dynamic filter is what makes the join compute one, so a caller wiring one up by hand owns the consumer check. This is safe because the Post phase `FilterPushdown` rule is the last rule that mutates the plan (only `SanityCheckPlan` follows, which changes nothing), and the optimizer calls `handle_child_pushdown_result` on the node with its post-pushdown children already in place. The decision then travels as node state, surviving `replace_children` and the proto round trip. No new API, no new `PushedDown` state. As before, the discriminant is not consulted, because a node replying `PushedDown::No` may still retain the filter for statistics pruning. ## Are these changes tested? Yes. - `test_hashjoin_dynamic_filter_pushdown_is_used` is renamed to `test_hashjoin_dynamic_filter_requires_probe_consumer` (the old name referred to the now-deprecated `is_used()`) and strengthened: with no consumer the join now produces no dynamic filter at all, rather than producing one nothing reads. - New `test_hashjoin_dynamic_filter_survives_probe_subtree_replacement` reproduces the stage split — it runs filter pushdown, replaces the probe subtree with an equivalent scan that does not hold the filter, executes, and asserts the build-side bounds were still published. Both fail without the `exec.rs` change. Full workspace extended tests, sqllogictest, and `./dev/rust_lint.sh` pass. ## Are there any user-facing changes? One behavior change worth calling out: a `HashJoinExec` given a dynamic filter outside the filter pushdown rule (via the public `with_dynamic_filter_expr`) now computes it, where previously the runtime traversal could silently disable it. That is the point of the change — it is what lets a plan rewritten after optimization keep producing filters — but it does change the meaning of a public API, so this may warrant the `api change` label. A minor side effect: `gather_filters_for_pushdown` only pushes a self filter when `dynamic_filter.is_none()`, so on a plan with no consumer a repeated Post-phase run now creates and pushes a fresh filter instead of finding one already attached. Same result, slightly more work in replan loops. ## Note on overlapping work @jayshrivastava raised this in https://github.com/apache/datafusion/issues/18856#issuecomment-5359395043 and has #24528 open, which adds a third `PushedDown` state to reach the same goal. This is the smaller alternative: it removes the runtime check without changing the pushdown protocol. It is also only possible because of the `apply_expressions` work in #24018. Happy to close this in favour of that approach if preferred. -- 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]
