You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/24 17:13:10 UTC
(datafusion) branch gh-readonly-queue/main/pr-24601-2326917aa8d691aa5123518ca602ce63ab18ffc6 created (now a62e84a8db)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24601-2326917aa8d691aa5123518ca602ce63ab18ffc6
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a62e84a8db fix: decide hash join dynamic filter production at planning time (#24601)
This branch includes the following new commits:
new a62e84a8db fix: decide hash join dynamic filter production at planning time (#24601)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: decide hash join dynamic filter production at planning time (#24601)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24601-2326917aa8d691aa5123518ca602ce63ab18ffc6
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a62e84a8db793138919fadc2935bd99e309a99c6
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Mon Aug 24 17:12:46 2026 +0000
fix: decide hash join dynamic filter production at planning time (#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.
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
Co-authored-by: Jayant Shrivastava <js...@gmail.com>
---
.../tests/physical_optimizer/filter_pushdown.rs | 225 ++++++++++++++++++---
datafusion/physical-plan/src/execution_plan.rs | 7 +-
.../physical-plan/src/joins/hash_join/exec.rs | 61 +++---
datafusion/physical-plan/src/lib.rs | 4 +-
4 files changed, 240 insertions(+), 57 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
index 3c6f554a94..26e6e0c74c 100644
--- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
+++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
@@ -66,6 +66,9 @@ use datafusion_physical_plan::{
aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy},
coalesce_partitions::CoalescePartitionsExec,
collect,
+ execution_plan::{
+ ChildrenPropertiesMode, ReplaceChildrenOptions, plan_contains_expression_id,
+ },
filter::{FilterExec, FilterExecBuilder},
joins::{HashJoinExec, PartitionMode},
projection::ProjectionExec,
@@ -2964,29 +2967,12 @@ async fn test_hashjoin_hash_table_pushdown_collect_left() {
);
}
-// Not portable to sqllogictest: verifies whether the optimized probe-side plan
-// retains the HashJoinExec's dynamic filter expression. The with_support(false)
-// branch has no SQL analog because parquet supports filter pushdown.
+// Not portable to sqllogictest: verifies that the HashJoinExec only keeps its
+// dynamic filter when the optimized probe-side plan retains the expression, i.e.
+// when there is something to consume it. The with_support(false) branch has no
+// SQL analog because parquet supports filter pushdown.
#[test]
-fn test_hashjoin_dynamic_filter_pushdown_is_used() {
- fn contains_expression_id(plan: &Arc<dyn ExecutionPlan>, expression_id: u64) -> bool {
- let mut found = false;
- plan.apply(|node| {
- node.apply_expressions(&mut |root| {
- root.apply(|expr| {
- if expr.expression_id() == Some(expression_id) {
- found = true;
- Ok(TreeNodeRecursion::Stop)
- } else {
- Ok(TreeNodeRecursion::Continue)
- }
- })
- })
- })
- .unwrap();
- found
- }
-
+fn test_hashjoin_dynamic_filter_requires_probe_consumer() {
for (probe_supports_pushdown, expected_consumer) in [(false, false), (true, true)] {
let build_side_schema = Arc::new(Schema::new(vec![
Field::new("a", DataType::Utf8, false),
@@ -3050,20 +3036,199 @@ fn test_hashjoin_dynamic_filter_pushdown_is_used() {
.downcast_ref::<HashJoinExec>()
.expect("Plan should be HashJoinExec");
let dynamic_filters = hash_join.dynamic_expressions_produced();
- let expression_id = dynamic_filters
- .first()
- .expect("Dynamic filter should be created")
- .expression_id()
- .expect("Dynamic filters always have an expression ID");
+ // The join keeps a dynamic filter only if the pushdown left a consumer for it
+ // in the probe subtree. Otherwise it is dropped at planning time so that
+ // `execute` skips build side bounds accumulation entirely.
assert_eq!(
- contains_expression_id(hash_join.right(), expression_id),
- expected_consumer,
- "probe consumer should be {expected_consumer} when pushdown support is {probe_supports_pushdown}"
+ dynamic_filters.len(),
+ usize::from(expected_consumer),
+ "dynamic filter should {}be produced when pushdown support is {probe_supports_pushdown}",
+ if expected_consumer { "" } else { "not " }
);
+
+ if let Some(dynamic_filter) = dynamic_filters.first() {
+ let expression_id = dynamic_filter
+ .expression_id()
+ .expect("Dynamic filters always have an expression ID");
+ assert!(
+ plan_contains_expression_id(hash_join.right(), expression_id).unwrap(),
+ "probe subtree should contain the dynamic filter it accepted"
+ );
+ }
}
}
+// Not portable to sqllogictest: stands in for a distributed planner that splits
+// an already-optimized plan into stages, leaving the HashJoinExec and the scan
+// consuming its dynamic filter on different workers. Whether to produce the
+// filter is decided during pushdown, so it has to survive the probe subtree
+// being swapped out afterwards.
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_survives_probe_subtree_replacement() {
+ let (build_side_schema, build_scan, probe_side_schema, probe_scan) =
+ hashjoin_pushdown_scans();
+
+ let on = vec![
+ (
+ col("a", &build_side_schema).unwrap(),
+ col("a", &probe_side_schema).unwrap(),
+ ),
+ (
+ col("b", &build_side_schema).unwrap(),
+ col("b", &probe_side_schema).unwrap(),
+ ),
+ ];
+ let plan = Arc::new(
+ HashJoinExec::try_new(
+ Arc::clone(&build_scan),
+ probe_scan,
+ on,
+ None,
+ &JoinType::Inner,
+ None,
+ PartitionMode::CollectLeft,
+ datafusion_common::NullEquality::NullEqualsNothing,
+ false,
+ )
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+
+ let mut config = ConfigOptions::default();
+ config.execution.parquet.pushdown_filters = true;
+ config.optimizer.enable_dynamic_filter_pushdown = true;
+ let plan = FilterPushdown::new_post_optimization()
+ .optimize(plan, &config)
+ .unwrap();
+
+ // The probe scan accepted the filter, so the join kept it.
+ assert_eq!(plan.dynamic_expressions_produced().len(), 1);
+
+ // Swap the probe subtree for one that does not hold the filter: in a real
+ // deployment the consumer ends up in another stage, on another worker, and is
+ // not reachable from this node at all.
+ let detached_probe = TestScanBuilder::new(Arc::clone(&probe_side_schema))
+ .with_support(true)
+ .with_batches(vec![
+ record_batch!(
+ ("a", Utf8, ["aa", "ab", "ac", "ad"]),
+ ("b", Utf8, ["ba", "bb", "bc", "bd"]),
+ ("e", Float64, [1.0, 2.0, 3.0, 4.0])
+ )
+ .unwrap(),
+ ])
+ .build();
+ let plan = plan
+ .replace_children(
+ vec![build_scan, detached_probe],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )
+ .unwrap();
+ assert_eq!(plan.dynamic_expressions_produced().len(), 1);
+
+ let session_ctx =
+ SessionContext::new_with_config(SessionConfig::from(config).with_batch_size(10));
+ session_ctx.register_object_store(
+ ObjectStoreUrl::parse("test://").unwrap().as_ref(),
+ Arc::new(InMemory::new()),
+ );
+ collect(Arc::clone(&plan), session_ctx.state().task_ctx())
+ .await
+ .unwrap();
+
+ // The build side bounds were still computed and published, so whatever holds
+ // the other end of this filter sees them.
+ let dynamic_filter = plan
+ .dynamic_expressions_produced()
+ .into_iter()
+ .next()
+ .expect("dynamic filter should be retained");
+ insta::assert_snapshot!(
+ format!("{dynamic_filter}"),
+ @"DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ]",
+ );
+}
+
+// Not portable to sqllogictest: custom `PhysicalOptimizerRule`s are appended
+// after the built in ones, so a user rule runs after the Post phase
+// `FilterPushdown` that decides whether to keep the join's dynamic filter. Such a
+// rule can bring a consumer back by re-running the pushdown; there is nothing to
+// undo first, because a join with no consumer holds no dynamic filter.
+//
+// Re-running the Post phase is an already supported mode rather than something
+// this test invents: see `post_phase_is_idempotent_on_hash_join` below, added by
+// apache/datafusion#22523 because AQE (datafusion-ballista#1359) re-runs the
+// optimizer chain after every completed stage.
+#[test]
+fn test_hashjoin_dynamic_filter_recreated_when_pushdown_reruns() {
+ let (build_side_schema, build_scan, probe_side_schema, _) = hashjoin_pushdown_scans();
+
+ let probe_batches = vec![
+ record_batch!(
+ ("a", Utf8, ["aa", "ab", "ac", "ad"]),
+ ("b", Utf8, ["ba", "bb", "bc", "bd"]),
+ ("e", Float64, [1.0, 2.0, 3.0, 4.0])
+ )
+ .unwrap(),
+ ];
+ let unsupported_probe = TestScanBuilder::new(Arc::clone(&probe_side_schema))
+ .with_support(false)
+ .with_batches(probe_batches.clone())
+ .build();
+
+ let on = vec![
+ (
+ col("a", &build_side_schema).unwrap(),
+ col("a", &probe_side_schema).unwrap(),
+ ),
+ (
+ col("b", &build_side_schema).unwrap(),
+ col("b", &probe_side_schema).unwrap(),
+ ),
+ ];
+ let plan = Arc::new(
+ HashJoinExec::try_new(
+ Arc::clone(&build_scan),
+ unsupported_probe,
+ on,
+ None,
+ &JoinType::Inner,
+ None,
+ PartitionMode::CollectLeft,
+ datafusion_common::NullEquality::NullEqualsNothing,
+ false,
+ )
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+
+ let mut config = ConfigOptions::default();
+ config.execution.parquet.pushdown_filters = true;
+ config.optimizer.enable_dynamic_filter_pushdown = true;
+
+ // Nothing in the probe side accepts the filter, so the join drops it.
+ let plan = FilterPushdown::new_post_optimization()
+ .optimize(plan, &config)
+ .unwrap();
+ assert_eq!(plan.dynamic_expressions_produced().len(), 0);
+
+ // A later rule swaps in a probe side that does accept filters and re-runs the
+ // pushdown. The join creates a fresh filter and finds its consumer.
+ let supported_probe = TestScanBuilder::new(Arc::clone(&probe_side_schema))
+ .with_support(true)
+ .with_batches(probe_batches)
+ .build();
+ let plan = plan
+ .replace_children(
+ vec![build_scan, supported_probe],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )
+ .unwrap();
+ let plan = FilterPushdown::new_post_optimization()
+ .optimize(plan, &config)
+ .unwrap();
+ assert_eq!(plan.dynamic_expressions_produced().len(), 1);
+}
+
/// Regression test for https://github.com/apache/datafusion/issues/20109.
///
/// Not portable to sqllogictest: the regression specifically targets the
diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index a4d081b3d9..c0fa4c6bed 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -1122,7 +1122,12 @@ where
///
/// This traverses both the execution plan and the children of each expression root
/// reported by [`ExecutionPlan::apply_expressions`].
-pub(crate) fn plan_contains_expression_id(
+///
+/// Producers of dynamic filters use this to find out whether anything downstream
+/// holds the filter they pushed, since a node that replies
+/// [`PushedDown::No`](crate::filter_pushdown::PushedDown::No) may still retain it
+/// for statistics pruning.
+pub fn plan_contains_expression_id(
plan: &Arc<dyn ExecutionPlan>,
expression_id: u64,
) -> Result<bool> {
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index aa06015be7..fb8077264a 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -999,6 +999,11 @@ impl HashJoinExec {
/// Set the dynamic filter on this hash join.
///
+ /// Setting a dynamic filter is what makes the join compute and publish
+ /// build-side bounds during execution. [`Self::handle_child_pushdown_result`]
+ /// sets one after finding a consumer for it in the probe side. Callers wiring a
+ /// filter up by hand take on that check themselves.
+ ///
/// Resets any internal state that depends on any existing dynamic filter.
///
/// Validates that the filter's children reference valid columns in
@@ -1449,20 +1454,9 @@ impl ExecutionPlan for HashJoinExec {
consider using CoalescePartitionsExec or the EnforceDistribution rule"
);
- // Only compute a dynamic filter when the probe subtree contains a consumer.
- // Searching from `self` would always find the producer expression owned by this join.
- let enable_dynamic_filter_pushdown = if self
+ let enable_dynamic_filter_pushdown = self
.allow_join_dynamic_filter_pushdown(context.session_config().options())
- {
- self.dynamic_filter
- .as_ref()
- .and_then(|df| df.filter.expression_id())
- .map(|id| plan_contains_expression_id(&self.right, id))
- .transpose()?
- .unwrap_or(false)
- } else {
- false
- };
+ && self.dynamic_filter.is_some();
let join_metrics = BuildProbeJoinMetrics::new(partition, &self.metrics);
@@ -1814,21 +1808,40 @@ impl ExecutionPlan for HashJoinExec {
let right_child_self_filters = &child_pushdown_result.self_filters[1]; // We only push down filters to the right child
// We expect 0 or 1 self filters
if let Some(filter) = right_child_self_filters.first() {
- // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said
- // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating
let predicate = Arc::clone(&filter.predicate);
if let Ok(dynamic_filter) =
Arc::downcast::<DynamicFilterPhysicalExpr>(predicate)
{
- // We successfully pushed down our self filter - we need to make a new node with the dynamic filter
- let new_node = self
- .builder()
- .with_dynamic_filter(Some(HashJoinExecDynamicFilter {
- filter: dynamic_filter,
- build_accumulator: OnceLock::new(),
- }))
- .build_exec()?;
- result = result.with_updated_node(new_node);
+ // Note that we don't check `PushedDownPredicate::discriminant`: a node
+ // that replies `PushedDown::No` may still retain the filter for
+ // statistics pruning, so the reply does not tell us whether anyone will
+ // actually read the filter. Instead we look for a consumer holding the
+ // expression in the probe subtree.
+ //
+ // `self` here is the join with its post-pushdown children, so anything
+ // that accepted the filter is already wired into `self.right`. Searching
+ // from `self` would always find the producer expression pushed by this
+ // join. This is the last chance to make the decision: the Post-phase
+ // `FilterPushdown` rule is the final rule that mutates the plan.
+ let has_consumer = dynamic_filter
+ .expression_id()
+ .map(|id| plan_contains_expression_id(&self.right, id))
+ .transpose()?
+ .unwrap_or(false);
+ if has_consumer {
+ // Our self filter reached a consumer: rebuild the node holding onto
+ // the dynamic filter so that `execute` populates it from the build
+ // side. If it did not, we leave `dynamic_filter` as `None` and skip
+ // the (not cheap) bounds accumulation entirely.
+ let new_node = self
+ .builder()
+ .with_dynamic_filter(Some(HashJoinExecDynamicFilter {
+ filter: dynamic_filter,
+ build_accumulator: OnceLock::new(),
+ }))
+ .build_exec()?;
+ result = result.with_updated_node(new_node);
+ }
}
}
Ok(result)
diff --git a/datafusion/physical-plan/src/lib.rs b/datafusion/physical-plan/src/lib.rs
index 673d43f9c1..5ff6cec374 100644
--- a/datafusion/physical-plan/src/lib.rs
+++ b/datafusion/physical-plan/src/lib.rs
@@ -49,8 +49,8 @@ pub use crate::execution_plan::{
AsPhysicalExprRef, ChildrenPropertiesMode, ExecutionPlan, ExecutionPlanProperties,
PlanProperties, ReplaceChildrenOptions, apply_expression_roots, collect,
collect_partitioned, displayable, execute_input_stream, execute_stream,
- execute_stream_partitioned, get_plan_string, replace_children_if_necessary,
- with_new_children_if_necessary,
+ execute_stream_partitioned, get_plan_string, plan_contains_expression_id,
+ replace_children_if_necessary, with_new_children_if_necessary,
};
pub use crate::metrics::Metric;
pub use crate::ordering::InputOrderMode;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org