Compute Dynamic Filters only when a consumer supports them by LiaCastaneda · Pull Request #19546 · apache/datafusion

LiaCastaneda · 2025-12-29T13:16:51Z

Which issue does this PR close?

Rationale for this change

Currently, DataFusion computes bounds for all queries that contain a HashJoinExec node whenever the option enable_dynamic_filter_pushdown is set to true (default). It might make sense to compute these bounds only when we explicitly know there is a consumer that will use them.

What changes are included in this PR?

As suggested in #17527 (comment), this PR adds an is_used() method to DynamicFilterPhysicalExpr that checks if any consumers are holding a reference to the filter using Arc::strong_count().

During filter pushdown, consumers that accept the filter and use it later in execution have to retain a reference to Arc. For example, scan nodes like ParquetSource.

Are these changes tested?

I added a unit test in dynamic_filters.rs (test_is_used) that verifies the Arc reference counting behavior.
Existing integration tests in datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs validate the end-to-end behavior. These tests verify that dynamic filters are computed and filled when consumers are present.

Are there any user-facing changes?

new is_used() function

LiaCastaneda · 2025-12-29T14:14:16Z

This is another (desired) alternative to #18938
cc @adriangb this PR implements the is_used approach.

adriangb

I like this! The advantages over #19387 are:

No API change / breaking changes
Less code churn for us and users
Complexity is contained within dynamic filters and even there within producers
Should work for distributed systems (whatever is broadcasting updates to filters will also need to hold onto a reference to the dynamic filter)

This also means that if we run into issues with this approach it's easy to back out of 😄

Is there any way we can add a test showing that if there are no downstream consumers we don't compute the filters?

LiaCastaneda · 2025-12-29T15:22:24Z

Is there any way we can add a test showing that if there are no downstream consumers we don't compute the filters?

I added test_hashjoin_dynamic_filter_pushdown_not_used that creates a TestScanNode with support == false in the probe node and with enable_dynamic_filter_pushdown enabled so this variable

let enable_dynamic_filter_pushdown = context
            .session_config()
            .options()
            .optimizer
            .enable_join_dynamic_filter_pushdown
            && self
                .dynamic_filter
                .as_ref()
                .map(|df| df.filter.is_used())
                .unwrap_or(false);

should still return false even if the config option is enabled

adriangb

Maybe append to the same test the positive case (probe side does support pushdown, is_used is true) just to prove the point? Could even be in a loop to avoid code duplication.

LiaCastaneda · 2025-12-29T17:00:16Z

+        let _consumer = Arc::clone(&dynamic_filter)
+            .with_new_children(vec![])
+            .unwrap();


I had to add a consumer in these tests, otherwise is_used will return false, no filters will be computed and wait_complete will never return. I will add an is_used check inside wait_complete as well, I can't imagine this ever happenning (unless we call wait_complete on a probe node that does not accept dynamic filters which would be wrong usage) but its worth adding just in case.

adriangb · 2025-12-31T01:06:19Z

@LiaCastaneda thank you! Maybe a nice follow up would be to split up the CASE structure so that each dynamic piece is its own unit and can be computed independently, and maybe remove the barrier? I can write an issue to describe.

LiaCastaneda · 2025-12-31T08:59:31Z

By computed independently and removing the barrier, do you mean computing and emitting each filter for each partition progressively?

adriangb · 2025-12-31T13:34:06Z

I opened #19580 😄
There is also #16973 which may interest you.

LiaCastaneda · 2025-12-31T15:08:16Z

nice! I will take look

LiaCastaneda · 2026-01-22T12:53:51Z

I shot myself in the foot with this PR and noticed it when trying to upgrade DataFusion in our service 😿 . I made a small follow-up for this #19937.

## Which issue does this PR close? ## Rationale for this change The current v52 signature `pub async fn wait_complete(self: &Arc<Self>)` (introduced in #19546) is a bit unergonomic. The method requires `&Arc<DynamicFilterPhysicalExpr>`, but when working with `Arc<dyn PhysicalExpr>`, downcasting only gives you `&DynamicFilterPhysicalExpr`. Since you can't convert `&DynamicFilterPhysicalExpr` to `Arc<DynamicFilterPhysicalExpr>`, the method becomes impossible to call. The `&Arc<Self>` param was used to check` is_used()` via Arc strong count, but this was overly defensive. ## What changes are included in this PR? - Changed `DynamicFilterPhysicalExpr::wait_complete` signature from `pub async fn wait_complete(self: &Arc<Self>)` to `pub async fn wait_complete(&self)`. - Removed the `is_used()` check from `wait_complete()` - this method, like `wait_update()`, should only be called on filters that have consumers. If the caller doesn't know whether the filter has consumers, they should call `is_used()` first to avoid waiting indefinitely. This approach avoids complex signatures and dependencies between the APIs methods. ## Are these changes tested? Yes, existing tests cover this functionality, I removed the "mock" consumer from `test_hash_join_marks_filter_complete_empty_build_side` and `test_hash_join_marks_filter_complete` since the fix in #19734 makes is_used check the outer struct `strong_count` as well. ## Are there any user-facing changes? The signature of `wait_complete` changed.

## Which issue does this PR close? ## Rationale for this change The current v52 signature `pub async fn wait_complete(self: &Arc<Self>)` (introduced in apache#19546) is a bit unergonomic. The method requires `&Arc<DynamicFilterPhysicalExpr>`, but when working with `Arc<dyn PhysicalExpr>`, downcasting only gives you `&DynamicFilterPhysicalExpr`. Since you can't convert `&DynamicFilterPhysicalExpr` to `Arc<DynamicFilterPhysicalExpr>`, the method becomes impossible to call. The `&Arc<Self>` param was used to check` is_used()` via Arc strong count, but this was overly defensive. ## What changes are included in this PR? - Changed `DynamicFilterPhysicalExpr::wait_complete` signature from `pub async fn wait_complete(self: &Arc<Self>)` to `pub async fn wait_complete(&self)`. - Removed the `is_used()` check from `wait_complete()` - this method, like `wait_update()`, should only be called on filters that have consumers. If the caller doesn't know whether the filter has consumers, they should call `is_used()` first to avoid waiting indefinitely. This approach avoids complex signatures and dependencies between the APIs methods. ## Are these changes tested? Yes, existing tests cover this functionality, I removed the "mock" consumer from `test_hash_join_marks_filter_complete_empty_build_side` and `test_hash_join_marks_filter_complete` since the fix in apache#19734 makes is_used check the outer struct `strong_count` as well. ## Are there any user-facing changes? The signature of `wait_complete` changed. (cherry picked from commit bef1368)

* Fix dynamic filter is_used function (apache#19734) ## Which issue does this PR close?  - Closes apache#19715. ## Rationale for this change The:is_used() API incorrectly returned false for custom `DataSource` implementations that didn't call reassign_expr_columns() -> with_new_children() . This caused `HashJoinExec` to skip computing dynamic filters even when they were actually being used. ## What changes are included in this PR? Updated is_used() to check both outer and inner Arc counts ## Are these changes tested? Functionality is covered by existing test `test_hashjoin_dynamic_filter_pushdown_is_used`. I was not sure if to add a repro since it would require adding a custom `DataSource`, the current tests in datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs use `FileScanConfig` ## Are there any user-facing changes? no (cherry picked from commit 278950a) * Simplify wait_complete function (apache#19937) ## Which issue does this PR close? ## Rationale for this change The current v52 signature `pub async fn wait_complete(self: &Arc<Self>)` (introduced in apache#19546) is a bit unergonomic. The method requires `&Arc<DynamicFilterPhysicalExpr>`, but when working with `Arc<dyn PhysicalExpr>`, downcasting only gives you `&DynamicFilterPhysicalExpr`. Since you can't convert `&DynamicFilterPhysicalExpr` to `Arc<DynamicFilterPhysicalExpr>`, the method becomes impossible to call. The `&Arc<Self>` param was used to check` is_used()` via Arc strong count, but this was overly defensive. ## What changes are included in this PR? - Changed `DynamicFilterPhysicalExpr::wait_complete` signature from `pub async fn wait_complete(self: &Arc<Self>)` to `pub async fn wait_complete(&self)`. - Removed the `is_used()` check from `wait_complete()` - this method, like `wait_update()`, should only be called on filters that have consumers. If the caller doesn't know whether the filter has consumers, they should call `is_used()` first to avoid waiting indefinitely. This approach avoids complex signatures and dependencies between the APIs methods. ## Are these changes tested? Yes, existing tests cover this functionality, I removed the "mock" consumer from `test_hash_join_marks_filter_complete_empty_build_side` and `test_hash_join_marks_filter_complete` since the fix in apache#19734 makes is_used check the outer struct `strong_count` as well. ## Are there any user-facing changes? The signature of `wait_complete` changed. (cherry picked from commit bef1368)

## Which issue does this PR close? ## Rationale for this change The current v52 signature `pub async fn wait_complete(self: &Arc<Self>)` (introduced in apache#19546) is a bit unergonomic. The method requires `&Arc<DynamicFilterPhysicalExpr>`, but when working with `Arc<dyn PhysicalExpr>`, downcasting only gives you `&DynamicFilterPhysicalExpr`. Since you can't convert `&DynamicFilterPhysicalExpr` to `Arc<DynamicFilterPhysicalExpr>`, the method becomes impossible to call. The `&Arc<Self>` param was used to check` is_used()` via Arc strong count, but this was overly defensive. ## What changes are included in this PR? - Changed `DynamicFilterPhysicalExpr::wait_complete` signature from `pub async fn wait_complete(self: &Arc<Self>)` to `pub async fn wait_complete(&self)`. - Removed the `is_used()` check from `wait_complete()` - this method, like `wait_update()`, should only be called on filters that have consumers. If the caller doesn't know whether the filter has consumers, they should call `is_used()` first to avoid waiting indefinitely. This approach avoids complex signatures and dependencies between the APIs methods. ## Are these changes tested? Yes, existing tests cover this functionality, I removed the "mock" consumer from `test_hash_join_marks_filter_complete_empty_build_side` and `test_hash_join_marks_filter_complete` since the fix in apache#19734 makes is_used check the outer struct `strong_count` as well. ## Are there any user-facing changes? The signature of `wait_complete` changed.

@jayshrivastava

…che#24601) ## Which issue does this PR close? - Related to apache#18856 - Informs datafusion-contrib/datafusion-distributed#634 Does not close apache#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 (apache#17527: skip build-side bounds accumulation when nothing will read the result). Its placement in `execute()` is a leftover from apache#19546, which implemented it as `Arc::strong_count`, a signal only meaningful once the whole plan is assembled. Since apache#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 apache#18856 (comment) and has apache#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 apache#24018. Happy to close this in favour of that approach if preferred. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: Jayant Shrivastava <jshrivastava03@gmail.com>

LiaCastaneda added 2 commits December 29, 2025 12:14

Compute Dynamic Filters only when a consumer supports them

e116a9e

use strong_count of inner struct instead

e6130c8

github-actions Bot added physical-expr Changes to the physical-expr crates physical-plan Changes to the physical-plan crate labels Dec 29, 2025

Adjust unit test

1097b7d

LiaCastaneda marked this pull request as ready for review December 29, 2025 14:14

adriangb approved these changes Dec 29, 2025

View reviewed changes

add integration test

061d2b1

github-actions Bot added the core Core DataFusion crate label Dec 29, 2025

Fix cargo doc

c33c073

adriangb reviewed Dec 29, 2025

View reviewed changes

Comment thread datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs Outdated

Change testing approach

bb1db12

adriangb approved these changes Dec 29, 2025

View reviewed changes

LiaCastaneda added 3 commits December 29, 2025 17:13

test when is_used should be true and false

392b7c9

clippy

0ef0893

Adjust test in exec.rs

95581d3

LiaCastaneda commented Dec 29, 2025

View reviewed changes

LiaCastaneda added 2 commits December 29, 2025 18:16

Add is_used check inside wait_complete

a035dd1

test name

c1712bf

adriangb added this pull request to the merge queue Dec 31, 2025

Merged via the queue into apache:main with commit f1e5c94 Dec 31, 2025
32 checks passed

LiaCastaneda mentioned this pull request Dec 31, 2025

Compute Dynamic Filters only when a consumer supports them #18938

Closed

LiaCastaneda mentioned this pull request Jan 22, 2026

Simplify wait_complete function #19937

Merged

adriangb mentioned this pull request Aug 24, 2026

fix: decide hash join dynamic filter production at planning time #24601

Merged

Navigation Menu

Uh oh!

Compute Dynamic Filters only when a consumer supports them - #19546

Compute Dynamic Filters only when a consumer supports them#19546
adriangb merged 11 commits into
apache:mainfrom
LiaCastaneda:lia/compute-dynamic-filters-only-when-consumer-supports-them

LiaCastaneda commented Dec 29, 2025

Uh oh!

LiaCastaneda commented Dec 29, 2025

Uh oh!

adriangb left a comment •

edited

Loading

Uh oh!

LiaCastaneda commented Dec 29, 2025 •

edited

Loading

Uh oh!

Uh oh!

adriangb left a comment

Uh oh!

LiaCastaneda Dec 29, 2025 •

edited

Loading

Uh oh!

Uh oh!

adriangb commented Dec 31, 2025

Uh oh!

LiaCastaneda commented Dec 31, 2025

Uh oh!

adriangb commented Dec 31, 2025

Uh oh!

LiaCastaneda commented Dec 31, 2025

Uh oh!

LiaCastaneda commented Jan 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

LiaCastaneda commented Dec 29, 2025

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

LiaCastaneda commented Dec 29, 2025

Uh oh!

adriangb left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LiaCastaneda commented Dec 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

adriangb left a comment

Choose a reason for hiding this comment

Uh oh!

LiaCastaneda Dec 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

adriangb commented Dec 31, 2025

Uh oh!

LiaCastaneda commented Dec 31, 2025

Uh oh!

adriangb commented Dec 31, 2025

Uh oh!

LiaCastaneda commented Dec 31, 2025

Uh oh!

LiaCastaneda commented Jan 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

adriangb left a comment •

edited

Loading

LiaCastaneda commented Dec 29, 2025 •

edited

Loading

LiaCastaneda Dec 29, 2025 •

edited

Loading