Enable dynamic filter pushdown for LEFT/RIGHT/SEMI/ANTI/Mark joins; surface probe metadata in plans; add join-preservation docs #17090

kosiew · 2025-08-08T11:20:39Z

Which issue does this PR close?

Closes Enable dynamic filter pushdown for non-inner joins in physical plans #16973

Rationale for this change

Dynamic filter pushdown can significantly reduce I/O and compute by pruning non‑matching partitions/files on the fly. Previously this optimization was limited in scope. This PR expands it to additional join types and clarifies which input is pruned for each join. It also makes execution plans more transparent by explicitly annotating the probe side/keys for HashJoinExec, aiding debugging and user understanding.

What changes are included in this PR?

Optimizer / Execution
- Enable dynamic filter pushdown for LEFT, RIGHT, SEMI, ANTI, and LeftMark joins (full joins remain unsupported; only equi-join keys contribute).
- HashJoinExec now records and exposes probe_side and probe_keys in displayed physical plans to reflect runtime pruning behavior.
- Breaking: DynamicFilterPhysicalExpr::update now requires an additional key_count argument to handle composite/equi key tracking.
Tests
- Update sqllogictest expected plans to include probe_side=... and probe_keys=... across tpch and other suites (e.g., predicates.slt, subquery.slt, union.slt, and multiple tpch/plans/q*.slt.part).

This PR updates 30 slt, slt.part files.

Documentation
- Add docs/source/library-user-guide/join-preservation.md summarizing which join sides are preserved for output filtering and ON‑clause evaluation.
- Extend Upgrading guide with a new section describing dynamic filter pushdown for joins, configuration knobs, limitations (null handling, non‑equi predicates), and examples.
- Update User config docs (configs.md) to clarify the semantics of datafusion.optimizer.enable_dynamic_filter_pushdown, including the dynamic filter target per join type and prerequisites for pushdown (e.g., Parquet pushdown enabled).
- Draft 50.0.0 changelog entry detailing the breaking API change and new dynamic filter pushdown behavior, with guidance and references.

Are these changes tested?

Yes. This PR updates the sqllogictest goldens to reflect the new plan annotations and the broader dynamic filter pushdown coverage:

Multiple updated .slt and tpch plan snapshots now show probe_side/probe_keys on HashJoinExec where applicable.
Existing query result correctness is preserved; changes are confined to physical plan shape/metadata and runtime pruning behavior.
Formats without predicate pushdown (CSV/JSON) continue to pass unchanged semantically; they simply do not benefit from pruning.

Are there any user-facing changes?

Yes (docs/observability only):
- Physical plans printed in EXPLAIN now include probe_side and probe_keys for HashJoinExec.
- New user‑guide page on join preservation and expanded configuration docs.
- Changelog entry for 50.0.0.
Breaking API change:
- DynamicFilterPhysicalExpr::update requires a new key_count argument. Downstream users implementing custom dynamic filters must update their code.
- Please add the api change label.
Behavioral notes (non‑breaking):
- Dynamic filter pushdown across more join types is enabled by default and can be disabled via datafusion.optimizer.enable_dynamic_filter_pushdown=false.
- Effectiveness depends on source format pushdown support (e.g., Parquet with execution.parquet.pushdown_filters=true).
- Only equi‑join conjuncts generate dynamic filters; full joins and non‑equi predicates remain out of scope for now.

feat: enhance predicate handling for join optimization - Retain inferred predicates that cannot be pushed through joins as join filters for dynamic filter pushdown. - Update join filter assertions in tests to reflect new logic. - Add tests for dynamic filter pushdown scenarios, including: - Left join with a filter on the preserved side. - Right join with a filter on the preserved side. - Handling filters that do not restrict nulls. ```

…ecution plan clarity

Dandandan · 2025-08-09T07:11:07Z

datafusion/optimizer/src/push_down_filter.rs

@@ -2730,7 +2787,7 @@ mod tests {
        assert_optimized_plan_equal!(
            plan,
            @r"
-        Right Join: Using test.a = test2.a
+        Right Join: Using test.a = test2.a Filter: test.a <= Int64(1)


These filters seem redundant (always true)?

adriangb · 2025-08-11T23:05:06Z

Does this actually close #16973? I don't see that it changes filter pushdown at the physical plan level at all which is what #16973 is talking about. The example in that issue involves a TopK and HashJoin operator.

kosiew · 2025-08-12T09:44:08Z

@Dandandan , @adriangb ,

This PR is not ready for review yet.

…xec tests

…ceExec output

kosiew · 2025-08-15T09:25:21Z

@adriangb
PR is ready for review.

…ces in filter pushdown tests

…accessor for TestSource

adriangb · 2025-08-15T12:52:26Z

This is a monumental piece of work, I’m astounded! Thank you so much for working on this.

I’ll try to review it but I immediately will ask if we can somehow split it up into multiple logical PRs to make it easier to review and isolate future issues that may pop up.

I’m also wondering if you’ve seen #17196 and #17188. I imagine this PR may fix them but it would be nice to fix those bugs first with a more targeted change / smaller PR first before adding more complexity.

adriangb · 2025-08-15T18:45:57Z

I also think it's important that we get something like #17177 in place before adding more support to avoid regressions

adriangb

Amazing work overall! A lot of the diff is updating debug outputs / slt tests. I think it will help a lot to split this up into multiple PRs so that e.g. that can be reviewed separately from a smaller diff with complex logic changes.

adriangb · 2025-08-15T18:48:56Z

datafusion/common/src/join_type.rs

@@ -111,6 +111,63 @@ impl JoinType {
                | JoinType::RightAnti
        )
    }
+    /// Returns true if the left side of this join preserves its input rows
+    /// for filters applied *after* the join.
+    #[inline]


I see a lot of use of #[inline]. My understanding is that without specific evidence that it helps performance it may actually sometimes hurt it and it's best to not throw it around unless it's very obvious or can be proven to help performance.

adriangb · 2025-08-15T18:49:36Z

datafusion/common/src/join_type.rs

+    /// Returns true if the left side of this join preserves its input rows
+    /// for filters applied *after* the join.
+    #[inline]
+    pub const fn preserves_left_for_output_filters(self) -> bool {


I arrived at this same refactor in #17153 - I think it's a good one. Can we pull this out into its own PR?

adriangb · 2025-08-15T18:50:01Z

datafusion/core/tests/dataframe/mod.rs

+    |               |       Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]                                                                                               |
+    |               |         TableScan: t2 projection=[]                                                                                                                   |
+    | physical_plan | CoalesceBatchesExec: target_batch_size=8192                                                                                                           |
+    |               |   HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(count(*)@0, CAST(t1.a AS Int64)@2)], projection=[a@0, b@1], probe_side=Left, probe_keys=0 |


Can we pull out the addition of these fields to the debug output into its own PR?

adriangb · 2025-08-15T18:51:01Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

+    /// Number of keys currently contained in this dynamic filter.
+    /// Uses relaxed atomics as this counter is for diagnostics only.
+    key_count: Arc<AtomicUsize>,


If I'm just reading this module I'd have no idea what key_count is. Is it a join specific thing? What is a "key" in this context? If we actually do need this I propose we add it in an isolated PR.

adriangb · 2025-08-15T18:51:24Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -217,8 +234,25 @@ impl DynamicFilterPhysicalExpr {
        current.expr = new_expr;
        // Increment the generation to indicate that the expression has changed.
        current.generation += 1;
+        // Relaxed ordering is sufficient as `key_count` is only used for
+        // observability and does not synchronize with other data.


observability is super vague - what is it actually used for?

adriangb · 2025-08-15T18:52:37Z

datafusion/physical-optimizer/src/filter_pushdown.rs

@@ -708,6 +711,7 @@ impl<T: Clone> FilteredVec<T> {
    }
 }

+#[inline]


Same comment about putting inlines without any obvious justification

adriangb · 2025-08-15T18:54:30Z

docs/source/library-user-guide/join-preservation.md

+# Join preservation
+
+Dynamic filter pushdown and other optimizations rely on whether a join preserves
+rows from its inputs. The tables below summarise which sides are preserved for
+post-join output filtering and for evaluation of `ON`-clause predicates.
+
+## Output filtering


Thanks so much. This is an instant commit / approve (with verification that it's correct) if made as it's own PR.

adriangb · 2025-08-15T18:55:45Z

dev/changelog/50.0.0.md

+  [tests](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/hash_join.rs#L2033-L2049)). Formats without predicate pushdown (CSV/JSON) will not benefit.
+  Full joins and non‑equi (range or composite) predicates are not yet supported;
+  see [#7955](https://github.com/apache/datafusion/issues/7955). Dynamic filters
+  add planning overhead for high-cardinality keys; disable via:


Could you help me understand where the overhead comes from at the moment? Is it measurable at all?

adriangb · 2025-08-15T18:56:07Z

dev/changelog/50.0.0.md

+- Enable dynamic filter pushdown for left, right, semi, anti, and mark joins
+  [#16445](https://github.com/apache/datafusion/pull/16445) (adriangb). Mark joins
+  push filters to the side opposite the preserved input (`dynamic_filter_side`; see
+  [tests](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/hash_join.rs#L2033-L2049)). Formats without predicate pushdown (CSV/JSON) will not benefit.


Actually if they have statistics they will still benefit (those two generally don't)

adriangb · 2025-08-15T18:57:15Z

datafusion/physical-plan/src/topk/mod.rs

-            }
+            filter.update(predicate, self.heap.len())?;
+        } else {
+            // Even when the dynamic predicate is a tautology we still update


tautology is a very fancy word - would it be possible to give an explanation of what the term means?

alamb · 2025-08-15T20:34:55Z

Amazing work overall! A lot of the diff is updating debug outputs / slt tests. I think it will help a lot to split this up into multiple PRs so that e.g. that can be reviewed separately from a smaller diff with complex logic changes.

I agree with @adriangb that splitting this into multiple PRs would make this much easier to thoroughly review

From my past experience, handling pushdown for outerjoins in general is quite subtle (especially in the presence of nulls, etc) and I have chased down many very subtle bugs in my past lives. Testing (with nulls) is especially important too

adriangb · 2025-08-15T22:20:00Z

Testing (with nulls) is especially important too

Yes a LOT of specific tests + fuzz tests are going to be needed to be certain we don't introduce bugs

kosiew · 2025-08-17T03:57:34Z

Thanks @adriangb , @alamb for your review.

Putting this into draft and I will split this into smaller PRs.

adriangb · 2025-08-19T23:22:27Z

Thanks @adriangb , @alamb for your review.

Putting this into draft and I will split this into smaller PRs.

Amazing, thank you! Also please consider working on the fuzz tests - they'll help make sure we don't create bugs.

kosiew · 2025-08-20T10:47:17Z

@adriangb ,

please consider working on the fuzz tests - they'll help make sure we don't create bugs.

Do you mean #17177 uncovered some edge cases that Dynamic Filter push down should address?

adriangb · 2025-08-20T11:28:58Z

@adriangb ,

please consider working on the fuzz tests - they'll help make sure we don't create bugs.

Do you mean #17177 uncovered some edge cases that Dynamic Filter push down should address?

No I'm just saying we do need to merge that or similar before making more changes

kosiew added 2 commits August 8, 2025 19:18

Merge branch 'main' into non-inner-join-filter-pushdown-16973

2a194f1

github-actions bot added the optimizer Optimizer rules label Aug 8, 2025

Refactor join logic in test file to improve filter application and ex…

15c8897

…ecution plan clarity

github-actions bot added the sqllogictest SQL Logic Tests (.slt) label Aug 8, 2025

Fix filter application order in RightAnti join execution plan

c960bda

github-actions bot added the logical-expr Logical plan and expressions label Aug 9, 2025

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from 6dddfc6 to 8337f80 Compare August 9, 2025 05:32

github-actions bot added the core Core DataFusion crate label Aug 9, 2025

Dandandan reviewed Aug 9, 2025

View reviewed changes

github-actions bot removed logical-expr Logical plan and expressions core Core DataFusion crate labels Aug 9, 2025

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from 9f9f0ac to 5e4472d Compare August 9, 2025 10:15

github-actions bot added logical-expr Logical plan and expressions core Core DataFusion crate labels Aug 9, 2025

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from f8310ec to 5ae7863 Compare August 9, 2025 12:21

kosiew changed the title ~~Improve Filter Handling in Join Optimization: Retain Inferred Predicates as Join Filters~~ DRAFT - Improve Filter Handling in Join Optimization: Retain Inferred Predicates as Join Filters Aug 12, 2025

kosiew added 5 commits August 12, 2025 19:06

Enhance dynamic filter pushdown documentation for joins pushdown

1076e8b

Implement dynamic filter pushdown for various join types in HashJoinE…

fb1c85c

…xec tests

Implement dynamic filter handling for various join types in HashJoinExec

adf1207

Refactor static filter pushdown test to include predicate in DataSour…

081bed0

…ceExec output

Add predicate information to explain tree output for parquet format

4675cbc

kosiew added 2 commits August 15, 2025 16:54

refactor(tests): remove redundant hash join parent filter pushdown test

1395b49

refactor(tests): consolidate tests

4799c01

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from 01a8636 to 4799c01 Compare August 15, 2025 08:55

kosiew changed the title ~~DRAFT - Expand Dynamic Filter Pushdown to Additional Join Types with Preservation Safety and Plan Display Enhancements~~ Enable dynamic filter pushdown for LEFT/RIGHT/SEMI/ANTI/Mark joins; surface probe metadata in plans; add join-preservation docs Aug 15, 2025

kosiew marked this pull request as ready for review August 15, 2025 09:24

kosiew requested a review from Dandandan August 15, 2025 09:24

kosiew added 2 commits August 15, 2025 18:15

refactor(tests): clean up imports and remove unused join type referen…

b2cc0a5

…ces in filter pushdown tests

Rearrange tests to minimize diff

7a3c6db

kosiew mentioned this pull request Aug 15, 2025

Dynamic filter no-op for NULL-only build keys (NullEqualsNothing) #17206

Open

docs(tests): add issue reference for dynamic filter pushdown test

08a0e29

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from 17f3d21 to 08a0e29 Compare August 15, 2025 10:56

refactor(tests): add dynamic filter location assertion and predicate …

7b6f778

…accessor for TestSource

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from a436167 to 80c0f29 Compare August 15, 2025 11:29

fix tests

3023d88

kosiew force-pushed the non-inner-join-filter-pushdown-16973 branch from 80c0f29 to 3023d88 Compare August 15, 2025 11:40

kosiew added 3 commits August 15, 2025 19:51

refactor(tests): remove unused predicate accessor from TestSource

4beb2f7

test: add async test for hash join with probe filter

f999914

Merge branch 'main' into non-inner-join-filter-pushdown-16973

704aba7

adriangb reviewed Aug 15, 2025

View reviewed changes

kosiew marked this pull request as draft August 17, 2025 03:56

adriangb mentioned this pull request Aug 17, 2025

Blog post about TopK filter pushdown #15513

Open

Enable dynamic filter pushdown for LEFT/RIGHT/SEMI/ANTI/Mark joins; surface probe metadata in plans; add join-preservation docs #17090

Are you sure you want to change the base?

Enable dynamic filter pushdown for LEFT/RIGHT/SEMI/ANTI/Mark joins; surface probe metadata in plans; add join-preservation docs #17090

Uh oh!

Conversation

kosiew commented Aug 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

adriangb commented Aug 11, 2025

Uh oh!

kosiew commented Aug 12, 2025

Uh oh!

kosiew commented Aug 15, 2025

Uh oh!

adriangb commented Aug 15, 2025

Uh oh!

adriangb commented Aug 15, 2025

Uh oh!

adriangb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb commented Aug 15, 2025

Uh oh!

adriangb commented Aug 15, 2025

Uh oh!

kosiew commented Aug 17, 2025

Uh oh!

adriangb commented Aug 19, 2025

Uh oh!

kosiew commented Aug 20, 2025

Uh oh!

adriangb commented Aug 20, 2025

Uh oh!

Uh oh!

kosiew commented Aug 8, 2025 •

edited

Loading