maropu commented on a change in pull request #32303: URL: https://github.com/apache/spark/pull/32303#discussion_r637627094
########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ########## @@ -1663,6 +1675,27 @@ class Analyzer(override val catalogManager: CatalogManager) } } + /** + * Optionally resolve the name parts using the outer query plan and wrap resolved attributes + * with [[OuterReference]]s. + */ + private def resolveOuterReference( Review comment: nit: `tryResolveOuterReference`? ########## File path: sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 ########## @@ -643,12 +643,12 @@ setQuantifier ; relation - : relationPrimary joinRelation* + : LATERAL? relationPrimary joinRelation* ; joinRelation - : (joinType) JOIN right=relationPrimary joinCriteria? - | NATURAL joinType JOIN right=relationPrimary + : (joinType) JOIN LATERAL? right=relationPrimary joinCriteria? + | NATURAL joinType JOIN LATERAL? right=relationPrimary Review comment: Oh, super cleaner than the previous one. Nice. ########## File path: sql/core/src/test/resources/sql-tests/inputs/join-lateral.sql ########## @@ -0,0 +1,97 @@ +-- Test cases for lateral join + +CREATE VIEW t1(c1, c2) AS VALUES (0, 1), (1, 2); +CREATE VIEW t2(c1, c2) AS VALUES (0, 2), (0, 3); + +-- lateral join with single column select +SELECT * FROM t1, LATERAL (SELECT c1); +SELECT * FROM t1, LATERAL (SELECT c1 FROM t2); +SELECT * FROM t1, LATERAL (SELECT t1.c1 FROM t2); +SELECT * FROM t1, LATERAL (SELECT t1.c1 + t2.c1 FROM t2); + +-- lateral join with star expansion Review comment: Could you add tests for `quotedRegexColumnNames`, too? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala ########## @@ -168,6 +168,21 @@ object EliminateOuterJoin extends Rule[LogicalPlan] with PredicateHelper { } } +/** + * Rewrite lateral joins by rewriting all dependent joins (if any) inside the right + * sub-tree of the lateral join and converting the lateral join into a base join type. + */ +object RewriteLateralJoin extends Rule[LogicalPlan] with PredicateHelper { + + def apply(plan: LogicalPlan): LogicalPlan = plan transformUp { + case j @ Join(left, right, LateralJoin(joinType), condition, _) => + val conditions = condition.map(splitConjunctivePredicates).getOrElse(Nil) + val newRight = DecorrelateInnerQuery.rewriteDomainJoins(left, right, conditions) + // TODO: handle the COUNT bug Review comment: Can you add the jira number (SPARK-15370) here? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ########## @@ -1607,33 +1622,30 @@ class Analyzer(override val catalogManager: CatalogManager) } } f1.copy(arguments = f1.arguments.flatMap { - case s: Star => s.expand(child, resolver) + case s: Star => expand(s) case o => o :: Nil }) case c: CreateNamedStruct if containsStar(c.valExprs) => val newChildren = c.children.grouped(2).flatMap { - case Seq(k, s : Star) => CreateStruct(s.expand(child, resolver)).children + case Seq(k, s : Star) => CreateStruct(expand(s)).children case kv => kv } c.copy(children = newChildren.toList ) case c: CreateArray if containsStar(c.children) => c.copy(children = c.children.flatMap { - case s: Star => s.expand(child, resolver) + case s: Star => expand(s) case o => o :: Nil }) case p: Murmur3Hash if containsStar(p.children) => p.copy(children = p.children.flatMap { - case s: Star => s.expand(child, resolver) + case s: Star => expand(s) case o => o :: Nil }) case p: XxHash64 if containsStar(p.children) => p.copy(children = p.children.flatMap { - case s: Star => s.expand(child, resolver) + case s: Star => expand(s) case o => o :: Nil }) - // count(*) has been replaced by count(1) - case o if containsStar(o.children) => - throw QueryCompilationErrors.invalidStarUsageError(s"expression '${o.prettyName}'") Review comment: Did this error handling is moved into `CheckAnalysis` https://github.com/apache/spark/pull/32303/files#diff-583171e935b2dc349378063a5841c5b98b30a2d57ac3743a9eccfe7bffcb8f2aR177-R179 ? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ########## @@ -276,6 +276,7 @@ class Analyzer(override val catalogManager: CatalogManager) ResolveAliases :: ResolveSubquery :: ResolveSubqueryColumnAliases :: + ResolveLateralJoin :: Review comment: Since `ResolveLateralJoin` has a similar functionality with `ResolveReferences`, how about putting this rule just after `ResolveReferences`? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ########## @@ -2395,6 +2421,70 @@ class Analyzer(override val catalogManager: CatalogManager) } } + /** + * This rule resolves lateral joins. + */ + object ResolveLateralJoin extends Rule[LogicalPlan] { Review comment: `ResolveReferencesInLateralJoin`? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala ########## @@ -174,13 +174,21 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog { hof.failAnalysis( s"cannot resolve '${hof.sql}' due to argument data type mismatch: $message") } + // Check if there are expressions with children containing star expressions. + case e if e.children.exists(_.isInstanceOf[Star]) => + e.failAnalysis(s"Invalid usage of '*' in expression '${e.prettyName}'") } operator transformExpressionsUp { case a: Attribute if !a.resolved => val from = operator.inputSet.toSeq.map(_.qualifiedName).mkString(", ") a.failAnalysis(s"cannot resolve '${a.sql}' given input columns: [$from]") + case s @ UnresolvedStar(Some(target)) => Review comment: We still need this error check even if we check it in L178? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala ########## @@ -174,13 +174,21 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog { hof.failAnalysis( s"cannot resolve '${hof.sql}' due to argument data type mismatch: $message") } + // Check if there are expressions with children containing star expressions. + case e if e.children.exists(_.isInstanceOf[Star]) => + e.failAnalysis(s"Invalid usage of '*' in expression '${e.prettyName}'") Review comment: We cannot use `QueryCompilationErrors.invalidStarUsageError` here? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala ########## @@ -244,7 +244,8 @@ object DecorrelateInnerQuery extends PredicateHelper { case _ => Join(child, domain, Inner, None, JoinHint.NONE) } } else { - throw QueryExecutionErrors.cannotRewriteDomainJoinWithConditionsError(conditions, d) Review comment: ditto: It seems no one uses this method now? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala ########## @@ -254,6 +262,27 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog { s"join condition '${condition.sql}' " + s"of type ${condition.dataType.catalogString} is not a boolean.") + case Join(left, right, LateralJoin(_), _, _) => + // Check all outer references in the right operand are a subset of output columns + // of the left operand. + def checkOuterReferences(plan: LogicalPlan, outer: LogicalPlan): Unit = { + plan match { + case Join(left, _, LateralJoin(_), _, _) => + checkOuterReferences(left, outer) + case p: LogicalPlan => p transformExpressions { + case o @ OuterReference(e) if !outer.outputSet.contains(e) => + val column = outer.outputSet.toSeq.map(_.qualifiedName).mkString(", ") + o.failAnalysis( + s"Found an outer column reference '${e.sql}' in a lateral subquery " + + s"that is not present in the preceding FROM items of its own query " + Review comment: nit: remove `s` ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala ########## @@ -402,10 +402,8 @@ case class UnresolvedStar(target: Option[Seq[String]]) extends Star with Unevalu throw QueryCompilationErrors.starExpandDataTypeNotSupportedError(target.get) } } else { - val from = input.inputSet.map(_.name).mkString(", ") - val targetString = target.get.mkString(".") - throw QueryCompilationErrors.cannotResolveStarExpandGivenInputColumnsError( Review comment: It seems no one uses this method now? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala ########## @@ -174,13 +174,21 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog { hof.failAnalysis( s"cannot resolve '${hof.sql}' due to argument data type mismatch: $message") } + // Check if there are expressions with children containing star expressions. + case e if e.children.exists(_.isInstanceOf[Star]) => + e.failAnalysis(s"Invalid usage of '*' in expression '${e.prettyName}'") Review comment: If we add this rule here, I think we need to update the comment above. https://github.com/apache/spark/pull/32303/files#diff-583171e935b2dc349378063a5841c5b98b30a2d57ac3743a9eccfe7bffcb8f2aL165-L168 ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala ########## @@ -107,6 +107,11 @@ case class UsingJoin(tpe: JoinType, usingColumns: Seq[String]) extends JoinType override def sql: String = "USING " + tpe.sql } +case class LateralJoin(tpe: JoinType) extends JoinType { + require(Seq(Inner, LeftOuter, Cross).contains(tpe), "Unsupported lateral join type " + tpe) Review comment: > Currently INNER, CROSS and LEFT OUTER join types are supported with lateral join If we can support the other join types in future, how about checking this requirement in the `AstBuilder` side? ########## File path: sql/core/src/test/resources/sql-tests/inputs/postgreSQL/groupingsets.sql ########## @@ -499,14 +499,14 @@ SELECT a, b, count(*), max(a), max(b) FROM gstest3 GROUP BY GROUPING SETS(a, b,( -- COMMIT; -- More rescan tests --- [SPARK-27877] ANSI SQL: LATERAL derived table(T491) +-- Expressions referencing the outer query are not supported outside of WHERE/HAVING clauses: -- select * from (values (1),(2)) v(a) left join lateral (select v.a, four, ten, count(*) from onek group by cube(four,ten)) s on true order by v.a,four,ten; -- [SPARK-27878] Support ARRAY(sub-SELECT) expressions -- select array(select row(v.a,s1.*) from (select two,four, count(*) from onek group by cube(two,four) order by two,four) s1) from (values (1),(2)) v(a); -- Rescan logic changes when there are no empty grouping sets, so test -- that too: --- [SPARK-27877] ANSI SQL: LATERAL derived table(T491) +-- Expressions referencing the outer query are not supported outside of WHERE/HAVING clauses: Review comment: We can support this case in future? If we can, can you file jira for it and update the jira number above (`[SPARK-27877] ANSI SQL: LATERAL derived table(T491)`)? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ########## @@ -2395,6 +2421,70 @@ class Analyzer(override val catalogManager: CatalogManager) } } + /** + * This rule resolves lateral joins. + */ + object ResolveLateralJoin extends Rule[LogicalPlan] { Review comment: Q: Btw, is it difficult to merge this resolution logic with the `ResolveReferences` one? ########## File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala ########## @@ -182,6 +182,16 @@ object DeduplicateRelations extends Rule[LogicalPlan] { if findAliases(projectList).intersect(conflictingAttributes).nonEmpty => Seq((oldVersion, oldVersion.copy(projectList = newAliases(projectList)))) + // Handle projects that create conflicting outer references. + case oldVersion @ Project(projectList, _) Review comment: cc: @Ngone51 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org