Refactor filter pushdown APIs to enable joins to pass through filters #16732

adriangb · 2025-07-10T00:41:19Z

Like #16642 this was triggered by attempting to make progress on HashJoinExec.

In particular I realized a fundamental assumption of the existing code, that the optimizer knows how to combine results from children, is flawed and wrong. This requires changes to the public APIs.

I've taken the opportunity to do some thinking and re-designing and while there is a bit more complexity in some places I think others are a lot simpler.

adriangb · 2025-07-10T00:54:51Z

Note that I added a HashJoinExec implementation to motivate this PR but remove it in 5940cca because it lacks nuance necessary for a final complete correct implementation (it doesn't take into account the join type, etc.).

I asked Claude to analyze the logical filter pushdown on joins and it reported the following:

  Join Types and Filter Push Down Rules

  For WHERE clause filters (lr_is_preserved):
  - Inner joins: Filters can be pushed to both sides
  - Left joins: Filters can only be pushed to the left side
  - Right joins: Filters can only be pushed to the right side
  - Full joins: Filters cannot be pushed to either side
  - Semi/Anti joins: Filters can be pushed to the preserved side only

  For ON clause filters (on_lr_is_preserved):
  - Inner joins: Filters can be pushed to both sides
  - Left joins: Filters can only be pushed to the right side
  - Right joins: Filters can only be pushed to the left side
  - Full joins: Filters cannot be pushed to either side
  - Semi/Anti joins: Different rules apply based on join variant

  Filter Restrictions

  The can_evaluate_as_join_condition function at line 255 shows that filters can only be converted to join conditions if they:

  Allowed expressions:
  - Column references
  - Literals
  - Placeholders
  - Scalar variables
  - Binary expressions
  - LIKE/SimilarTo predicates
  - NOT expressions
  - IS NULL/IS NOT NULL
  - CASE expressions
  - Cast expressions
  - Try cast expressions
  - Scalar functions

  Disallowed expressions:
  - Subqueries (EXISTS, IN subquery, scalar subquery)
  - Outer reference columns
  - UNNEST expressions

  Additionally, for non-inner joins, inferred predicates must strictly filter out NULLs to be pushed down to avoid incorrect results.

…pushdown

adriangb

leaving some notes

adriangb · 2025-07-10T01:15:49Z

datafusion/physical-plan/src/filter_pushdown.rs

-    /// The combined result of pushing down each parent filter into each child.
-    /// For example, given the fitlers `[a, b]` and children `[1, 2, 3]` the matrix of responses:
-    ///
-    // | filter | child 1     | child 2   | child 3   | result      |
-    // |--------|-------------|-----------|-----------|-------------|
-    // | a      | Supported   | Supported | Supported | Supported   |
-    // | b      | Unsupported | Supported | Supported | Unsupported |
-    ///


This is the assumption that was fundamentally flawed

adriangb · 2025-07-10T01:16:38Z

datafusion/physical-plan/src/filter_pushdown.rs

+    /// The parent filters that were pushed down as received by the current node when [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result) was called.
+    /// Note that this may *not* be the same as the filters that were passed to the children as the current node may have modified them
+    /// (e.g. by reassigning column indices) when it returned them from [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result) in a [`FilterDescription`].
+    /// Attached to each filter is a [`PredicateSupportDiscriminant`] *per child* that indicates whether the filter was supported or unsupported by each child.
+    /// To get combined results see [`ChildFitlerPushdownResult::any`] and [`ChildFitlerPushdownResult::all`].
+    pub parent_filters: Vec<ChildFitlerPushdownResult>,


I think part of the reason I made that assumption in the original design is was that I didn't want a Vec<Vec<PredicateSupport>>, in part because you could technically have different lengths or different filters. This new structure makes that much more sound.

adriangb · 2025-07-10T01:17:23Z

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

-        Ok(FilterPushdownPropagation::transparent(
-            child_pushdown_result,
-        ))
+        Ok(FilterPushdownPropagation::all(child_pushdown_result))


Note that all is much more explicit and clear than transparent. We could even name it something more verbose?

Maybe if_all?

adriangb · 2025-07-10T01:17:44Z

datafusion/physical-plan/src/filter.rs

+            filters: vec![
+                PredicateSupportDiscriminant::Supported;
+                child_pushdown_result.parent_filters.len()
+            ],


I think this is now nicer to make 😄

adriangb · 2025-07-10T01:18:22Z

datafusion/physical-plan/src/coalesce_batches.rs

-        Ok(FilterPushdownPropagation::transparent(
-            child_pushdown_result,
-        ))
+        Ok(FilterPushdownPropagation::all(child_pushdown_result))


For nodes with 1 child all and any are equivalent, but I think it's safer to pick all by default.

adriangb · 2025-07-10T01:18:48Z

datafusion/physical-optimizer/src/filter_pushdown.rs

-/// Support state of each predicate for the children of the node.
-/// These predicates are coming from the parent node.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum ParentPredicateStates {
-    NoChildren,
-    Unsupported,
-    Supported,
-}


I was able to get rid of this ugly helper enum

kosiew · 2025-07-10T01:27:27Z

quick change request:

ChildFitlerPushdownResult -> ChildFilterPushdownResult

alamb

I reviewed this code carefully and it looks good to me. Thank you for pushing this along @adriangb

Some thoughts (nothing required)

Documentation:

I made a PR with some documentation suggestions for your consideration: pydantic#32

Consolidate some structures?

I feel like there is a maze of many small structures that all are very similar but not quite the same. I was wondering if it would be possible to try and unify some of them

Specifically, FilterPushdownPropagation and ChildPushdownResult feel very similar and ChildPushdownResult and ChildFilterDescription

Naming

Something else that occurs to me with some fresh eyes is maybe we could change the naming to make things easier to follow

Something else that might help could be naming -- instead of "support" maybe we could use the term "Pushdown"

So instead of PredicateSupportDiscriminant

/// Used by [`FilterPushdownPropagation`] and [`PredicateSupport`] to
/// communicate the result of attempting to push down a filter into a child
/// node.
///
/// This is the same as [`PredicateSupport`], but without the wrapped expression.
#[derive(Debug, Clone, Copy)]
pub enum PredicateSupportDiscriminant {
    /// The predicate was successfully pushed down into the child node.
    Supported,
    /// The predicate could not be pushed down into the child node.
    Unsupported,
}

maybe something like

enum PushedDown {
  Yes,
  No
}

And the instead of PredicateSupport

#[derive(Debug, Clone)]
pub enum PredicateSupport {
    /// The predicate was successfully pushed down into the child node.
    Supported(Arc<dyn PhysicalExpr>),
    /// The predicate could not be pushed down into the child node.
    Unsupported(Arc<dyn PhysicalExpr>),
}

It would be

#[derive(Debug, Clone)]
pub enum PushedDownPredicate {
   push_down: PushedDown,
   predicate: Arc<dyn PhysicalExpr>
}

alamb · 2025-07-10T11:51:06Z

datafusion/physical-plan/src/filter.rs

-        let unsupported_self_filters = child_pushdown_result.self_filters[0]
+        let unsupported_parent_filters =
+            child_pushdown_result.parent_filters.iter().filter_map(|f| {
+                matches!(f.all(), PredicateSupportDiscriminant::Unsupported)


As a minor readability thing, I would find something like this easier to read:

Suggested change

matches!(f.all(), PredicateSupportDiscriminant::Unsupported)

f.all().is_unsupported()

Something similar to is_some() and is_none() for Option

Maybe I'll try and make a PR to see what that looks like

Unless there's some magic I'm not thinking about wouldn't that require returning something from all() that can have a is_unsupported() method? I'm weary of adding more tiny structs.

We could do it via an extension trait but not sure that reduces the cognitive burden

alamb · 2025-07-10T12:06:26Z

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

-        Ok(FilterPushdownPropagation::transparent(
-            child_pushdown_result,
-        ))
+        Ok(FilterPushdownPropagation::all(child_pushdown_result))


Maybe if_all?

alamb · 2025-07-10T12:11:54Z

datafusion/physical-plan/src/filter_pushdown.rs

    }
+
+    /// Convert the [`PredicateSupport`] into a [`PredicateSupportDiscriminant`].
+    pub fn discriminant(&self) -> PredicateSupportDiscriminant {


Maybe PredicateSupport would be easier to manage if it was a struct with a discrimnant field rather than an enum

adriangb · 2025-07-10T13:38:38Z

Thank you for the review @alamb

I incorporated your suggestion for restructuring the enum / struct into a struct with a discriminant field. It is much better.

I also renamed all to if_all and did some more docs tweaking.

adriangb · 2025-07-10T16:50:33Z

Note that I added a HashJoinExec implementation to motivate this PR but remove it in 5940cca because it lacks nuance necessary for a final complete correct implementation (it doesn't take into account the join type, etc.).

I spoke with @alamb and we agreed that we'd start with only inner joins (which are always safe) and if we ever want to support anything fancier we'll deal with the complexity then

alamb · 2025-07-11T20:30:48Z

Thanks again @adriangb

github-actions bot added optimizer Optimizer rules core Core DataFusion crate datasource Changes to the datasource crate physical-plan Changes to the physical-plan crate labels Jul 10, 2025

adriangb added 8 commits July 9, 2025 20:05

refactor filter pushdown to remove assumptions blocking hashjoinexec …

37c90e7

…pushdown

fix assertion

c6bfa01

fix

1c4b399

fix any/all

3166ed0

remove hashjoinexec impl

1f1da87

lint

e279b85

resolve merge

c9a330c

fix imports

3ae2e4e

adriangb force-pushed the hash-join-allow-filters branch from 167e740 to 3ae2e4e Compare July 10, 2025 01:12

fix merge

7e7d366

adriangb commented Jul 10, 2025

View reviewed changes

adriangb requested review from alamb, berkaysynnada and kosiew July 10, 2025 01:19

fix typo

20c74b9

lint

542d3f0

xudong963 self-requested a review July 10, 2025 10:25

alamb mentioned this pull request Jul 10, 2025

Improve some documentation in filter pushdown pydantic/datafusion#32

Merged

alamb reviewed Jul 10, 2025

View reviewed changes

alamb approved these changes Jul 10, 2025

View reviewed changes

alamb reviewed Jul 10, 2025

View reviewed changes

alamb and others added 3 commits July 10, 2025 07:39

Improve some documentation in filter pushdown (#32)

46f0469

refactor to a struct with a field instead of enum

fb6b8a2

better docs, rename methods

39d1d0c

fix cargo doc

caa3cb6

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

alamb merged commit 4bc66c8 into apache:main Jul 11, 2025
27 checks passed

alamb added the performance Make DataFusion faster label Jul 18, 2025

	matches!(f.all(), PredicateSupportDiscriminant::Unsupported)
	f.all().is_unsupported()

Refactor filter pushdown APIs to enable joins to pass through filters #16732

Refactor filter pushdown APIs to enable joins to pass through filters #16732

Uh oh!

Conversation

adriangb commented Jul 10, 2025

Uh oh!

adriangb commented Jul 10, 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!

kosiew commented Jul 10, 2025

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Documentation:

Consolidate some structures?

Naming

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!

adriangb commented Jul 10, 2025

Uh oh!

adriangb commented Jul 10, 2025

Uh oh!

Uh oh!

alamb commented Jul 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants