Filter pushdown into cross join by mustafasrepo · Pull Request #8626 · apache/datafusion

mustafasrepo · 2023-12-22T15:31:51Z

Which issue does this PR close?

Closes #.

Rationale for this change

Current filters are not pushed down to the cross join (equality predicates are pushed down). For this reason we generate following plan for the query

SELECT *
FROM annotated_data as l, annotated_data as r
WHERE l.a > r.a

FilterExec: filter=a@0 > a@1
--CrossJoinExec: 
----RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], has_header=true
----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], has_header=true

However, if we were to pushdown filter into CrossJoin (by converting it into inner join with filter condition ) we could have produced following plan

NestedLoopJoinExec: join_type=Inner, filter=a@0 > a@1
--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], has_header=true
--CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a, b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST], has_header=true

which would be more memory efficient especially in large tables. This PR adds this support

What changes are included in this PR?

Are these changes tested?

Yes

Are there any user-facing changes?

alamb · 2023-12-22T21:09:25Z

I changed the title of this PR as I think it is more correct to say the filter is pushed "into" the join as the filter doesn't go below it

alamb · 2023-12-22T21:10:05Z

@jackwener or @liukun4515 by any chance do you have time to review this PR?

ozankabak

I just reviewed this and it LGTM as well 🚀

viirya · 2023-12-23T18:26:10Z

@@ -47,81 +48,93 @@ impl EliminateCrossJoin {
 /// For above queries, the join predicate is available in filters and they are moved to
 /// join nodes appropriately
 /// This fix helps to improve the performance of TPCH Q19. issue#78


The current doc (of PushDownFilter) doesn't have such example, maybe update it with this change?

viirya · 2023-12-23T18:49:23Z

                    return Ok(None);
                }
+                None
+            }


This looks like a missing case of reordering joins to eliminate CrossJoin, do we have a test case can cover this?

Can you expand on this so we also address it in the follow-on PR (in case you see a bug or gap here)?

I tried to add a simple test case for this feature. However, I couldn't write a simple test to reproduce this. However, without this reordering support test fails. If I can come up with a simple test, I will add it.

Oh, I meant previously this rule doesn't cover the case of reordering joins to eliminate CrossJoin without a Filter on top of Join (because it did only match LogicalPlan::Filter(filter). So if adding this new matching case LogicalPlan::Join is to address it, it means we don't have test case covering it previously. Then we may need to add one.

viirya · 2023-12-23T19:09:05Z

+fn convert_cross_join_to_inner_join(cross_join: CrossJoin) -> Result<Join> {
+    let CrossJoin { left, right, .. } = cross_join;
+    let join_schema = build_join_schema(left.schema(), right.schema(), &JoinType::Inner)?;
+    // predicate is given


Is this comment put wrong place?

viirya · 2023-12-23T19:11:56Z

    }
 }

+/// Convert cross join to join by pushing down filter predicate to the join condition


Looks like this function only does "Convert cross join to join"?

"pushing down filter predicate" is done by push_down_all_join. Newbies might be confused by this.

Exactly, I have updated the comment to reflect this

viirya · 2023-12-23T19:20:57Z

+fn convert_to_cross_join_if_beneficial(plan: LogicalPlan) -> Result<LogicalPlan> {
+    if let LogicalPlan::Join(join) = &plan {
+        // Can be converted back to cross join
+        if join.on.is_empty() && join.filter.is_none() {
+            return LogicalPlanBuilder::from(join.left.as_ref().clone())
+                .cross_join(join.right.as_ref().clone())?
+                .build();
+        }
+    }
+    Ok(plan)
+}


Hmm, after push_down_all_join, if there are predicates that cannot be join condition, they will be into a Filter on top of join node. In the case, this function will skip to convert the under join to cross join?

To be more precise filter can be pushed down below the join completely. In this case we may end up with joins empty equality predicate and empty filter condition.
As an example

Filter(l.a>l.b AND r.a>r.b) --Join (on=[], filter=None) ----LeftTable(a, b) ----RightTable(a, b)

will be converted to the plan below after push_down_all_join

Join (on=[], filter=None) --Filter(l.a>l.b) ----LeftTable(a, b) --Filter(r.a>r.b) ----RightTable(a, b)

this util ensures that plan above is converted to the plan below

CrossJoin --Filter(l.a>l.b) ----LeftTable(a, b) --Filter(r.a>r.b) ----RightTable(a, b)

However, if the original plan were

Filter(l.a>r.b) --Join (on=[], filter=None) ----LeftTable(a, b) ----RightTable(a, b)

after push_down_all_join we will end up with following plan

Join (on=[], filter=Some(l.a>r.b)) --LeftTable(a, b) --RightTable(a, b)

in this case join top cannot be converted to the cross join.

I mean if after push_down_all_join there are predicates remaining in Filter, but Join has empty on and empty filter, because you match the plan by if let LogicalPlan::Join(join) = &plan, the Join won't be converted to a Cross Join even looks like it should be (based on the logic here).

As far as I know, after after push_down_all_join predicates remaining in Filter will be pushed into the join filter. Hence, we are sure that top operator is Join.

This is how push_down_all_join handles remaining predicates in Filter.

if keep_predicates.is_empty() { Ok(plan) } else { // wrap the join on the filter whose predicates must be kept match conjunction(keep_predicates) { Some(predicate) => Ok(LogicalPlan::Filter(Filter::try_new( // Filter on top on Join predicate, Arc::new(plan), // this is Join )?)), None => Ok(plan), } }

And after it, convert_to_cross_join_if_beneficial is called. Where does the Filter push down? Am I missing it?

let plan = push_down_all_join( predicates, vec![], &join_plan, left, right, vec![], true, )?; convert_to_cross_join_if_beneficial(plan)?

And, the reason that the predicate is kept in Filter in push_down_all_join is because it cannot be pushed into Join filter or pushed down through Join. How does it push into/push down later?

I think you are right. We will discuss and file a follow-on PR today to fix.

mustafasrepo · 2023-12-25T13:19:14Z

@viirya I have opened a PR to fix remaining points PR.

mustafasrepo added 2 commits December 22, 2023 15:44

Initial commit

394325b

Simplifications

739bfdf

github-actions Bot added optimizer Optimizer rules sqllogictest SQL Logic Tests (.slt) labels Dec 22, 2023

alamb mentioned this pull request Dec 22, 2023

DataFusion weekly project plan (Andrew Lamb) - Dec 18, 2023 #8577

Closed

7 tasks

alamb changed the title ~~Filter pushdown through cross join~~ Filter pushdown into cross join Dec 22, 2023

Review

b522470

Dandandan approved these changes Dec 23, 2023

View reviewed changes

Review Part 2

50a0f84

ozankabak approved these changes Dec 23, 2023

View reviewed changes

More idiomatic Rust

ed28ce2

viirya reviewed Dec 23, 2023

View reviewed changes

mustafasrepo merged commit 3698693 into apache:main Dec 25, 2023

mustafasrepo mentioned this pull request Dec 25, 2023

[MINOR]: Add new test for filter pushdown into cross join #8648

Merged

appletreeisyellow mentioned this pull request Jan 4, 2024

chore: temporary branch for IOx update (12-25-2023 to 12-31-2023) appletreeisyellow/datafusion#4

Closed

matthewgapp mentioned this pull request Jan 11, 2024

matt/feat/recursive ctes/config flag matthewgapp/arrow-datafusion#3

Closed

Uh oh!

Conversation

mustafasrepo commented Dec 22, 2023

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!

alamb commented Dec 22, 2023

Uh oh!

alamb commented Dec 22, 2023

Uh oh!

ozankabak left a comment

Choose a reason for hiding this comment

Uh oh!

viirya Dec 23, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Dec 23, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

mustafasrepo Dec 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

viirya Dec 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

mustafasrepo commented Dec 25, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

viirya Dec 23, 2023 •

edited

Loading

viirya Dec 23, 2023 •

edited

Loading

mustafasrepo Dec 25, 2023 •

edited

Loading

viirya Dec 25, 2023 •

edited

Loading