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/05/30 04:27:30 UTC
(datafusion) branch gh-readonly-queue/main/pr-22523-b6d4c252634f6ab94b1363c751d7bbd101d86dd2 created (now 2e7b8e1b2d)
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-22523-b6d4c252634f6ab94b1363c751d7bbd101d86dd2
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 2e7b8e1b2d fix(physical-plan): make HashJoinExec dynamic filter pushdown idempotent (#22523)
This branch includes the following new commits:
new 2e7b8e1b2d fix(physical-plan): make HashJoinExec dynamic filter pushdown idempotent (#22523)
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(physical-plan): make HashJoinExec dynamic filter pushdown idempotent (#22523)
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-22523-b6d4c252634f6ab94b1363c751d7bbd101d86dd2
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 2e7b8e1b2d3cd800823b3813f2f7efdde0d21af8
Author: Xuanyi Li <sh...@outlook.com>
AuthorDate: Fri May 29 21:27:10 2026 -0700
fix(physical-plan): make HashJoinExec dynamic filter pushdown idempotent (#22523)
## Which issue does this PR close?
Related to https://github.com/apache/datafusion-ballista/issues/1359
## Rationale
Ballista's Adaptive Query Execution (AQE) planner re-invokes
DataFusion's full `PhysicalOptimizer` chain after every completed stage.
`FilterPushdown::new_post_optimization()` is not idempotent on plans
containing `HashJoinExec`.
In the `Post` phase, `HashJoinExec::gather_filters_for_pushdown`
unconditionally creates a new `DynamicFilterPhysicalExpr` and installs
it on the probe-side child via `with_self_filter`. After pass 1 the join
already carries a `dynamic_filter: Some(...)`, and the shared
`Arc<DynamicFilterPhysicalExpr>` is already wired into the probe-side
scan's predicate. On pass 2 a *second* dynamic filter is created and
ANDed onto the existing predicate, producing `DynamicFilter AND
DynamicFilter`. Each subsequent pass adds another duplicate, compounding
indefinitely in AQE replan loops.
## What changes are included in this PR?
- **Guard in `HashJoinExec::gather_filters_for_pushdown`**: skip
dynamic-filter creation when `self.dynamic_filter.is_some()`, meaning a
previous pass already installed one. The existing `Arc` remains valid
and correctly wired into the probe-side scan.
- **Comment** explaining why the guard is needed (AQE replan context).
- **Test** `post_phase_is_idempotent_on_hash_join` in
`tests/physical_optimizer/filter_pushdown.rs`: builds a `HashJoinExec`,
runs `FilterPushdown::new_post_optimization()` twice, and asserts
structural equality via `get_plan_string`.
## Are these changes tested?
Yes. The new test fails without the fix (plan strings diverge due to
duplicated dynamic filter predicates) and passes with it.
## Are there any user-facing changes?
No. Dynamic filter pushdown is an internal optimization; the idempotence
guard only affects re-optimization scenarios (AQE).
---------
Co-authored-by: Claude Opus 4.7 (1M context) <no...@anthropic.com>
---
.../tests/physical_optimizer/filter_pushdown.rs | 39 ++++++++++++++++++++++
.../physical-plan/src/joins/hash_join/exec.rs | 8 ++++-
2 files changed, 46 insertions(+), 1 deletion(-)
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
index b420326596..909b80cada 100644
--- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
+++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
@@ -3269,3 +3269,42 @@ fn test_filter_pushdown_through_sort_with_projection() {
"
);
}
+
+/// `FilterPushdown::new_post_optimization()` must be idempotent. When applied
+/// to a HashJoinExec, the rule installs a dynamic filter on the probe-side
+/// scan; before the fix in `HashJoinExec::gather_filters_for_pushdown`, every
+/// invocation created a *new* `DynamicFilterPhysicalExpr` and ANDed it onto
+/// the probe side's existing predicate, producing
+/// `DynamicFilter AND DynamicFilter AND ...` after N passes.
+///
+/// AQE (datafusion-ballista#1359) re-runs the optimizer chain after every
+/// completed stage, so this would compound indefinitely without the guard.
+#[test]
+fn post_phase_is_idempotent_on_hash_join() {
+ use crate::physical_optimizer::test_utils::{hash_join_exec, parquet_exec, schema};
+ use datafusion_common::JoinType;
+ use datafusion_physical_expr::expressions::Column;
+ use datafusion_physical_optimizer::filter_pushdown::FilterPushdown;
+ use datafusion_physical_plan::get_plan_string;
+ use datafusion_physical_plan::joins::utils::JoinOn;
+
+ let s = schema();
+ let left = parquet_exec(Arc::clone(&s));
+ let right = parquet_exec(Arc::clone(&s));
+ let join_on: JoinOn = vec![(
+ Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()),
+ Arc::new(Column::new_with_schema("a", &right.schema()).unwrap()),
+ )];
+ let plan = hash_join_exec(left, right, join_on, None, &JoinType::Inner).unwrap();
+
+ let config = ConfigOptions::new();
+ let rule = FilterPushdown::new_post_optimization();
+ let once = rule.optimize(plan, &config).unwrap();
+ let twice = rule.optimize(Arc::clone(&once), &config).unwrap();
+
+ assert_eq!(
+ get_plan_string(&once),
+ get_plan_string(&twice),
+ "second invocation of FilterPushdown::new_post_optimization mutated the plan",
+ );
+}
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index f7391feb29..03387c316b 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1635,8 +1635,14 @@ impl ExecutionPlan for HashJoinExec {
ChildFilterDescription::all_unsupported(&parent_filters)
};
- // Add dynamic filters in Post phase if enabled
+ // Add dynamic filters in Post phase if enabled. Skip when this join
+ // already carries a dynamic filter from a previous pass — the shared
+ // `Arc<DynamicFilterPhysicalExpr>` is still wired into the probe-side
+ // scan's predicate, and re-creating it would AND a fresh duplicate
+ // onto every Post-phase invocation (apache/datafusion-ballista#1359
+ // surfaces this in AQE replan loops).
if phase == FilterPushdownPhase::Post
+ && self.dynamic_filter.is_none()
&& self.allow_join_dynamic_filter_pushdown(config)
{
// Add actual dynamic filter to right side (probe side)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org