pushdown support for predicates in `ON` clause of joins #2647

korowa · 2022-05-29T13:42:59Z

Which issue does this PR close?

Closes #2619.

Rationale for this change

Pushdown support for predicates in JOIN ON clause and simplification of join planning.

What changes are included in this PR?

Join planning now just puts join filter into plan node without splitting it into pushable to inputs predicates.

flter_push_down rule checks for pushable predicates in join filter expression and replans them as filters for corresponding inputs.

Are there any user-facing changes?

No

Does this PR break compatibility with Ballista?

No

jackwener · 2022-05-30T02:06:52Z

datafusion/sql/src/planner.rs

        \n    TableScan: person projection=None\
-        \n    Filter: #orders.order_id > Int64(1)\
-        \n      TableScan: orders projection=None";
+        \n    TableScan: orders projection=None";


it seems that it can't pushdown filter to TableScan👀

You're right, it can not anymore - now all predicate movements are on filter_push_down optimizer, and there is no need for planner to define should it push some filters down or not - it just creates plan node as it is.

So to be clear, when run normally the pushdown will still happen, but they are no longer pushed down by the planner, rather they are pushed down at a later stage.

👍

alamb · 2022-05-31T12:17:15Z

I plan to review this later today or tomorrow if no one else has a chance to do so before

alamb

I looked at this PR carefully -- really nice work @korowa

I have some suggestions that might make the code easier to follow / less brittle to other changes in the future, but i think what you have here is well tested and looks correct to me.

I think the assumptions that the filter_expr is the last expression that comes back from Expr::expressions() is fine but could potentially be confusing to future readers (I imagine myself in some time looking at this code). Maybe some additional documentation could help here

In any event, thank you so much 🏅

alamb · 2022-06-01T20:17:12Z

datafusion/core/src/optimizer/filter_push_down.rs

@@ -178,13 +177,35 @@ fn lr_is_preserved(plan: &LogicalPlan) -> (bool, bool) {
    }
 }

+// For a given JOIN logical plan, determine whether each side of the join is preserved
+// in terms on join filtering.
+// Predicates from join filter can only be pushed to preserved join side.


alamb · 2022-06-01T20:22:11Z

datafusion/core/src/optimizer/filter_push_down.rs

 // Determine which predicates in state can be pushed down to a given side of a join.
 // To determine this, we need to know the schema of the relevant join side and whether
 // or not the side's rows are preserved when joining. If the side is not preserved, we
 // do not push down anything. Otherwise we can push down predicates where all of the
 // relevant columns are contained on the relevant join side's schema.
 fn get_pushable_join_predicates<'a>(
-    state: &'a State,
+    filters: &'a [(Expr, HashSet<Column>)],


I think it would help to document what the filters here represents, either in a doc string ("pairs of Exprs and a set of columns that represent ...") or maybe as a struct?

struct PushFilters { /// expr represents an On clause predicate expr: Expr, /// columns represents columns that appear in `expr` cols: HashSet<Column> }

I've added type Predicate (we already have type Predicates for similar kind of thing, so uniform naming might be better) and replaced all tuples of expression and its columns in signatures/attributes with it - hope it looks better.

alamb · 2022-06-01T20:22:36Z

datafusion/core/src/optimizer/filter_push_down.rs

-    let to_left = get_pushable_join_predicates(&state, left.schema(), left_preserved);
-    let to_right = get_pushable_join_predicates(&state, right.schema(), right_preserved);
-
+    let to_left =


I think this code is very nicely written and easy to follow 👍

alamb · 2022-06-01T20:26:40Z

datafusion/core/src/optimizer/filter_push_down.rs

    let right = optimize(right, right_state)?;

    // create a new Join with the new `left` and `right`
    let expr = plan.expressions();
+    let expr = if !on_filter.is_empty() && on_to_keep.is_empty() {


I don't really follow how we know the last element here in expr are the on expression -- doesn't that implicitly depend on the order of expressions returned from Expr::expressions()?

I wonder if we can make it more explicit somehow

I guess it's the downside of approach when we represent complex (or complex enough) structure (or some parts of it) with simple list/dict/tuple/etc, however after looking at how LogicalPlan::Repartition and LogicalPlan::Aggregate translated into expressions I supposed that we can rely on expressions vector element order in this case.

Regarding your comment above -- I've added explanation of what's going on with last element and what's expected expressions() behaviour -- hope it can be helpful for next person who access this part of code.

Thank you -- I think the comment helps a lot.

alamb · 2022-06-01T20:28:38Z

datafusion/core/src/optimizer/filter_push_down.rs

@@ -1351,7 +1427,7 @@ mod tests {
        let right = LogicalPlanBuilder::from(right_table_scan)
            .project(vec![col("a"), col("b"), col("c")])?
            .build()?;
-        let filter = col("test.a")
+        let filter = col("test.c")


Why is this changed? Because test.a is a join predicate as well?

Exactly, I've decided not to mix these cases and moved test for filtering on one of join keys to join_filter_on_common

alamb · 2022-06-01T20:30:14Z

datafusion/core/src/optimizer/filter_push_down.rs

+        );
+
+        let expected = "\
+        Inner Join: #test.a = #test2.a\


I recommend using different columns here -- like test.a = test2.b so that you can validate that correct column was pushed to each side

alamb · 2022-06-01T20:30:41Z

datafusion/expr/src/logical_plan/plan.rs

                .iter()
                .flat_map(|(l, r)| vec![Expr::Column(l.clone()), Expr::Column(r.clone())])
+                .chain(


This is a good catch

datafusion/expr/src/utils.rs

alamb · 2022-06-01T20:34:49Z

datafusion/sql/src/planner.rs

                        .map(|filter| join.filter(filter))
                        .unwrap_or(Ok(join))?
                        .build()
-                } else if join_type == JoinType::Inner && !normalized_filters.is_empty() {


alamb · 2022-06-01T20:35:41Z

datafusion/sql/src/planner.rs

        \n    TableScan: person projection=None\
-        \n    Filter: #orders.order_id > Int64(1)\
-        \n      TableScan: orders projection=None";
+        \n    TableScan: orders projection=None";


So to be clear, when run normally the pushdown will still happen, but they are no longer pushed down by the planner, rather they are pushed down at a later stage.

👍

…_pushdown

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

…a/arrow-datafusion into join_on_predicate_pushdown

alamb · 2022-06-02T10:46:42Z

I retriggered failing CI tests -- they seemed to be unrelated

alamb

Looks great -- thank you again @korowa

alamb · 2022-06-02T10:47:33Z

datafusion/core/src/optimizer/filter_push_down.rs

    let right = optimize(right, right_state)?;

    // create a new Join with the new `left` and `right`
    let expr = plan.expressions();
+    let expr = if !on_filter.is_empty() && on_to_keep.is_empty() {


Thank you -- I think the comment helps a lot.

alamb · 2022-06-02T18:05:18Z

Thanks again @korowa

github-actions bot added core Core DataFusion crate datafusion Changes in the datafusion crate logical-expr Logical plan and expressions optimizer Optimizer rules sql SQL Planner labels May 29, 2022

korowa force-pushed the join_on_predicate_pushdown branch from 7af4483 to 7e2432f Compare May 29, 2022 16:06

pushdown support for JOIN ON predicates

cbe65ed

korowa force-pushed the join_on_predicate_pushdown branch from 7e2432f to cbe65ed Compare May 29, 2022 18:34

jackwener reviewed May 30, 2022

View reviewed changes

alamb approved these changes Jun 1, 2022

View reviewed changes

korowa and others added 4 commits June 2, 2022 07:36

Merge remote-tracking branch 'upstream/master' into join_on_predicate…

2246d86

…_pushdown

Update datafusion/expr/src/utils.rs

cec9d5d

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Merge branch 'join_on_predicate_pushdown' of https://github.com/korow…

813b301

…a/arrow-datafusion into join_on_predicate_pushdown

review comments

1221009

alamb approved these changes Jun 2, 2022

View reviewed changes

alamb merged commit f547262 into apache:master Jun 2, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

pushdown support for predicates in `ON` clause of joins #2647

pushdown support for predicates in `ON` clause of joins #2647

korowa commented May 29, 2022

jackwener May 30, 2022

korowa May 30, 2022

alamb Jun 1, 2022

alamb commented May 31, 2022

alamb left a comment

alamb Jun 1, 2022

alamb Jun 1, 2022

korowa Jun 2, 2022 •

edited

Loading

alamb Jun 1, 2022

alamb Jun 1, 2022

korowa Jun 2, 2022

alamb Jun 2, 2022

alamb Jun 1, 2022

korowa Jun 2, 2022 •

edited

Loading

alamb Jun 1, 2022

korowa Jun 2, 2022

alamb Jun 1, 2022

alamb Jun 1, 2022

alamb Jun 1, 2022

alamb commented Jun 2, 2022

alamb left a comment

alamb Jun 2, 2022

alamb commented Jun 2, 2022

pushdown support for predicates in ON clause of joins #2647

pushdown support for predicates in ON clause of joins #2647

Conversation

korowa commented May 29, 2022

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

Does this PR break compatibility with Ballista?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented May 31, 2022

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

korowa Jun 2, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

korowa Jun 2, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Jun 2, 2022

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Jun 2, 2022

pushdown support for predicates in `ON` clause of joins #2647

pushdown support for predicates in `ON` clause of joins #2647

korowa Jun 2, 2022 •

edited

Loading

korowa Jun 2, 2022 •

edited

Loading