fix(physical-plan): make HashJoinExec dynamic filter pushdown idempotent (#22523)

## Which issue does this PR close?

Related to apache/datafusion-ballista#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) <noreply@anthropic.com>

Author: wirybeaver

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",
+    );
+}

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)