Support for non equality predicates in `ON` clause of `LEFT`, `RIGHT,` and `FULL` joins #2591

korowa · 2022-05-22T21:50:25Z

Which issue does this PR close?

Closes #2509 , closes #2496 .

Rationale for this change

Support non equality predicates / filters in JOIN ON SQL clause

What changes are included in this PR?

Logical plan

Join logical plan now requires Option<Expr> field - filter which should be applied to "equijoined" data. Join planning logic left almost untouched:

Inner still planned as Join -> Filter (it allows proper filter pushdown)
in case of Left / Right planner still pushes down predicates relates only to inner join input, and now it allows predicates based on outer input
Full allows predicates in ON clause

Physical plan

Now, after building left/right indices vectors as a result of equijoin part of ON clause, HashJoin applies filter expression (if any has been provided) to batch of rows with according indices and produce new vectors with indices of joined rows after filtering. Intermediate batch contains only required for filter expression columns.

HashJoinExec physical plan node requires new Option<JoinFilter> field - JoinFilter structure encapsulates all necessary data to create intermediate batch and apply filter:

physical expression - filter expression itself, built against intermediate batch schema, it thus requires following two fields to evaluate expression while execution
column indices - stores indices and join sides on columns included in intermediate batch
schema - intermediate batch schema

Are there any user-facing changes?

Plan builder and DF join methods now require optional expression as an argument.

Does this PR break compatibility with Ballista?

New fields added to both logical and physical plan join nodes.
Related PR - apache/datafusion-ballista#36

korowa · 2022-05-23T06:24:33Z

Though this PR is functionally fine, I suppose it not to be most efficient in terms of runtime and resource usage (rebuild stage in case of RIGHT/OUTER join is a bit confusing) but it seems to fits in with columnar style data processing.

Anyway, suggestions/comments/questions are welcome - it wold be great to validate if it's an appropriate (at least for now) join filter implementation or there are better ways of doing this.

andygrove · 2022-05-23T13:15:28Z

Thank you @korowa for being the first to test out the new process around building against Ballista with changes there as well. Please let me know if you have any feedback on the process or suggestions to make it easier.

yjshen · 2022-05-23T15:41:27Z

@korowa Great to see this happening! How do you think to support filter in SortMergeJoinExec as well?

Cc @richox, you might be interested in this as well.

Ted-Jiang

@korowa Great work! 👍 i have left some comments.

Ted-Jiang · 2022-05-23T15:46:27Z

datafusion/core/src/physical_plan/hash_join.rs

+            let mask = as_boolean_array(&filter_result);
+
+            let left_filtered = PrimitiveArray::<UInt64Type>::from(
+                compute::filter(&left_indices, mask)?.data().clone(),


I think here, we apply filter after build_batch_from_indices means : apply filter after join operator. Is there a chance filter before join.
if wrong plz correct me.

Yes, at physical level filter always applied after actual join operation.

And you're right - there are cases when it's fine to filter inputs before join step - but here it's responsibility of logical planner / optimizer - they both are able to check if predicate (or its part) could be pushed before join - closer to scan.

So, when it comes to physical join, logical plan supposed to contain only t1.field < t2.field-like predicates which could not be applied before join

I agree with @korowa that moving filters out of the On clause is better left to the planner and optimizer.

This is an important point though -- perhaps we could add a comment to the docstring of the HashJoin operator explaining that filter should ideally only contain expressions that can not be pushed to the inputs of the join, one way or the other

@korowa @alamb ❤️ Thanks your explanations! I think after merge this Pr, we should add a check in filter_push_down.rs to improve this situation.

datafusion/core/src/physical_plan/join_utils.rs

alamb · 2022-05-23T19:02:30Z

I have this on my queue to review carefully tomorrow

Co-authored-by: Yang Jiang <37145547+Ted-Jiang@users.noreply.github.com>

korowa · 2022-05-23T20:06:10Z

Thank you @korowa for being the first to test out the new process around building against Ballista with changes there as well. Please let me know if you have any feedback on the process or suggestions to make it easier.

The guide in PR template is pretty clear, thank you!

Talking about suggestions - the only thing that comes into my mind is Github Action triggered by "PR merged" event which could

reset dev/build-arrow-ballista.sh
search for related open PR in ballista, and merge it

sounds complicated but unfortunately this is the only way I can imagine to keep two repos synced between each other 😞

korowa · 2022-05-23T20:08:47Z

@korowa Great to see this happening! How do you think to support filter in SortMergeJoinExec as well?

Cc @richox, you might be interested in this as well.

I'll check it out - there definitely should be point where joined records from both sides are collected as batch, which could be used for filter evaluation.

alamb

Thank you very much @korowa -- this is very very cool. 😍

I think this PR needs a few more tests but otherwise looks very nice. At a minimum, I think we need:

basic sql_integration tests for RIGHT and FULL join with non equijoin ON predicates
tests for the logic that does single table predicate pushdown to join inputs

I had some code structure comments / suggestions, but I don't think any of them is required. I just had some ideas which may help to improve the code readability.

It would also be neat to add some more sophisticated tests that use multiple record batches, as join inputs, but perhaps that is beyond the scope of this PR

cc @Dandandan

datafusion/core/src/dataframe.rs

alamb · 2022-05-24T18:43:54Z

datafusion/core/src/optimizer/filter_push_down.rs

                &right,
                JoinType::Inner,
                (vec![Column::from_name("a")], vec![Column::from_name("a")]),
+                None,


I recommend tests:

A filter on an inner join and ensuring that the filter pushdown pushes it correctly

A filter on an outer (left and full) showing that it is not pushed down (or that it is only pushed down on the non-preserved relation)

(if the extra filter is not pushed, I think it would be fine to file a follow on ticket to add that functionality -- I bet some others in the community might want to pick it up)

pushdown logic for ON condition is not supported in this PR, I've added tests (currently ignored) for this, and added an issue #2619

datafusion/core/src/optimizer/projection_push_down.rs

alamb · 2022-05-24T18:49:50Z

datafusion/core/tests/sql/joins.rs

-
-    assert!(res.is_err());
-    assert_eq!(format!("{}", res.unwrap_err()), "This feature is not implemented: Unsupported expressions in Left JOIN: [#t1_id >= Utf8(\"44\")]");
+        "SELECT t1_id, t1_name, t2_name FROM t1 LEFT JOIN t2 ON t1_id = t2_id AND t1_id >= 44 ORDER BY t1_id";


alamb · 2022-05-24T18:55:08Z

datafusion/core/tests/sql/joins.rs

-    assert_eq!(format!("{}", res.unwrap_err()), "This feature is not implemented: Unsupported expressions in Left JOIN: [#t1_id >= Utf8(\"44\")]");
+        "SELECT t1_id, t1_name, t2_name FROM t1 LEFT JOIN t2 ON t1_id = t2_id AND t1_id >= 44 ORDER BY t1_id";
+    let expected = vec![
+        "+-------+---------+---------+",


This answer looks good to me

alamb · 2022-05-24T19:21:49Z

datafusion/core/src/physical_plan/planner.rs

                        })
                        .collect::<Result<join_utils::JoinOn>>()?;

+                    let join_filter = match filter {


The intermediate schema is very much tied to how the Join is implemented, right? In other words, if we changed the order that the hash Join internally stored its columns I wonder if some / all of this code would need to change (and be hard to find / know to change). Thus I wonder if it might be best put into the join node itself

I thought about it, but decided that it would be kind of mess to add a bunch of logical plan related imports and entities to hash_join.
And I guess, if join internal storage is going to change, that will lead us to changing ColumnIndices somehow, and planner still is going to be affected (but I may be wrong).

datafusion/core/src/physical_plan/hash_join.rs

alamb · 2022-05-24T19:26:25Z

datafusion/core/src/physical_plan/hash_join.rs

+            let mask = as_boolean_array(&filter_result);
+
+            let left_filtered = PrimitiveArray::<UInt64Type>::from(
+                compute::filter(&left_indices, mask)?.data().clone(),


I agree with @korowa that moving filters out of the On clause is better left to the planner and optimizer.

This is an important point though -- perhaps we could add a comment to the docstring of the HashJoin operator explaining that filter should ideally only contain expressions that can not be pushed to the inputs of the join, one way or the other

alamb · 2022-05-24T19:30:02Z

datafusion/core/src/physical_plan/hash_join.rs

+
+    match join_type {
+        JoinType::Inner | JoinType::Left => {
+            // For both INNER and LEFT joins, input arrays contains only indices for matched data.


Something bothers me about this code treating Left and Right joins differently -- I would expect they look like mirrors of each other and Full to be treated differently

That was my initial plan, but it turned out there are no nulls appended in case of left join while building matched indices. All null values are handled later, using visited left side indices

alamb · 2022-05-24T19:31:41Z

datafusion/core/src/physical_plan/hash_join.rs

+            // In case of RIGHT and FULL join, left_indices could contain null values - these rows,
+            // where no match has been found, should retain in result arrays (thus join condition is satified)
+            //
+            // So, filter should be applied only to matched rows, and in case right (outer, batch) index


This logic I think should also apply left joins.

As for FULL joins, I think it means if there is no match for either left or right inputs, it should a row should still be produced.

I think the best way to resolve this comment is to add some tests showing correct answers for RIGHT and FULL joins in the sql_integration suite

Integration tests for more join types added

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

into join_filter

datafusion/core/src/optimizer/filter_push_down.rs

alamb · 2022-05-25T19:37:30Z

datafusion/core/tests/sql/joins.rs

+    Ok(())
+}

+#[tokio::test]


these tests look good 👍

datafusion/core/src/optimizer/filter_push_down.rs

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

into join_filter

alamb

🚀 thanks again @korowa !

korowa · 2022-05-26T10:24:10Z

🚀 thanks again @korowa !

Thank you for review!

github-actions bot added datafusion development-process Related to development process of DataFusion labels May 22, 2022

korowa force-pushed the join_filter branch from 1317c65 to 6b4ca91 Compare May 22, 2022 22:23

optional join filters

3f88a87

korowa force-pushed the join_filter branch from 6b4ca91 to 3f88a87 Compare May 23, 2022 05:19

korowa mentioned this pull request May 23, 2022

Filter field for JoinNode and HashJoinExecNode apache/datafusion-ballista#36

Merged

Ted-Jiang reviewed May 23, 2022

View reviewed changes

korowa and others added 2 commits May 23, 2022 22:09

Update datafusion/core/src/physical_plan/join_utils.rs

5b5ee8e

Co-authored-by: Yang Jiang <37145547+Ted-Jiang@users.noreply.github.com>

Merge remote-tracking branch 'upstream/master' into join_filter

8bd00b6

fix limit_push_down rule

10b48d1

korowa force-pushed the join_filter branch from 55a7404 to 10b48d1 Compare May 24, 2022 08:02

alamb changed the title ~~Optional filter in JOIN ON clause~~ Add support for non equality predicates in ON clause of LEFT, RIGHT, and FULL joins May 24, 2022

alamb changed the title ~~Add support for non equality predicates in ON clause of LEFT, RIGHT, and FULL joins~~ Support for non equality predicates in ON clause of LEFT, RIGHT, and FULL joins May 24, 2022

alamb reviewed May 24, 2022

View reviewed changes

korowa and others added 3 commits May 25, 2022 22:01

additional tests & PR comments

1391916

Suggestions from code review

0c9a5ff

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

Merge branch 'join_filter' of https://github.com/korowa/arrow-datafusion

e3242db

into join_filter

alamb reviewed May 25, 2022

View reviewed changes

linter warnings fixed

875d8bb

korowa force-pushed the join_filter branch from b3ebfd1 to 875d8bb Compare May 25, 2022 19:43

korowa mentioned this pull request May 25, 2022

Move JOIN ON predicates push down logic from planner to optimizer #2619

Closed

alamb approved these changes May 25, 2022

View reviewed changes

datafusion/core/src/optimizer/filter_push_down.rs Show resolved Hide resolved

datafusion/core/src/optimizer/filter_push_down.rs Show resolved Hide resolved

datafusion/core/src/optimizer/filter_push_down.rs Show resolved Hide resolved

korowa and others added 2 commits May 25, 2022 23:39

Apply suggestions from code review

1b6b5e6

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

docstrings updated

837ced4

korowa added 2 commits May 25, 2022 23:39

Merge branch 'join_filter' of https://github.com/korowa/arrow-datafusion

59920d9

into join_filter

Merge remote-tracking branch 'upstream/master' into join_filter

1a3b288

xudong963 added the enhancement New feature or request label May 26, 2022

alamb approved these changes May 26, 2022

View reviewed changes

alamb merged commit b6fb0dd into apache:master May 26, 2022

korowa mentioned this pull request May 29, 2022

reset ballista branch to apache/master #2641

Merged

Support for non equality predicates in ON clause of LEFT, RIGHT, and FULL joins #2591

Support for non equality predicates in ON clause of LEFT, RIGHT, and FULL joins #2591

Uh oh!

Conversation

korowa commented May 22, 2022 • edited by alamb 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 there any user-facing changes?

Does this PR break compatibility with Ballista?

Uh oh!

korowa commented May 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

andygrove commented May 23, 2022

Uh oh!

yjshen commented May 23, 2022

Uh oh!

Ted-Jiang 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!

Uh oh!

alamb commented May 23, 2022

Uh oh!

korowa commented May 23, 2022

Uh oh!

korowa commented May 23, 2022

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

korowa May 25, 2022 • 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!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

korowa commented May 26, 2022

Uh oh!

Reviewers

Assignees

Support for non equality predicates in `ON` clause of `LEFT`, `RIGHT,` and `FULL` joins #2591

Support for non equality predicates in `ON` clause of `LEFT`, `RIGHT,` and `FULL` joins #2591

korowa commented May 22, 2022 •

edited by alamb

Loading

korowa commented May 23, 2022 •

edited

Loading

korowa May 25, 2022 •

edited

Loading