[SPARK-18874][SQL] First phase: Deferring the correlated predicate pull up to Optimizer phase

[dilipbiswal]

What changes were proposed in this pull request?

Currently Analyzer as part of ResolveSubquery, pulls up the correlated predicates to its
originating SubqueryExpression. The subquery plan is then transformed to remove the correlated
predicates after they are moved up to the outer plan. In this PR, the task of pulling up
correlated predicates is deferred to Optimizer. This is the initial work that will allow us to
support the form of correlated subqueries that we don't support today. The design document
from @nsyca can be found in the following link :
DesignDoc

The brief description of code changes (hopefully to aid with code review) can be be found in the
following link:
CodeChanges

How was this patch tested?

The test case PRs were submitted earlier using.
16337 16759 16841 16915 16798 16712 16710 16760 16802

[SparkQA]

Test build #72992 has finished for PR 16954 at commit 33844bb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class In(value: Expression, list: Seq[Expression]) extends Predicate
• abstract class SubqueryExpression(
• case class ListQuery(
• case class Exists(

[dilipbiswal]

cc @hvanhovell @gatorsmile @nsyca

[gatorsmile]

cc @hvanhovell I did the internal review. It is ready for you to review it. Thanks!

[dilipbiswal]

In this test, the query is not a valid query as its missing group by. It used to pass before as we implicitly added r.c into aggregate grouping as part of pulling up correlated predicates in analyzer.

[dilipbiswal]

This function is mostly copied from Analyzer. The only difference is since validation is done by analyzer , we only process Filter, Project and Aggregate here.

[dilipbiswal]

Plan example is available in Code Changes in section 3.

[hvanhovell]

Maybe we should create a special InSubQuery expression. This looks like a lot of work for what we are doing here.

[nsyca]

The following definition is moved from PredicateSubquery.hasPredicateSubquery with the removing of the PredicateSubquery in the first case pattern.

[nsyca]

The following definition is moved from PredicateSubquery.hasNullAwarePredicateWithinNot with a code change to reflect the new representation of IN subquery.

[nsyca]

The following definition is moved from a nested definition of the same name in ResovleSubquery.pullOuterCorrelatedPredicates in Analyzer.scala.

[nsyca]

The following three definitions of stripOuterReference(s) are taken from a segment of code embedded in ResolveSubquery.rewriteSubQuery (Analyzer.scala).

[nsyca]

The following definition is moved unchanged from a nested definition of the same name in ResolveSubquery.pullOutCorrelatedPredicates (Analyzer.scala)

[dilipbiswal]

Now the de-duplication of sub plan happens in optimizer when we actually pull up the correlated predicates. Thus the project case is simplified.

[dilipbiswal]

@hvanhovell Hi Herman, was wondering if you had some time to look into this PR ? Please let me know your thoughts.

[hvanhovell]

I took an initial pass (not quite there yet). Most of it is in order. I left a few comments.

I will take another pass this week.

[hvanhovell]

Nit: spacing is of here

[dilipbiswal]

@hvanhovell Thanks.. will change.

[hvanhovell]

Should we move this into CheckAnalysis?

[hvanhovell]

Also use foreachUp instead of transformUp for the tree traversal in this method.

[dilipbiswal]

@hvanhovell The code here validates the correlated references as well as collects them to rewrite the outer plan to record the outer references. You are suggesting to move the "check" portion of checkAndGetOuter... () to checkAnalysis ? I will change to use foreachUp. Thanks.

[hvanhovell]

Ok, then leave as it is. I thought it was only doing validation.

[hvanhovell]

This might be somewhat expensive. We could also use something that terminates early.

[hvanhovell]

So I am all for decent documentation, but do you think we can find a somewhat smaller example?

[dilipbiswal]

@hvanhovell I will work on a smaller example. I think we don't need the subquery expression in the left plan. That should cut down the plan size.

[nsyca]

@dilipbiswal You probably are already on the right track that one of the subquery expression is not required. I guess either T1 intersect T1 where exists (T2) or a self-join scenario like T1, T1 Tx where exists (T2 where <Q>.col = T2.col) with <Q> be T1 or Tx should also do.

[hvanhovell]

Nit: ArrayBuffer.empty[Seq[Expression]] should also work. Why is it useful to collect sequences of expressions instead of just expressions?

[dilipbiswal]

@hvanhovell Thank you. I will change to use ArrayBuffer[Expression] instead.

[hvanhovell]

This only works with aggregates that are already in the Aggregate operator, this seems like a regression. What does Hive do?

[dilipbiswal]

@hvanhovell At the time of executing this rule, the aggregate expressions from the subquery plan are already pushed down to the Aggregate operator through ResolveAggregateFunctions. Here we are just updating the outer references in the subquery plan.

[hvanhovell]

Ok, awesome.

[hvanhovell]

This seems weird. Is this also the current behavior?

[dilipbiswal]

@hvanhovell Here is where i saw the code that handles the promotion between date and timestamp types. code

Please let me know if i missed something here.

[hvanhovell]

Shouldn't we factor that code out then? Now we have the same logic in two places.

[dilipbiswal]

@hvanhovell Thanks!!. I had tried to do this before as well as this came up during the internal review. I have made another try. Please let me know what you think.

[hvanhovell]

Ok please do this in a follow-up

[hvanhovell]

O wait, you have changed it.

[hvanhovell]

IIUC this is all done to get a better error message right?

[dilipbiswal]

@hvanhovell Actually previously checkInputDataTypes() called from checkAnalysis never had to deal with In subquery expressions as it gets rewritten to PredicateSubquery. With the change in this PR we now see the original IN subquery expressions and thus we need to make sure the types are same between LHS and RHS of the IN subquery expression.

[hvanhovell]

I am pretty sure this is wrong. This will work on the following expression: NOT(a) AND b IN(SELECT q FROM t)

[dilipbiswal]

@hvanhovell Actually I copied this function from existing PredicateSubquery.hasNullAwarePredicateWithinNot. I also found it a little odd that we are doing a e.find .. but was afraid to change this.

[dilipbiswal]

@hvanhovell I actually had debugged this before and had completely forgotten about it. The case you mention actually works ok because of the way we invoke this functions. Please see the caller at - code

[dilipbiswal]

@hvanhovell I have refactored the code a bit and i hope, this function is bit clearer to follow. I have moved the code from its caller in checkAnalysis to this function. Please let me know what you think.

[hvanhovell]

Use foreachDown?

[dilipbiswal]

@hvanhovell Actually herman, in this place, we are actually doing some intermediary transformation (even though we are not returning the transformed expression). After striping the outer references from the AggregateExpression, we are returning the newExpr. Example: min(outer(b))
when we hit 'min', we strip out the outer references and return min(b) to the next transformation so that we don't encounter outer(b) in the next pass and consider it an outer expression. May be there is a better way to achieve this :-)

[dilipbiswal]

@hvanhovell Hello Herman, I have addressed all your comments except one where you are suggesting an optimization to failOnOuterReferenceInSubTree(). I am trying to see how i can improve that code. In the meantime, i wanted to push the updates i have so far. Please go through when you have some time. Thanks a lot.

[SparkQA]

Test build #73646 has finished for PR 16954 at commit 886a744.

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

[SparkQA]

Test build #73712 has finished for PR 16954 at commit 3b4bb90.

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

[gatorsmile]

Nit: s" -> "

and also move the space from the beginning of the second line to the end of the first line.

[gatorsmile]

left hand side -> left hand side (LHS)
right hand side -> right hand side (RHS)

[gatorsmile]

lhs -> LHS. Please correct all the similar cases in comments.

[gatorsmile]

Please add a comment to explain what this condition means.

[gatorsmile]

Please use case (name, value) => instead of pair

[gatorsmile]

case (p, In(e, Seq(ListQuery(sub, conditions, _))))

[gatorsmile]

case (p, Not(In(e, Seq(ListQuery(sub, conditions, _))))) =>

[gatorsmile]

case Exists(sub, conditions, _)

[gatorsmile]

newPlan = Join(newPlan, sub, ExistenceJoin(exists), conditions.reduceLeftOption(And))

[gatorsmile]

val newConditions = (inConditions ++ conditions).reduceLeftOption(And)
newPlan = Join(newPlan, sub, ExistenceJoin(exists), newConditions)

[gatorsmile]

case In(value, Seq(ListQuery(sub, conditions, _)))

[gatorsmile]

getValueExpression(e) -> getValueExpression(value)

[gatorsmile]

newplan -> newPlan
newcond -> newCond

val (newPlan, newCond) = if (outer.nonEmpty) {

[gatorsmile]

case ScalarSubquery(sub, children, exprId) if children.nonEmpty =>

[gatorsmile]

if you want, you also can use cond to replace children

[gatorsmile]

case Exists(sub, children, exprId) if children.nonEmpty =>

[gatorsmile]

case ListQuery(sub, _, exprId) =>

[hvanhovell]

Please check this in the guard of the rule. This currently throws a very hard to understand error.

[hvanhovell]

Use orElse?

[hvanhovell]

I think it is better if you cast on the castedLhs here:

val newLhs = castedLhs match {
  case Seq(lhs) => lhs
  case _ =>
    CreateNamedStruct(cns.nameExprs.zip(castedLhs).flatMap {
      case (name, value) => Seq(name, value)
    })
}

[dilipbiswal]

@hvanhovell Can you please double check the above code ? I am unable to compile the code as cns is not defined. In this block, i was looking at the original left hand side expression to see if its a named struct and if so i construct a named struct back. castedLhs would always match a Seq(lhs) , no ? Please let me know if i am missing something here.

[hvanhovell]

My bad, I never compile these snippets. You have a point there. We could just use CreateStruct (since we really don't care about the name). So that would look something like this (again not compiled):

val newLhs = castedLhs match {
  case Seq(lhs) => lhs
  case _ => CreateStruct(castedLhs)
}

[dilipbiswal]

@hvanhovell Thanks a lot. You are right, we don't care about the names. This looks much better.

[hvanhovell]

It is better to display the offending columns. Now the user has to find the offending columns by herself/himself.

[dilipbiswal]

@hvanhovell The new error message looks like following. Does this look okay to you ?

Error in query: cannot resolve '(named_struct('c1', at1.`c1`, 'c2', at1.`c2`) IN (listquery()))' due to data type mismatch: 
The data type of one or more elements in the left hand side of an IN subquery
is not compatible with the data type of the output of the subquery
Mismatched columns:
[(at1.`c1`:decimal(10,0), at2.`c1`:timestamp), (at1.`c2`:timestamp, at2.`c2`:decimal(10,0))]
Left side:
[decimal(10,0), timestamp].
Right side:
[timestamp, decimal(10,0)].

[hvanhovell]

Use catalog strings please.

[hvanhovell]

This still looks fishy to me. It is also in the original code base, so we don't have to fix this now. Can you open a JIRA to track this?

[dilipbiswal]

@hvanhovell Sure. will do

[hvanhovell]

Shouldn't we check if the aggregate contains inner references? You could use collect leaves here: a.collectLeaves.forall(_.isInstanceOf[OuterReference])

[dilipbiswal]

@hvanhovell Sounds good Herman. I will change the guard to look for "all outer references" now. I will do a small followup to raise an appropriate exception if there are mixture of outer and inner references within a AggregateExpression as i have to write new tests for it. Hope thats ok with you.

[hvanhovell]

Yeah just address this in a follow-up

[hvanhovell]

Just use flatMap here? Instead of foreach?

[hvanhovell]

getOuterReferences is kind of magical in the sense that it also isolates the aggregate expression. Could you add a single line of comment to highlight this?

[dilipbiswal]

@hvanhovell Will do.

[hvanhovell]

Why make create two objects SubqueryExpression and SubExprUtils? Are they that different?

[dilipbiswal]

@hvanhovell Actually @nsyca and i had discussed a bit about this. SubqueryExpression object has methods that operates strictly in the context of SubqueryExpression where as the utils has methods that mostly deals with OuterReferences.. So they can operate on the subquery plans referred to from SubqueryExpression or may be in the future if we support queries of the form.

select * from t1 left outer join (select * from t2 where t2.c1 = t1.c1) on …

If you think we should merge this two then let me know and i will do it :-).

[hvanhovell]

Ok, so the the SubExprUtils should be moved to the OuterReference companion :)...

Let's leave it for now.

[hvanhovell]

@dilipbiswal can you update. I left a few minor comments, but it is good to go anyway (pending the update).

[dilipbiswal]

@hvanhovell I am in the process of incorporating the comments. I will send an update soon.

[SparkQA]

Test build #74489 has started for PR 16954 at commit 19cdbb0.

[dilipbiswal]

retest this please

[SparkQA]

Test build #74495 has finished for PR 16954 at commit 19cdbb0.

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

[hvanhovell]

LGTM. Merging to master. Thanks for the hard work, and your patience!

[dilipbiswal]

@hvanhovell @gatorsmile @nsyca Thank you very much !!