Improve push down filter of join by JasonLi-cn · Pull Request #13184 · apache/datafusion

JasonLi-cn · 2024-10-30T12:21:31Z

Which issue does this PR close?

Closes #.

Rationale for this change

In the current version, the push down filter of Join is not very complete. For example:

> CREATE TABLE t1 (c0 CHAR, c1 INT);
0 row(s) fetched.
Elapsed 0.006 seconds.

> CREATE TABLE t2 (c0 CHAR, c1 INT);
0 row(s) fetched.
Elapsed 0.005 seconds.

> EXPLAIN SELECT * FROM t1 LEFT JOIN t2 ON t1.c0 = t2.c0 WHERE abs(t2.c1) > 5;
+---------------+---------------------------------------------------------------------------+
| plan_type     | plan                                                                      |
+---------------+---------------------------------------------------------------------------+
| logical_plan  | Filter: abs(t2.c1) > Int32(5)                                             |
|               |   Left Join: t1.c0 = t2.c0                                                |
|               |     TableScan: t1 projection=[c0, c1]                                     |
|               |     TableScan: t2 projection=[c0, c1]                                     |
| physical_plan | CoalesceBatchesExec: target_batch_size=8192                               |
|               |   FilterExec: abs(c1@3) > 5                                               |
|               |     RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1 |
|               |       CoalesceBatchesExec: target_batch_size=8192                         |
|               |         HashJoinExec: mode=Partitioned, join_type=Left, on=[(c0@0, c0@0)] |
|               |           MemoryExec: partitions=1, partition_sizes=[0]                   |
|               |           MemoryExec: partitions=1, partition_sizes=[0]                   |
|               |                                                                           |
+---------------+---------------------------------------------------------------------------+
2 row(s) fetched.
Elapsed 0.012 seconds.

Expect

> EXPLAIN SELECT * FROM t1 LEFT JOIN t2 ON t1.c0 = t2.c0 WHERE abs(t2.c1) > 5;
+---------------+-------------------------------------------------------------------------------+
| plan_type     | plan                                                                          |
+---------------+-------------------------------------------------------------------------------+
| logical_plan  | Inner Join: t1.c0 = t2.c0                                                      |
|               |   TableScan: t1 projection=[c0, c1]                                           |
|               |   Filter: abs(t2.c1) > Int32(5)                                               |
|               |     TableScan: t2 projection=[c0, c1]                                         |
| physical_plan | CoalesceBatchesExec: target_batch_size=8192                                   |
|               |   HashJoinExec: mode=Partitioned, join_type=Inner, on=[(c0@0, c0@0)]           |
|               |     CoalesceBatchesExec: target_batch_size=8192                               |
|               |       RepartitionExec: partitioning=Hash([c0@0], 16), input_partitions=1      |
|               |         MemoryExec: partitions=1, partition_sizes=[0]                         |
|               |     CoalesceBatchesExec: target_batch_size=8192                               |
|               |       RepartitionExec: partitioning=Hash([c0@0], 16), input_partitions=16     |
|               |         RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1 |
|               |           CoalesceBatchesExec: target_batch_size=8192                         |
|               |             FilterExec: abs(c1@1) > 5                                         |
|               |               MemoryExec: partitions=1, partition_sizes=[0]                   |
|               |                                                                               |
+---------------+-------------------------------------------------------------------------------+

BTW: If eliminate_outer_join can convert Left Join to Inner Join in this query, the filter abs(t2.c1) > Int32(5) can also be pushed down. So the next plan is to improve eliminate_outer_join rule. It's a little difficult.

What changes are included in this PR?

Modify the push_down_all_join functioninpush_down_filter` rule.

Are these changes tested?

yes

Are there any user-facing changes?

comphead · 2024-10-31T18:11:43Z

Thanks everyone

eejbyfeldt

I think this PR has correctness issues see: #13211

eejbyfeldt · 2024-11-01T13:33:18Z

datafusion/sqllogictest/test_files/push_down_filter_join.slt

+99 NULL NULL NULL
+
+query ITIT
+SELECT * FROM t1 LEFT JOIN t2 ON t1.t1_id = t2.t2_id AND t2.t2_name = 'z' ORDER BY t1_id, t1_name, t2_id, t2_name


I think the predicates here are "to simple" and a lot of the outer joins here will be converted to Inner joins by eliminate_outer_join rule and therefore these tests do not provide the desired coverage.

eejbyfeldt · 2024-11-01T13:55:44Z

datafusion/optimizer/src/push_down_filter.rs

+        let columns = predicate.column_refs();
+        macro_rules! restrict_null {
+            () => {{
+                let predicate_cloned = predicate.clone();
+                let cols = columns.iter().cloned();
+                is_restrict_null_predicate(predicate_cloned, cols).unwrap_or(false)
+            }};
+        }
+
+        if checker.left_only(&columns) && (left_preserved || restrict_null!()) {
            left_push.push(predicate);
-        } else if right_preserved && checker.is_right_only(&predicate) {
+        } else if checker.right_only(&columns) && (right_preserved || restrict_null!()) {


This seems incorrect. Why would a predicate being restrict null allow us to push it? It filtering it out null seems to be a argument against why it can be pushed down.

Yes, this is wrong. I neglected to modify JoinType here. Thank you @eejbyfeldt for your help in finding this problem.

This reverts commit 7ae1ccb.

alamb · 2024-11-02T10:23:51Z

For anyone following along, this PR appears to have had some correctness issues so @eejbyfeldt reverted it #13229

@JasonLi-cn are you willing to create a new PR that we can work through the correctness issues

Dandandan · 2024-11-02T10:53:56Z

I think for this pushdown to be correct, the join type can be changed.

I think this can be done in two phases:

if there is a filter for a left join on a right side (or left side for right join, etc.), the join can be transformed to inner join as the filter will remove all rows without match
the existing filter pushdown will push down the below the inner join

Dandandan · 2024-11-02T12:09:36Z

I think for this pushdown to be correct, the join type can be changed.

I think this can be done in two phases:

if there is a filter for a left join on a right side (or left side for right join, etc.), the join can be transformed to inner join as the filter will remove all rows without match

the existing filter pushdown will push down the below the inner join

Just adding to it, I think supporting this would be relatively simple. Currently eliminate_outer_join only supports columns that are non nullable, we can support expressions that filter out any nulls like most binary operators.

datafusion/datafusion/optimizer/src/eliminate_outer_join.rs

Line 175 in 89e96b4

fn extract_non_nullable_columns(

Dandandan · 2024-11-02T16:12:58Z

Hm actually, the transformation is already supported by EliminateOuterJoin, the only thing that's necessary is adding support for more expressions like abs like used in the example.

Dandandan · 2024-11-02T16:23:21Z

I tracked this here #13232

JasonLi-cn · 2024-11-04T05:05:19Z

For anyone following along, this PR appears to have had some correctness issues so @eejbyfeldt reverted it #13229

@JasonLi-cn are you willing to create a new PR that we can work through the correctness issues

I'm sorry that this PR has introduced bugs. I will continue to follow up this issue, and then please continue to help review. 🙏

improve push down filter of join

8e6a0b1

github-actions bot added optimizer Optimizer rules sqllogictest SQL Logic Tests (.slt) labels Oct 30, 2024

JasonLi-cn added 2 commits October 30, 2024 20:26

chores: add License of slt file

bcc13b4

fix cargo doc

d327267

Dandandan approved these changes Oct 30, 2024

View reviewed changes

comphead merged commit 7ae1ccb into apache:main Oct 31, 2024

eejbyfeldt mentioned this pull request Nov 1, 2024

PR: Improve push down filter of join #13184 introduces correctness issues #13211

Closed

eejbyfeldt reviewed Nov 1, 2024

View reviewed changes

eejbyfeldt added a commit to eejbyfeldt/datafusion that referenced this pull request Nov 2, 2024

Revert "Improve push down filter of join (apache#13184)"

554a6b1

This reverts commit 7ae1ccb.

eejbyfeldt mentioned this pull request Nov 2, 2024

Revert "Improve push down filter of join (#13184)" #13229

Merged

alamb pushed a commit that referenced this pull request Nov 2, 2024

Revert "Improve push down filter of join (#13184)" (#13229)

344f089

This reverts commit 7ae1ccb.

Dandandan mentioned this pull request Nov 2, 2024

Eliminate more outer joins by supporting more expressions #13232

Open

alamb mentioned this pull request Nov 5, 2024

Nov 5. 2024: This week in DataFusion #13265

Closed

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Improve push down filter of join#13184

Improve push down filter of join#13184
comphead merged 3 commits intoapache:mainfrom
JasonLi-cn:feature/optimize_push_down_join_predicate

JasonLi-cn commented Oct 30, 2024 •

edited

Loading

Uh oh!

comphead commented Oct 31, 2024

Uh oh!

eejbyfeldt left a comment •

edited

Loading

Uh oh!

eejbyfeldt Nov 1, 2024

Uh oh!

eejbyfeldt Nov 1, 2024

Uh oh!

JasonLi-cn Nov 4, 2024

Uh oh!

alamb commented Nov 2, 2024

Uh oh!

Dandandan commented Nov 2, 2024 •

edited

Loading

Uh oh!

Dandandan commented Nov 2, 2024

Uh oh!

Dandandan commented Nov 2, 2024

Uh oh!

Dandandan commented Nov 2, 2024

Uh oh!

JasonLi-cn commented Nov 4, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Conversation

JasonLi-cn commented Oct 30, 2024 • 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!

comphead commented Oct 31, 2024

Uh oh!

eejbyfeldt left a comment • 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!

eejbyfeldt Nov 1, 2024

Choose a reason for hiding this comment

Uh oh!

eejbyfeldt Nov 1, 2024

Choose a reason for hiding this comment

Uh oh!

JasonLi-cn Nov 4, 2024

Choose a reason for hiding this comment

Uh oh!

alamb commented Nov 2, 2024

Uh oh!

Dandandan commented Nov 2, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Dandandan commented Nov 2, 2024

Uh oh!

Dandandan commented Nov 2, 2024

Uh oh!

Dandandan commented Nov 2, 2024

Uh oh!

JasonLi-cn commented Nov 4, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

JasonLi-cn commented Oct 30, 2024 •

edited

Loading

eejbyfeldt left a comment •

edited

Loading

Dandandan commented Nov 2, 2024 •

edited

Loading