[SPARK-20175][SQL] Exists should not be evaluated in Join operator #17491

viirya · 2017-03-31T04:12:07Z

What changes were proposed in this pull request?

Similar to ListQuery, Exists should not be evaluated in Join operator too.

How was this patch tested?

Jenkins tests.

Please review http://spark.apache.org/contributing.html before opening a pull request.

viirya · 2017-03-31T04:13:40Z

cc @wzhfy @hvanhovell @nsyca

viirya · 2017-03-31T05:47:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala

+      val (withSubquery, withoutSubquery) =
+        splitConjunctivePredicates(condition).partition(SubqueryExpression.hasInOrExistsSubquery)
+      val newWithSubquery = withSubquery.map(_.transform {
+        case e @ Exists(sub, conditions, exprId) if conditions.isEmpty && !containsAgg(sub) =>


Currently take a conservative way, if there is already Aggregate, it skips the conversion.

The reason for this conversion is, I assume a count aggregation is cheaper than an existence join.

@viirya what we should do is to mark the outer join as an "early out" in the run-time join processing. An aggregation is not cheap as it needs to read the entire table to give the (first) answer. An "early out" logic is just like what we currently have in the LeftSemi join where only the first match of the join value is returned (and the subsequent matches are discarded). A LeftSemi join is a special case of an "early out" inner join where the columns of the right table are not permitted in the output of the join.

IMO, a step forward is to implement the "early out" mechanism in all the run-time join operators, nested-loop, sorted-merge, and hash; and inner, left outer, right outer. Then LeftSemi and LeftAnti will just be special cases of one of those operators.

For a simple count aggregation, I think it is cheap because it should prune columns and no data from the table will be shuffled.

@viirya @nsyca I am still looking at the code. If we do go the route to turn this into a scalar subquery, would a "limit" help prevent a scan of the inner table (assuming its very large). So something like "select Count(1) from (original subquery + limit 1)" ?

A simple test will tell. If you try to do a count on a 1-billion row table versus SELECT 1 FROM <tbl> LIMIT 1. Which one is better?

Yeah, that is a good idea. I think it definitely will help with an additional Limit operator.

SparkQA · 2017-03-31T06:09:14Z

Test build #75419 has finished for PR 17491 at commit b012550.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-03-31T06:29:47Z

retest this please.

SparkQA · 2017-03-31T06:32:34Z

Test build #75423 has started for PR 17491 at commit b012550.

viirya · 2017-03-31T07:10:33Z

retest this please.

SparkQA · 2017-03-31T09:23:46Z

Test build #75424 has finished for PR 17491 at commit b012550.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

nsyca · 2017-03-31T15:08:50Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala

+      val (withSubquery, withoutSubquery) =
+        splitConjunctivePredicates(condition).partition(SubqueryExpression.hasInOrExistsSubquery)
+      val newWithSubquery = withSubquery.map(_.transform {
+        case e @ Exists(sub, conditions, exprId) if conditions.isEmpty && !containsAgg(sub) =>


@viirya what we should do is to mark the outer join as an "early out" in the run-time join processing. An aggregation is not cheap as it needs to read the entire table to give the (first) answer. An "early out" logic is just like what we currently have in the LeftSemi join where only the first match of the join value is returned (and the subsequent matches are discarded). A LeftSemi join is a special case of an "early out" inner join where the columns of the right table are not permitted in the output of the join.

IMO, a step forward is to implement the "early out" mechanism in all the run-time join operators, nested-loop, sorted-merge, and hash; and inner, left outer, right outer. Then LeftSemi and LeftAnti will just be special cases of one of those operators.

nsyca · 2017-03-31T15:17:32Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala

      // A ListQuery defines the query which we want to search in an IN subquery expression.
      // Currently the only way to evaluate an IN subquery is to convert it to a
      // LeftSemi/LeftAnti/ExistenceJoin by `RewritePredicateSubquery` rule.
      // It cannot be evaluated as part of a Join operator.
+      // An Exists shouldn't be push into a Join operator too.


@dilipbiswal Here is another case of a regression from #16954. Would you think we should just say the following?

case SubqueryExpression => false

I think ScalarSubquery without correlated references can be pushed. Doesn't it?

I am not sure. The name of this is def canEvaluateWithinJoin so I assume it asks whether an input Expression can be processed as part of a Join operator. Can a ScalarSubquery be processed inside a Join?

@nsyca @viirya I just verified and the exists test fails the same way in 2.1. So its not a regression.

@nsyca Looking at this further, there is a SubqueryExec operator that can execute a ScalarSubquery and InSubquery (PlanSubqueries). As part of my change, i had removed the case for PredicateSubquery as we removed PredicateSubquery all together. I just quickly tried the following and got the query to work. I haven't verified the semantics but just tried something quickly. Basically if we were to keep the Exists expression as it is and push it down as a join condition and execute it as a InSubquery (possibly with a additional limit clause) there seems to be an infrastructure for it already. Or perhaps we may want to introduce a ExistSubquery exec operator that can work more efficiently.

case subquery: expressions.Exists => val executedPlan = new QueryExecution(sparkSession, subquery.plan).executedPlan InSubquery(Literal.TrueLiteral, SubqueryExec(s"subquery${subquery.exprId.id}", executedPlan), subquery.exprId)

What do you think Natt ?

What this code does is around the idea of treating an uncorrelated subquery as a black box. The subquery is processed as a self-contained operation and a list of values is returned. After that, the code evaluates as if this is an IN list predicate like IN (). In your code above, is represented as a "true" literal. That means the returned values from the subquery must be in Boolean type too.

Putting a LIMIT does help to short-circuit the processing to the first row. I still think putting a LIMIT explicitly as an extra LogicalPlan operator may have some negative side effect in the way that it prevents other Optimizer rules to further optimize the query. I have not thought about a concrete example to back my belief though.

I feel this optimization could be done better in the run-time area, rather than trying to shoehorn it in the Optimizer phase. What we can do is 1) propagate the notion of "early out" deeper to the operator on the RHS of the outer join. If it's a scan, stop scanning on the first row. 2) one more step further: cache the result of the RHS without a rescan because the next row from the parent table will always get the same answer from rescanning the subquery.

I am not sure. The name of this is def canEvaluateWithinJoin so I assume it asks whether an input Expression can be processed as part of a Join operator. Can a ScalarSubquery be processed inside a Join?

I remember ScalarSubquery without correlated reference will be evaluated as individual query plan and get its result back as an expression. So it should be no difference in run time compared with other expressions.

A Limit looks good to me for now. I can't think a negative side effect prevents possible optimization for the subquery plan. Doesn't it just like a re-written query with a limit clause added?

I think this is a corner usage case. To address this in run-time like the introduction of "early out" into physical join operators works, but it may involve a lot of code changes.

one more step further: cache the result of the RHS without a rescan because the next row from the parent table will always get the same answer from rescanning the subquery.

I quickly scan physical SortMergeJoin operator. If the streamed row matches the scanned group of rows, it will reuse the scanned group. Sounds it does what you said, if I don't miss something.

I think current join operators are smart enough that they won't re-scan the subquery if the next row still matches the scaned group of rows.

SparkQA · 2017-04-01T05:09:25Z

Test build #75439 has finished for PR 17491 at commit 88016bf.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-04-10T15:01:24Z

cc @cloud-fan

…to-join

SparkQA · 2017-04-11T03:30:40Z

Test build #75683 has finished for PR 17491 at commit 329f067.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-04-11T03:40:18Z

retest this please.

SparkQA · 2017-04-11T06:12:22Z

Test build #75687 has finished for PR 17491 at commit 329f067.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-04-11T09:18:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala

+          val expr = Alias(GreaterThan(countExpr.toAttribute, Literal(0)), e.toString)()
+          ScalarSubquery(
+            Project(Seq(expr),
+              Aggregate(Nil, Seq(countExpr), LocalLimit(Literal(1), sub))),


How useful is this optimization? It only works when Exists has no condition, is that a common case?

I do agree with @nsyca that we should implement "early-out" mechanism, which is more general

I think it is a special case. Then I will remove this optimization and minimize this pr's change.

We can address the early-out in other work.

Btw, I am not very sure this early-out can benefit the general usage, except for this kind of special case.

viirya · 2017-04-11T10:08:11Z

@cloud-fan The optimization rule is removed now. This patch now is just making Exists subquery without correlated references work. Please take a look again. Thanks.

SparkQA · 2017-04-11T12:20:15Z

Test build #75703 has finished for PR 17491 at commit 24ae5ce.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-04-11T12:33:24Z

LGTM, merging to master!

nsyca · 2017-04-11T14:32:47Z

@cloud-fan wrote: "How useful is this optimization? It only works when Exists has no condition, is that a common case?"

One of the common cases of this usage is an application of ACL where the application asks the database whether the user has a proper authority to access a certain set of data or not.

Ex:

select ... from controlled_table where exists (select 1 from acl_table where user = CURRENT_USER and role = ...)

From a runtime perspective, an optimal access plan is placing the ACL_TABLE as an outer of a nested-loop join with a semantic to fetch only the first qualified row, once the row exists, continue to process the inner table, CONTROLLED_TABLE, or avoiding access the inner completely if no qualified row from the outer.

viirya · 2017-04-11T15:04:04Z

I think the current approach will have a LeftSemi join for this Exists subquery. Is it still far from the optimal access plan you said?

## What changes were proposed in this pull request? Similar to `ListQuery`, `Exists` should not be evaluated in `Join` operator too. ## How was this patch tested? Jenkins tests. Please review http://spark.apache.org/contributing.html before opening a pull request. Author: Liang-Chi Hsieh <[email protected]> Closes apache#17491 from viirya/dont-push-exists-to-join.

Exists should not be evaluated in Join operator too.

b012550

viirya changed the title ~~[SQL][WIP] Exists should not be evaluated in Join operator~~ [SQL][WIP] Exists should not be evaluated in Join operator and can be converted to ScalarSubquery if no correlated reference Mar 31, 2017

viirya commented Mar 31, 2017

View reviewed changes

viirya changed the title ~~[SQL][WIP] Exists should not be evaluated in Join operator and can be converted to ScalarSubquery if no correlated reference~~ [SPARK-20175][SQL] Exists should not be evaluated in Join operator and can be converted to ScalarSubquery if no correlated reference Mar 31, 2017

nsyca reviewed Mar 31, 2017

View reviewed changes

Add a local limit operator to limit data scan.

88016bf

Merge remote-tracking branch 'upstream/master' into dont-push-exists-…

329f067

…to-join

cloud-fan reviewed Apr 11, 2017

View reviewed changes

Revert optimization for Exists subquery without correlated references.

24ae5ce

viirya changed the title ~~[SPARK-20175][SQL] Exists should not be evaluated in Join operator and can be converted to ScalarSubquery if no correlated reference~~ [SPARK-20175][SQL] Exists should not be evaluated in Join operator Apr 11, 2017

asfgit closed this in cd91f96 Apr 11, 2017

viirya deleted the dont-push-exists-to-join branch December 27, 2023 18:20

[SPARK-20175][SQL] Exists should not be evaluated in Join operator #17491

[SPARK-20175][SQL] Exists should not be evaluated in Join operator #17491

Conversation

viirya commented Mar 31, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

viirya commented Mar 31, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya Mar 31, 2017 • 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

SparkQA commented Mar 31, 2017

viirya commented Mar 31, 2017

SparkQA commented Mar 31, 2017

viirya commented Mar 31, 2017

SparkQA commented Mar 31, 2017

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

nsyca Mar 31, 2017 • edited Loading

Choose a reason for hiding this comment

viirya Apr 1, 2017 • edited Loading

Choose a reason for hiding this comment

SparkQA commented Apr 1, 2017

viirya commented Apr 10, 2017

SparkQA commented Apr 11, 2017

viirya commented Apr 11, 2017

SparkQA commented Apr 11, 2017

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

viirya commented Apr 11, 2017

SparkQA commented Apr 11, 2017

cloud-fan commented Apr 11, 2017

nsyca commented Apr 11, 2017

viirya commented Apr 11, 2017 • edited Loading

viirya commented Mar 31, 2017 •

edited

Loading

viirya Mar 31, 2017 •

edited

Loading

nsyca Mar 31, 2017 •

edited

Loading

viirya Apr 1, 2017 •

edited

Loading

viirya commented Apr 11, 2017 •

edited

Loading