Add `EXISTS` and `IN` subquery rewriting for correlated filters at filter depth 1 #2451

jon-chuang · 2022-05-05T08:42:08Z

Which issue does this PR close?

Closes: #2351
Related: #2421

Rationale for this change

We perform a simple pattern match in the LogicalPlan optimizer on InSubquery(Projection(Filter(..))) and Exists(Filter(..)), extracting correlated/dependent columns to be used in the semi/anti join

TODO:

Sanity abort if detect_correlated_columns in the remainder expression (needs to be correlation free to derive the benefits of semi/antijoin)? Future: rewrite infallibly as SEMIJOIN + UNION ALL.
Test no optimization on non-conjunctive clauses containing correlated predicates?

What changes are included in this PR?

Are there any user-facing changes?

Future Work

Rewrite nested disjunctive expressions in the filter as UNION ALL.
For instance

SELECT * FROM t 
WHERE 
  EXISTS (..sq1..) OR 
  (a IN (..sq2..) AND a < 100) OR 
  a > 100

=>

SELECT * FROM t LEFT SEMIJOIN sq1 on ..
UNION ALL
SELECT * FROM t LEFT SEMIJOIN sq2 on a == r.a WHERE a < 100
UNION ALL
SELECT * FROM t WHERE a > 100

jon-chuang · 2022-05-05T08:53:41Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

-        \n        TableScan: sq projection=None [a:UInt32, b:UInt32, c:UInt32]\
-        \n        Projection: #sq_nested.c [c:UInt32]\
-        \n          TableScan: sq_nested projection=None [a:UInt32, b:UInt32, c:UInt32]";
+        \n    Semi Join: #sq.a = #sq_nested.c [a:UInt32, b:UInt32, c:UInt32]\


Note: this is a quirk of the implementation. It removes projections below the semi join in some cases (which are inconsequential).

One way to remove the quirk is to allow for multi-column InSubquerys.

The predicate pullup rule (next PR) will be able to "commute" a projection and a filter as follows:

Project([col(a)]) Filter(col(b)=(col(t.b)) => Filter(col(b)=(col(t.b)) Project([col(a), col(b)])

This way, one can move the Filter to the top of the subquery tree, just like with Exists.

The predicate pullup rule (next PR) will be able to "commute" a projection and a filter as follows:

This sounds like a subset of the filter "pushdown" rule -- so maybe we can just invoke that again rather than adding a new special case rule

jon-chuang · 2022-05-05T08:56:03Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+                    // NOTE: We only pattern match against Projection(Filter(..)). We will have another optimization rule
+                    // which tries to pull up all correlated predicates in an InSubquery into a Projection(Filter(..))
+                    // at the root node of the InSubquery's subquery. The Projection at the root must have as its expression
+                    // a single Column.


This may be somewhat restrictive. ~~It can be expanded but it needs some work. In particular joins will need to accept expressions rather than just joining blindly on rows.~~

Edit: Actually, its probably fine to have the following:

Projection(col(a)) Filter(col(b)=col(outer.b)) Projection(alias(col(a).plus(col(c)), a), col(b))

Yes I agree that re-using a Projection to calculate expressions is reasonable.

I think there are certain types of OUTER joins where the actual evaluation needs to happen within the join (as the presence of a NULL means something different than the row being filtered)

alamb · 2022-05-06T18:04:43Z

Thanks @jon-chuang -- I will try and review this carefully tomorrow

alamb · 2022-05-08T12:58:05Z

Sorry I haven't had a chance to review this PR yet @jon-chuang -- I will put it on my queue and hopefully get to it shortly

alamb · 2022-05-09T14:08:51Z

I have made it 1/2 way through this PR -- I will finish it up later today. Thanks again for your patience @jon-chuang

alamb

This is really nicely done @jon-chuang -- thank you 🙏

cc @andygrove

For other reviewers, here is the PR for equality IN/NOT IN support: #2421

alamb · 2022-05-09T14:06:02Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+            Semi Join: #table_a.a = #table_b.a [a:UInt32, b:UInt32]\
+            \n  Projection: #table_a.a, #table_a.b [a:UInt32, b:UInt32]\
+            \n    TableScan: table_a projection=None [a:UInt32, b:UInt32, c:UInt32]\
+            \n  Filter: #table_b.b > Utf8(\"5\") [a:UInt32, b:UInt32, c:UInt32]\


I vaguely remember some corner cases related to these non equality predicates and SEMI joins, but when I tried to come up with some examples of issues with this plan I could not. 👍

alamb · 2022-05-09T14:06:53Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+        let subquery = LogicalPlanBuilder::from(table_b)
+            .filter(
+                (col("table_a.c").eq(col("table_b.c"))).and(
+                    (col("table_a.a").eq(col("table_b.a")))


I suggest at least one predicate that has the subquery / table_b on the left -- like table_b.a = table_a. as all these predicates have table_a on the left

alamb · 2022-05-09T14:07:59Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+        Ok(())
+    }
+
+    // We only test not exists for the simplest case since all the other code paths


👍

I think there might be some subtleties involving NULLs (like if the subquery only has nulls and there is a non equality filter like table_b.a > 5

alamb · 2022-05-09T14:08:21Z

datafusion/core/src/optimizer/utils.rs

+// returns a new [LogicalPlan] that wraps `plan` in a [LogicalPlan::Filter] with
+/// its predicate with all `predicates` ANDed.
+pub fn filter_by_all(plan: LogicalPlan, predicates: &[&Expr]) -> LogicalPlan {
+    if let Some(predicate) = combine_conjunctive(predicates) {


alamb · 2022-05-09T15:28:21Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+                    // NOTE: We only pattern match against Projection(Filter(..)). We will have another optimization rule
+                    // which tries to pull up all correlated predicates in an InSubquery into a Projection(Filter(..))
+                    // at the root node of the InSubquery's subquery. The Projection at the root must have as its expression
+                    // a single Column.


Yes I agree that re-using a Projection to calculate expressions is reasonable.

I think there are certain types of OUTER joins where the actual evaluation needs to happen within the join (as the presence of a NULL means something different than the row being filtered)

alamb · 2022-05-09T15:29:33Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+                                        &mut correlated_join_columns,
+                                    );
+
+                                // Strip the projection away and use its input for the semi/anti-join


Can you explain why this stripping is needed? I don't understand it

alamb · 2022-05-09T16:04:14Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

-        \n        TableScan: sq projection=None [a:UInt32, b:UInt32, c:UInt32]\
-        \n        Projection: #sq_nested.c [c:UInt32]\
-        \n          TableScan: sq_nested projection=None [a:UInt32, b:UInt32, c:UInt32]";
+        \n    Semi Join: #sq.a = #sq_nested.c [a:UInt32, b:UInt32, c:UInt32]\


The predicate pullup rule (next PR) will be able to "commute" a projection and a filter as follows:

This sounds like a subset of the filter "pushdown" rule -- so maybe we can just invoke that again rather than adding a new special case rule

alamb · 2022-05-09T16:04:33Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+                let new_input = subquery_filters.iter().try_fold(
+                    optimized_input,
+                    |outer_plan, &subquery_expr| {
+                        self.rewrite_correlated_subquery_as_join(


alamb · 2022-05-09T16:07:50Z

datafusion/core/src/optimizer/utils.rs

+    let mut contains_correlated_columns = false;
+    expr.accept(CorrelatedColumnsVisitor {
+        outer_schema,
+        contains_correlated_columns: &mut contains_correlated_columns,
+    })?;
+    Ok(contains_correlated_columns)


I don't think it really matters, but this type of pattern could also be expressed like this, to avoid some mut (assuming contains_correlated_columns has been changed to bool):

Suggested change

let mut contains_correlated_columns = false;

expr.accept(CorrelatedColumnsVisitor {

outer_schema,

contains_correlated_columns: &mut contains_correlated_columns,

})?;

Ok(contains_correlated_columns)

let visitor = expr.accept(CorrelatedColumnsVisitor {

outer_schema,

contains_correlated_columns: false,

})?;

Ok(visitor.contains_correlated_columns)

jackwener · 2022-05-26T07:04:37Z

It's a great job!😀❤My work is a little busy.
I will review it on weekends.

alamb · 2022-05-26T10:13:21Z

@jon-chuang do you have time to work on this PR (looks like it has accumulated some conflicts) and there are some suggestions about tests. If not, I can find some time to do a final polish and merge it in

jackwener · 2022-05-28T14:32:36Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

+            .filter(
+                (col("table_a.c").eq(col("table_b.c"))).and(
+                    (col("table_a.a").eq(col("table_b.a")))
+                        .and(col("table_a.b").eq(col("table_b.b"))),


I think we need test like A or (B and c)?

Because it can't be split .

jackwener · 2022-05-28T14:37:23Z

datafusion/core/src/optimizer/subquery_filter_to_join.rs

@@ -64,105 +233,41 @@ impl OptimizerRule for SubqueryFilterToJoin {
                utils::split_conjunction(predicate, &mut filters);


A or (B anc C) can't be splited.

I'm not sure that it will have a affact.

alamb · 2022-06-07T17:22:58Z

Marking as draft as it needs some work before merging

alamb · 2022-07-21T16:00:16Z

@avantgardnerio noted that #2885 has some non trivial overlap with this PR, if you ever get back to it @jon-chuang

alamb · 2022-08-06T14:37:07Z

I believe this PR has been superseded now, so closing. Please feel free to reopen if you plan to keep working on it.

add exists subquery rewriting and correlated filters at filter depth 1

6b44853

github-actions bot added the datafusion Changes in the datafusion crate label May 5, 2022

jon-chuang changed the title ~~Add exists subquery rewriting and correlated filters at filter depth 1~~ Add EXISTS and IN subquery rewriting for correlated filters at filter depth 1 May 5, 2022

jon-chuang commented May 5, 2022

View reviewed changes

jon-chuang added 2 commits May 5, 2022 11:16

minor

99ae035

clippy

066a865

jon-chuang force-pushed the exists-simple-pattern-match branch from 230e72f to 066a865 Compare May 5, 2022 15:28

jon-chuang mentioned this pull request May 5, 2022

FilterPushdown should be aware of and ignore correlated predicates #2459

Open

jon-chuang added 7 commits May 6, 2022 16:48

minor

3476819

move to utils

781e76a

minor

2c20e5d

more rewriting

c968082

rewrite correlated columns util

1e64c5e

clippy

c1d71b5

move more to utils

e7f0365

jon-chuang force-pushed the exists-simple-pattern-match branch from ff90c1a to e7f0365 Compare May 7, 2022 04:02

alamb approved these changes May 9, 2022

View reviewed changes

alamb mentioned this pull request May 11, 2022

TPC-H Query 21 #172

Closed

jackwener reviewed May 28, 2022

View reviewed changes

andygrove removed the datafusion Changes in the datafusion crate label Jun 3, 2022

alamb marked this pull request as draft June 7, 2022 17:22

alamb closed this Aug 6, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add `EXISTS` and `IN` subquery rewriting for correlated filters at filter depth 1 #2451

Add `EXISTS` and `IN` subquery rewriting for correlated filters at filter depth 1 #2451

jon-chuang commented May 5, 2022 •

edited

Loading

jon-chuang May 5, 2022

jon-chuang May 5, 2022

alamb May 9, 2022

jon-chuang May 5, 2022 •

edited

Loading

alamb May 9, 2022

alamb commented May 6, 2022

alamb commented May 8, 2022

alamb commented May 9, 2022

alamb left a comment

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

alamb May 9, 2022

jackwener commented May 26, 2022

alamb commented May 26, 2022

jackwener May 28, 2022

jackwener May 28, 2022

alamb commented Jun 7, 2022

alamb commented Jul 21, 2022

alamb commented Aug 6, 2022

		@@ -64,105 +233,41 @@ impl OptimizerRule for SubqueryFilterToJoin {
		utils::split_conjunction(predicate, &mut filters);

Add EXISTS and IN subquery rewriting for correlated filters at filter depth 1 #2451

Add EXISTS and IN subquery rewriting for correlated filters at filter depth 1 #2451

Conversation

jon-chuang commented May 5, 2022 • edited Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

Future Work

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jon-chuang May 5, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented May 6, 2022

alamb commented May 8, 2022

alamb commented May 9, 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

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

Choose a reason for hiding this comment

jackwener commented May 26, 2022

alamb commented May 26, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Jun 7, 2022

alamb commented Jul 21, 2022

alamb commented Aug 6, 2022

Add `EXISTS` and `IN` subquery rewriting for correlated filters at filter depth 1 #2451

Add `EXISTS` and `IN` subquery rewriting for correlated filters at filter depth 1 #2451

jon-chuang commented May 5, 2022 •

edited

Loading

jon-chuang May 5, 2022 •

edited

Loading