This is an automated email from the ASF dual-hosted git repository. gengliang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new 20edfdd [SPARK-35845][SQL] OuterReference resolution should reject ambiguous column names 20edfdd is described below commit 20edfdd39a83c52813f91e4028f816d06a6be99e Author: Wenchen Fan <wenc...@databricks.com> AuthorDate: Wed Jun 23 14:32:34 2021 +0800 [SPARK-35845][SQL] OuterReference resolution should reject ambiguous column names ### What changes were proposed in this pull request? The current OuterReference resolution is a bit weird: when the outer plan has more than one child, it resolves OuterReference from the output of each child, one by one, left to right. This is incorrect in the case of join, as the column name can be ambiguous if both left and right sides output this column. This PR fixes this bug by resolving OuterReference with `outerPlan.resolveChildren`, instead of something like `outerPlan.children.foreach(_.resolve(...))` ### Why are the changes needed? bug fix ### Does this PR introduce _any_ user-facing change? The problem only occurs in join, and join condition doesn't support correlated subquery yet. So this PR only improves the error message. Before this PR, people see ``` java.lang.UnsupportedOperationException Cannot generate code for expression: outer(t1a#291) ``` ### How was this patch tested? a new test Closes #33004 from cloud-fan/outer-ref. Authored-by: Wenchen Fan <wenc...@databricks.com> Signed-off-by: Gengliang Wang <gengli...@apache.org> --- .../spark/sql/catalyst/analysis/Analyzer.scala | 35 +++++++++++----------- .../catalyst/optimizer/DecorrelateInnerQuery.scala | 10 ++----- .../spark/sql/catalyst/optimizer/subquery.scala | 26 ++++++++-------- .../optimizer/DecorrelateInnerQuerySuite.scala | 6 ++-- .../negative-cases/invalid-correlation.sql | 9 ++++++ .../negative-cases/invalid-correlation.sql.out | 24 ++++++++++++++- 6 files changed, 68 insertions(+), 42 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 555be01..ba680ba 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -2285,8 +2285,8 @@ class Analyzer(override val catalogManager: CatalogManager) } /** - * Resolve the correlated expressions in a subquery by using the an outer plans' references. All - * resolved outer references are wrapped in an [[OuterReference]] + * Resolve the correlated expressions in a subquery, as if the expressions live in the outer + * plan. All resolved outer references are wrapped in an [[OuterReference]] */ private def resolveOuterReferences(plan: LogicalPlan, outer: LogicalPlan): LogicalPlan = { plan.resolveOperatorsDownWithPruning(_.containsPattern(UNRESOLVED_ATTRIBUTE)) { @@ -2295,7 +2295,7 @@ class Analyzer(override val catalogManager: CatalogManager) case u @ UnresolvedAttribute(nameParts) => withPosition(u) { try { - outer.resolve(nameParts, resolver) match { + outer.resolveChildren(nameParts, resolver) match { case Some(outerAttr) => wrapOuterReference(outerAttr) case None => u } @@ -2317,7 +2317,7 @@ class Analyzer(override val catalogManager: CatalogManager) */ private def resolveSubQuery( e: SubqueryExpression, - plans: Seq[LogicalPlan])( + outer: LogicalPlan)( f: (LogicalPlan, Seq[Expression]) => SubqueryExpression): SubqueryExpression = { // Step 1: Resolve the outer expressions. var previous: LogicalPlan = null @@ -2328,10 +2328,8 @@ class Analyzer(override val catalogManager: CatalogManager) current = executeSameContext(current) // Use the outer references to resolve the subquery plan if it isn't resolved yet. - val i = plans.iterator - val afterResolve = current - while (!current.resolved && current.fastEquals(afterResolve) && i.hasNext) { - current = resolveOuterReferences(current, i.next()) + if (!current.resolved) { + current = resolveOuterReferences(current, outer) } } while (!current.resolved && !current.fastEquals(previous)) @@ -2354,20 +2352,20 @@ class Analyzer(override val catalogManager: CatalogManager) * (2) Any aggregate expression(s) that reference outer attributes are pushed down to * outer plan to get evaluated. */ - private def resolveSubQueries(plan: LogicalPlan, plans: Seq[LogicalPlan]): LogicalPlan = { + private def resolveSubQueries(plan: LogicalPlan, outer: LogicalPlan): LogicalPlan = { plan.transformAllExpressionsWithPruning(_.containsPattern(PLAN_EXPRESSION), ruleId) { case s @ ScalarSubquery(sub, _, exprId, _) if !sub.resolved => - resolveSubQuery(s, plans)(ScalarSubquery(_, _, exprId)) + resolveSubQuery(s, outer)(ScalarSubquery(_, _, exprId)) case e @ Exists(sub, _, exprId, _) if !sub.resolved => - resolveSubQuery(e, plans)(Exists(_, _, exprId)) + resolveSubQuery(e, outer)(Exists(_, _, exprId)) case InSubquery(values, l @ ListQuery(_, _, exprId, _, _)) if values.forall(_.resolved) && !l.resolved => - val expr = resolveSubQuery(l, plans)((plan, exprs) => { + val expr = resolveSubQuery(l, outer)((plan, exprs) => { ListQuery(plan, exprs, exprId, plan.output) }) InSubquery(values, expr.asInstanceOf[ListQuery]) case s @ LateralSubquery(sub, _, exprId, _) if !sub.resolved => - resolveSubQuery(s, plans)(LateralSubquery(_, _, exprId)) + resolveSubQuery(s, outer)(LateralSubquery(_, _, exprId)) } } @@ -2377,14 +2375,17 @@ class Analyzer(override val catalogManager: CatalogManager) def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning( _.containsPattern(PLAN_EXPRESSION), ruleId) { case j: LateralJoin if j.left.resolved => - resolveSubQueries(j, j.children) + // We can't pass `LateralJoin` as the outer plan, as its right child is not resolved yet + // and we can't call `LateralJoin.resolveChildren` to resolve outer references. Here we + // create a fake Project node as the outer plan. + resolveSubQueries(j, Project(Nil, j.left)) // Only a few unary nodes (Project/Filter/Aggregate) can contain subqueries. case q: UnaryNode if q.childrenResolved => - resolveSubQueries(q, q.children) + resolveSubQueries(q, q) case j: Join if j.childrenResolved && j.duplicateResolved => - resolveSubQueries(j, j.children) + resolveSubQueries(j, j) case s: SupportsSubquery if s.childrenResolved => - resolveSubQueries(s, s.children) + resolveSubQueries(s, s) } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala index a2404f0..f0441e3 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala @@ -258,13 +258,7 @@ object DecorrelateInnerQuery extends PredicateHelper { def apply( innerPlan: LogicalPlan, outerPlan: LogicalPlan): (LogicalPlan, Seq[Expression]) = { - apply(innerPlan, Seq(outerPlan)) - } - - def apply( - innerPlan: LogicalPlan, - outerPlans: Seq[LogicalPlan]): (LogicalPlan, Seq[Expression]) = { - val outputSet = AttributeSet(outerPlans.flatMap(_.outputSet)) + val outputPlanInputAttrs = outerPlan.inputSet // The return type of the recursion. // The first parameter is a new logical plan with correlation eliminated. @@ -486,7 +480,7 @@ object DecorrelateInnerQuery extends PredicateHelper { } } val (newChild, joinCond, _) = decorrelate(BooleanSimplification(innerPlan), AttributeSet.empty) - val (plan, conditions) = deduplicate(newChild, joinCond, outputSet) + val (plan, conditions) = deduplicate(newChild, joinCond, outputPlanInputAttrs) (plan, stripOuterReferences(conditions)) } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala index ab9c827..7914d14 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala @@ -220,7 +220,7 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper */ private def pullOutCorrelatedPredicates( sub: LogicalPlan, - outer: Seq[LogicalPlan]): (LogicalPlan, Seq[Expression]) = { + outer: LogicalPlan): (LogicalPlan, Seq[Expression]) = { val predicateMap = scala.collection.mutable.Map.empty[LogicalPlan, Seq[Expression]] /** Determine which correlated predicate references are missing from this plan. */ @@ -272,10 +272,10 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper // In case of a collision, change the subquery plan's output to use // different attribute by creating alias(s). val baseConditions = predicateMap.values.flatten.toSeq - val (newPlan, newCond) = if (outer.nonEmpty) { - val outputSet = outer.map(_.outputSet).reduce(_ ++ _) + val outerPlanInputAttrs = outer.inputSet + val (newPlan, newCond) = if (outerPlanInputAttrs.nonEmpty) { val (plan, deDuplicatedConditions) = - DecorrelateInnerQuery.deduplicate(transformed, baseConditions, outputSet) + DecorrelateInnerQuery.deduplicate(transformed, baseConditions, outerPlanInputAttrs) (plan, stripOuterReferences(deDuplicatedConditions)) } else { (transformed, stripOuterReferences(baseConditions)) @@ -283,7 +283,7 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper (newPlan, newCond) } - private def rewriteSubQueries(plan: LogicalPlan, outerPlans: Seq[LogicalPlan]): LogicalPlan = { + private def rewriteSubQueries(plan: LogicalPlan): LogicalPlan = { /** * This function is used as a aid to enforce idempotency of pullUpCorrelatedPredicate rule. * In the first call to rewriteSubqueries, all the outer references from the subplan are @@ -296,7 +296,7 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper if (newCond.isEmpty) oldCond else newCond } - def decorrelate(sub: LogicalPlan, outer: Seq[LogicalPlan]): (LogicalPlan, Seq[Expression]) = { + def decorrelate(sub: LogicalPlan, outer: LogicalPlan): (LogicalPlan, Seq[Expression]) = { if (SQLConf.get.decorrelateInnerQueryEnabled) { DecorrelateInnerQuery(sub, outer) } else { @@ -306,16 +306,16 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper plan.transformExpressionsWithPruning(_.containsPattern(PLAN_EXPRESSION)) { case ScalarSubquery(sub, children, exprId, conditions) if children.nonEmpty => - val (newPlan, newCond) = decorrelate(sub, outerPlans) + val (newPlan, newCond) = decorrelate(sub, plan) ScalarSubquery(newPlan, children, exprId, getJoinCondition(newCond, conditions)) case Exists(sub, children, exprId, conditions) if children.nonEmpty => - val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) + val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, plan) Exists(newPlan, children, exprId, getJoinCondition(newCond, conditions)) case ListQuery(sub, children, exprId, childOutputs, conditions) if children.nonEmpty => - val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) + val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, plan) ListQuery(newPlan, children, exprId, childOutputs, getJoinCondition(newCond, conditions)) case LateralSubquery(sub, children, exprId, conditions) if children.nonEmpty => - val (newPlan, newCond) = decorrelate(sub, outerPlans) + val (newPlan, newCond) = decorrelate(sub, plan) LateralSubquery(newPlan, children, exprId, getJoinCondition(newCond, conditions)) } } @@ -326,7 +326,7 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning( _.containsPattern(PLAN_EXPRESSION)) { case j: LateralJoin => - val newPlan = rewriteSubQueries(j, j.children) + val newPlan = rewriteSubQueries(j) // Since a lateral join's output depends on its left child output and its lateral subquery's // plan output, we need to trim the domain attributes added to the subquery's plan output // to preserve the original output of the join. @@ -337,9 +337,9 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper } // Only a few unary nodes (Project/Filter/Aggregate) can contain subqueries. case q: UnaryNode => - rewriteSubQueries(q, q.children) + rewriteSubQueries(q) case s: SupportsSubquery => - rewriteSubQueries(s, s.children) + rewriteSubQueries(s) } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuerySuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuerySuite.scala index 93b2703..92995c2 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuerySuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuerySuite.scala @@ -44,7 +44,7 @@ class DecorrelateInnerQuerySuite extends PlanTest { outerPlan: LogicalPlan, correctAnswer: LogicalPlan, conditions: Seq[Expression]): Unit = { - val (outputPlan, joinCond) = DecorrelateInnerQuery(innerPlan, outerPlan) + val (outputPlan, joinCond) = DecorrelateInnerQuery(innerPlan, outerPlan.select()) assert(!hasOuterReferences(outputPlan)) comparePlans(outputPlan, correctAnswer) assert(joinCond.length == conditions.length) @@ -90,7 +90,7 @@ class DecorrelateInnerQuerySuite extends PlanTest { Project(Seq(a), Filter(OuterReference(a) === a, testRelation)) - val (outputPlan, joinCond) = DecorrelateInnerQuery(innerPlan, outerPlan) + val (outputPlan, joinCond) = DecorrelateInnerQuery(innerPlan, outerPlan.select()) val a1 = outputPlan.output.head val correctAnswer = Project(Seq(Alias(a, a1.name)(a1.exprId)), @@ -197,7 +197,7 @@ class DecorrelateInnerQuerySuite extends PlanTest { Inner, Some(OuterReference(x) === a), JoinHint.NONE) - val error = intercept[AssertionError] { DecorrelateInnerQuery(innerPlan, outerPlan) } + val error = intercept[AssertionError] { DecorrelateInnerQuery(innerPlan, outerPlan.select()) } assert(error.getMessage.contains("Correlated column is not allowed in join")) } diff --git a/sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql b/sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql index 109ffa7..1260fb7 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/subquery/negative-cases/invalid-correlation.sql @@ -71,3 +71,12 @@ WHERE t1a IN (SELECT t2a WHERE EXISTS (SELECT min(t2a) FROM t3)); +CREATE TEMPORARY VIEW t1_copy AS SELECT * FROM VALUES + (1, 2, 3) +AS t1(t1a, t1b, t1c); + +-- invalid because column name `t1a` is ambiguous in the subquery. +SELECT t1.t1a +FROM t1 +JOIN t1_copy +ON EXISTS (SELECT 1 FROM t2 WHERE t2a > t1a) diff --git a/sql/core/src/test/resources/sql-tests/results/subquery/negative-cases/invalid-correlation.sql.out b/sql/core/src/test/resources/sql-tests/results/subquery/negative-cases/invalid-correlation.sql.out index ea01d76..8734511 100644 --- a/sql/core/src/test/resources/sql-tests/results/subquery/negative-cases/invalid-correlation.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/subquery/negative-cases/invalid-correlation.sql.out @@ -1,5 +1,5 @@ -- Automatically generated by SQLQueryTestSuite --- Number of queries: 8 +-- Number of queries: 10 -- !query @@ -116,3 +116,25 @@ Aggregate [min(outer(t2a#x)) AS min(outer(t2.t2a))#x] +- Project [t3a#x, t3b#x, t3c#x] +- SubqueryAlias t3 +- LocalRelation [t3a#x, t3b#x, t3c#x] + + +-- !query +CREATE TEMPORARY VIEW t1_copy AS SELECT * FROM VALUES + (1, 2, 3) +AS t1(t1a, t1b, t1c) +-- !query schema +struct<> +-- !query output + + + +-- !query +SELECT t1.t1a +FROM t1 +JOIN t1_copy +ON EXISTS (SELECT 1 FROM t2 WHERE t2a > t1a) +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +cannot resolve 't1a' given input columns: [t2.t2a, t2.t2b, t2.t2c]; line 4 pos 44 --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org