[SPARK-20334][SQL] Return a better error message when correlated predicates contain aggregate expression that has mixture of outer and local references. #17636

dilipbiswal · 2017-04-14T05:28:27Z

What changes were proposed in this pull request?

Address a follow up in comment
Currently subqueries with correlated predicates containing aggregate expression having mixture of outer references and local references generate a codegen error like following :

SELECT t1a 
FROM   t1
GROUP  BY 1
HAVING EXISTS (SELECT 1 
               FROM  t2
               WHERE t2a < min(t1a + t2a));

Exception snippet.

Cannot evaluate expression: min((input[0, int, false] + input[4, int, false]))
	at org.apache.spark.sql.catalyst.expressions.Unevaluable$class.doGenCode(Expression.scala:226)
	at org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression.doGenCode(interfaces.scala:87)
	at org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$genCode$2.apply(Expression.scala:106)
	at org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$genCode$2.apply(Expression.scala:103)
	at scala.Option.getOrElse(Option.scala:121)
	at org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:103)

After this PR, a better error message is issued.

org.apache.spark.sql.AnalysisException
Error in query: Found an aggregate expression in a correlated 
predicate that has both outer and local references, which is not supported yet. 
Aggregate expression: min((t1.`t1a` + t2.`t2a`)), 
Outer references: t1.`t1a`, 
Local references: t2.`t2a`.;

How was this patch tested?

Added tests in SQLQueryTestSuite.

…s contain aggregate expression that has mixture of outer and local references

SparkQA · 2017-04-14T05:32:30Z

Test build #75792 has started for PR 17636 at commit a266c8d.

gatorsmile · 2017-04-14T05:35:33Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+                   |Aggregate expression: ${a.sql}
+                   |Outer references: ${outer.map(_.sql).mkString(", ")}
+                   |Local references: ${local.map(_.sql).mkString(", ")}
+             """.


@gatorsmile Thanks. fixed.

gatorsmile · 2017-04-14T05:37:53Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1305,6 +1329,8 @@ class Analyzer(
            case _: EqualTo | _: EqualNullSafe => false
            case _ => true
          }
+
+          correlated.foreach(checkMixedReferencesInsideAggregation(_))


Nit: correlated.foreach(checkMixedReferencesInsideAggregation)

@gatorsmile Thanks. fixed.

gatorsmile · 2017-04-14T05:43:30Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+                   |outer and local references, which is not supported yet.
+                   |Aggregate expression: ${a.sql}
+                   |Outer references: ${outer.map(_.sql).mkString(", ")}
+                   |Local references: ${local.map(_.sql).mkString(", ")}


Please add a comma at each line

|Aggregate expression: ${a.sql}, |Outer references: ${outer.map(_.sql).mkString(", ")}, |Local references: ${local.map(_.sql).mkString(", ")}.

gatorsmile · 2017-04-14T05:46:40Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+                   |Aggregate expression: ${a.sql}
+                   |Outer references: ${outer.map(_.sql).mkString(", ")}
+                   |Local references: ${local.map(_.sql).mkString(", ")}
+                 """.stripMargin.replace("\n", " ").trim()


If we do not add comma, we can remove .replace("\n", " ").trim()

@gatorsmile It will be too many lines sean ? I think adding comma is better ?

SparkQA · 2017-04-14T05:47:31Z

Test build #75794 has started for PR 17636 at commit bb1bdad.

gatorsmile · 2017-04-14T05:50:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1305,6 +1328,8 @@ class Analyzer(
            case _: EqualTo | _: EqualNullSafe => false
            case _ => true
          }
+
+          correlated.foreach(checkMixedReferencesInsideAggregation)


failOnOuterReference should be the single entry error checking for outer references.

Filter is special. We can still call failOnOuterReference(f), but do the extra checking for Filter in failOnOuterReference.

gatorsmile · 2017-04-14T05:50:48Z

Look good to me except the above comments.

viirya · 2017-04-14T07:40:47Z

The error message ...Aggregate expression: min((t2.t2a + t3.t3a))... shown in the description seems not from the example query.

viirya · 2017-04-14T07:51:42Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1219,7 +1241,8 @@ class Analyzer(

      // Make sure a plan's expressions do not contain outer references
      def failOnOuterReference(p: LogicalPlan): Unit = {
-        if (p.expressions.exists(containsOuter)) {
+        p.expressions.foreach(checkMixedReferencesInsideAggregation)


~~p is not definitely an Aggregate, so I think checkMixedReferencesInsideAggregation will not only apply to aggregate expression.~~

@viirya checkMixedReferencesInsideAggregation only looks for AggregateExpression ? For other types its just a pass-through.

ok. got it. :-) Btw, Filter can also have aggregate expression here.

viirya · 2017-04-14T07:54:19Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1219,7 +1241,8 @@ class Analyzer(

      // Make sure a plan's expressions do not contain outer references
      def failOnOuterReference(p: LogicalPlan): Unit = {
-        if (p.expressions.exists(containsOuter)) {
+        p.expressions.foreach(checkMixedReferencesInsideAggregation)
+        if (!p.isInstanceOf[Filter] && p.expressions.exists(containsOuter)) {


Is this new condition p.isInstanceOf[Filter] just a safe network? Actually we won't use failOnOuterReference on Filter.

@viirya In the new code, we are calling failOnOuterReference on Filter based on @gatorsmile's comment.

oh. I found it.

viirya · 2017-04-14T07:55:11Z

sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql

@@ -1,42 +1,72 @@
 -- The test file contains negative test cases
 -- of invalid queries where error messages are expected.

-create temporary view t1 as select * from values


Those just change for case, right?

@viirya Yeah.. since i was on this test case, thought i should fix the case.

viirya · 2017-04-14T08:22:00Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1219,7 +1241,8 @@ class Analyzer(

      // Make sure a plan's expressions do not contain outer references
      def failOnOuterReference(p: LogicalPlan): Unit = {
-        if (p.expressions.exists(containsOuter)) {
+        p.expressions.foreach(checkMixedReferencesInsideAggregation)
+        if (!p.isInstanceOf[Filter] && p.expressions.exists(containsOuter)) {


I think we should put the check of containsOuter before checkMixedReferencesInsideAggregation?

@viirya This is by choice. So we want to return the same error for TC.01.03 and TC.01.04. Currently we return a "..not supported outside of WHERE/HAVING" error for TC.01.04 which is misleading.

viirya · 2017-04-14T08:25:10Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1210,6 +1210,28 @@ class Analyzer(
    private def checkAndGetOuterReferences(sub: LogicalPlan): Seq[Expression] = {
      val outerReferences = ArrayBuffer.empty[Expression]

+      // Validate that correlated aggregate expression do not contain a mixture
+      // of outer and local references.
+      def checkMixedReferencesInsideAggregation(expr: Expression): Unit = {


nit: checkMixedReferencesInsideAggregation -> checkMixedReferencesInsideAggregationExpr

@viirya Will change it to checkMixedReferencesInsideAggregateExpr

viirya · 2017-04-14T08:27:55Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1219,7 +1241,8 @@ class Analyzer(

      // Make sure a plan's expressions do not contain outer references


We should change this comment accordingly too.

@viirya Ok. will do.

viirya · 2017-04-14T08:30:41Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1219,7 +1241,8 @@ class Analyzer(

      // Make sure a plan's expressions do not contain outer references
      def failOnOuterReference(p: LogicalPlan): Unit = {


failOnOuterReference -> failOnOuterOrMixedReference? Or is it fine to keep it unchanged?

@viirya ok, will change it to failOnInvalidOuterReference. That should include both the cases.

viirya · 2017-04-14T08:31:25Z

LGTM except for few minor comments.

SparkQA · 2017-04-14T09:36:21Z

Test build #75798 has finished for PR 17636 at commit ff88651.

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

SparkQA · 2017-04-14T12:02:47Z

Test build #75805 has finished for PR 17636 at commit c4e1a01.

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

dilipbiswal · 2017-04-17T16:53:50Z

cc @gatorsmile @hvanhovell @cloud-fan

gatorsmile · 2017-04-17T21:47:10Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

-      def failOnOuterReference(p: LogicalPlan): Unit = {
-        if (p.expressions.exists(containsOuter)) {
+      // Make sure a plan's expressions do not contain :
+      // 1. Aggregate expressions that has mixture of outer and local references.


Nit: has -> have

gatorsmile · 2017-04-17T22:10:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1362,7 +1389,7 @@ class Analyzer(
        // Note:
        // Generator with join=false is treated as Category 4.
        case g: Generate if g.join =>
-          failOnOuterReference(g)
+          failOnInvalidOuterReference(g)

        // Category 4: Any other operators not in the above 3 categories
        // cannot be on a correlation path, that is they are allowed only


Do we have any test case to cover this scenario?

@gatorsmile I couldn't find a test for this. I will add one for now in SubquerySuite. I will move the negative tests to the sqlquerytestsuite in a follow-up pr.

gatorsmile · 2017-04-17T22:11:59Z

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

@@ -367,6 +367,8 @@ case class OuterReference(e: NamedExpression)
  override def exprId: ExprId = e.exprId
  override def toAttribute: Attribute = e.toAttribute
  override def newInstance(): NamedExpression = OuterReference(e.newInstance())
+  override def sql: String = e.sql
+  override def toString: String = e.toString


Any reason to override this? If we keep it unchanged, will it be easier for us to find whether it is an outer reference in the plans?

@gatorsmile Thats a good point Sean. I will revert this and error raising code to make sure its printed properly.

gatorsmile · 2017-04-17T22:12:24Z

sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql

+FROM   t1
+WHERE  t1a in (SELECT   min(t2a)
+               FROM     t2
+               GROUP by t2c


Nit: by -> BY

gatorsmile · 2017-04-17T22:14:06Z

sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql

-                                having   t3b > t2b ))
+SELECT *
+FROM   t1
+WHERE  t1a in (SELECT   min(t2a)


Nit: in -> IN

SparkQA · 2017-04-18T00:11:45Z

Test build #75867 has finished for PR 17636 at commit af3d367.

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

SparkQA · 2017-04-18T02:26:36Z

Test build #75875 has finished for PR 17636 at commit 55c64ca.

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

dilipbiswal · 2017-04-18T02:50:40Z

retest this please

SparkQA · 2017-04-18T05:09:09Z

Test build #75880 has finished for PR 17636 at commit 55c64ca.

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

SparkQA · 2017-04-19T21:09:48Z

Test build #75952 has finished for PR 17636 at commit d986ddc.

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

dilipbiswal · 2017-04-19T22:10:47Z

@gatorsmile I have addressed the comments. Can you please look at this when you get a chance. Thanks !!

gatorsmile · 2017-04-19T23:45:27Z

sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala

@@ -815,7 +815,7 @@ class SubquerySuite extends QueryTest with SharedSQLContext {

  // Generate operator
  test("Correlated subqueries in LATERAL VIEW") {
-    withTempView("t1", "t2") {
+    withTempView("t1", "t2", "t3") {


uh... i took out the usage of t3 and forgot to take out from this line. Thanks for catching this.

gatorsmile · 2017-04-19T23:46:11Z

sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala

+            | from t2
+            | where exists (select *
+            |               from t1 lateral view explode(t2.arr_c2) q as c2
+            |               where t1.c1 = t2.c1)


Nit: capitalize the SQL keyword?

@gatorsmile The other test was in lower case . so i had it in lower case for consistency. I will change both the SQLs in this test. Thank you.

SparkQA · 2017-04-20T03:24:14Z

Test build #75960 has finished for PR 17636 at commit 2411f3e.

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

gatorsmile · 2017-04-20T04:52:00Z

LGTM

hvanhovell · 2017-04-20T20:35:03Z

Merging to master/branch-2.2. Thanks!

dilipbiswal · 2017-04-20T20:40:39Z

@hvanhovell Thanks a lot.
@gatorsmile Thank you for doing a very thorough review. Really appreciate it !!

…icates contain aggregate expression that has mixture of outer and local references. ## What changes were proposed in this pull request? Address a follow up in [comment](apache#16954 (comment)) Currently subqueries with correlated predicates containing aggregate expression having mixture of outer references and local references generate a codegen error like following : ```SQL SELECT t1a FROM t1 GROUP BY 1 HAVING EXISTS (SELECT 1 FROM t2 WHERE t2a < min(t1a + t2a)); ``` Exception snippet. ``` Cannot evaluate expression: min((input[0, int, false] + input[4, int, false])) at org.apache.spark.sql.catalyst.expressions.Unevaluable$class.doGenCode(Expression.scala:226) at org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression.doGenCode(interfaces.scala:87) at org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$genCode$2.apply(Expression.scala:106) at org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$genCode$2.apply(Expression.scala:103) at scala.Option.getOrElse(Option.scala:121) at org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:103) ``` After this PR, a better error message is issued. ``` org.apache.spark.sql.AnalysisException Error in query: Found an aggregate expression in a correlated predicate that has both outer and local references, which is not supported yet. Aggregate expression: min((t1.`t1a` + t2.`t2a`)), Outer references: t1.`t1a`, Local references: t2.`t2a`.; ``` ## How was this patch tested? Added tests in SQLQueryTestSuite. Author: Dilip Biswal <dbiswal@us.ibm.com> Closes apache#17636 from dilipbiswal/subquery_followup1.

[SPARK-20334] Return a better error message when correlated predicate…

a266c8d

…s contain aggregate expression that has mixture of outer and local references

gatorsmile reviewed Apr 14, 2017

View reviewed changes

Code review

bb1bdad

gatorsmile reviewed Apr 14, 2017

View reviewed changes

Code review

ff88651

viirya reviewed Apr 14, 2017

View reviewed changes

comments

c4e1a01

gatorsmile reviewed Apr 17, 2017

View reviewed changes

review comments

af3d367

gatorsmile reviewed Apr 17, 2017

View reviewed changes

review comments

55c64ca

Simplify test

d986ddc

gatorsmile reviewed Apr 19, 2017

View reviewed changes

review comments

2411f3e

asfgit closed this in d95e4d9 Apr 20, 2017

		@@ -1219,7 +1241,8 @@ class Analyzer(

		// Make sure a plan's expressions do not contain outer references

		@@ -1219,7 +1241,8 @@ class Analyzer(

		// Make sure a plan's expressions do not contain outer references
		def failOnOuterReference(p: LogicalPlan): Unit = {

[SPARK-20334][SQL] Return a better error message when correlated predicates contain aggregate expression that has mixture of outer and local references. #17636

[SPARK-20334][SQL] Return a better error message when correlated predicates contain aggregate expression that has mixture of outer and local references. #17636

Conversation

dilipbiswal commented Apr 14, 2017 • edited

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Apr 14, 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

Choose a reason for hiding this comment

SparkQA commented Apr 14, 2017

Choose a reason for hiding this comment

gatorsmile commented Apr 14, 2017

viirya commented Apr 14, 2017

viirya Apr 14, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya Apr 14, 2017 • edited

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

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 14, 2017

SparkQA commented Apr 14, 2017

SparkQA commented Apr 14, 2017

dilipbiswal commented Apr 17, 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

Choose a reason for hiding this comment

SparkQA commented Apr 18, 2017

SparkQA commented Apr 18, 2017

dilipbiswal commented Apr 18, 2017

SparkQA commented Apr 18, 2017

SparkQA commented Apr 19, 2017

dilipbiswal commented Apr 19, 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

SparkQA commented Apr 20, 2017

gatorsmile commented Apr 20, 2017

hvanhovell commented Apr 20, 2017

dilipbiswal commented Apr 20, 2017

dilipbiswal commented Apr 14, 2017 •

edited

viirya Apr 14, 2017 •

edited

viirya Apr 14, 2017 •

edited