peter-toth commented on a change in pull request #23531: [SPARK-24497][SQL]
Support recursive SQL query
URL: https://github.com/apache/spark/pull/23531#discussion_r248569542
##########
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -205,30 +206,145 @@ class Analyzer(
CleanupAliases)
)
+ object ResolveRecursiveReferneces extends Rule[LogicalPlan] {
+ def apply(plan: LogicalPlan): LogicalPlan = {
+ val recursiveTables = plan.collect {
+ case rt @ RecursiveTable(name, _, _) if rt.anchorResolved => name -> rt
+ }.toMap
+
+ plan.resolveOperatorsUp {
+ case UnresolvedRecursiveReference(name) if
recursiveTables.contains(name) =>
+ RecursiveReference(name,
recursiveTables(name).output.map(_.newInstance()))
+ case other => other
+ }
+ }
+ }
+
/**
* Analyze cte definitions and substitute child plan with analyzed cte
definitions.
*/
object CTESubstitution extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
- case With(child, relations) =>
+ case With(child, relations, allowRecursion) =>
substituteCTE(child, relations.foldLeft(Seq.empty[(String,
LogicalPlan)]) {
case (resolved, (name, relation)) =>
- resolved :+ name -> executeSameContext(substituteCTE(relation,
resolved))
- })
+ val recursiveTableName = if (allowRecursion) Some(name) else None
+ resolved :+
+ name -> executeSameContext(substituteCTE(relation, resolved,
recursiveTableName))
+ }, None)
case other => other
}
- def substituteCTE(plan: LogicalPlan, cteRelations: Seq[(String,
LogicalPlan)]): LogicalPlan = {
- plan resolveOperatorsDown {
- case u: UnresolvedRelation =>
- cteRelations.find(x => resolver(x._1, u.tableIdentifier.table))
- .map(_._2).getOrElse(u)
- case other =>
- // This cannot be done in ResolveSubquery because ResolveSubquery
does not know the CTE.
- other transformExpressions {
- case e: SubqueryExpression =>
- e.withNewPlan(substituteCTE(e.plan, cteRelations))
+ def substituteCTE(
+ plan: LogicalPlan,
+ cteRelations: Seq[(String, LogicalPlan)],
+ recursiveTableName: Option[String]): LogicalPlan = {
+ def substitute(
+ plan: LogicalPlan,
+ inSubQuery: Boolean = false): (LogicalPlan, Boolean) = {
+ val references = mutable.Set.empty[UnresolvedRecursiveReference]
+
+ def newReference(recursiveTableName: String) = {
+ val recursiveReference =
UnresolvedRecursiveReference(recursiveTableName)
+ references += recursiveReference
+
+ recursiveReference
+ }
+
+ val newPlan = plan resolveOperatorsDown {
+ case u: UnresolvedRelation =>
+ val table = u.tableIdentifier.table
+
+ val recursiveReference = recursiveTableName.find(resolver(_,
table)).map { name =>
+ if (inSubQuery) {
+ throw new AnalysisException(
+ s"Recursive reference ${name} can't be used in a subquery")
+ }
+
+ newReference(name)
+ }
+
+ recursiveReference
+ .orElse(cteRelations.find(x => resolver(x._1, table)).map(_._2))
+ .getOrElse(u)
+
+ case other =>
+ // This cannot be done in ResolveSubquery because ResolveSubquery
does not know the CTE.
+ other transformExpressions {
+ case e: SubqueryExpression => e.withNewPlan(substitute(e.plan,
true)._1)
+ }
+ }
+
+ (newPlan, !references.isEmpty)
+ }
+
+ plan match {
+ case SubqueryAlias(name, u: Union) if recursiveTableName.isDefined =>
+ def combineUnions(union: Union): Seq[LogicalPlan] =
union.children.flatMap {
+ case u: Union => combineUnions(u)
+ case o => Seq(o)
}
+
+ val substitutedTerms = combineUnions(u).map(substitute(_))
+ val (anchorTerms, recursiveTerms) = substitutedTerms.partition(!_._2)
+
+ if (!recursiveTerms.isEmpty) {
+ if (anchorTerms.isEmpty) {
+ throw new AnalysisException("There should be at least 1 anchor
term defined in a " +
+ s"recursive query $name")
+ }
+
+ val recursiveTermPlans = recursiveTerms.map(_._1)
+
+ def traversePlanAndCheck(
+ plan: LogicalPlan,
+ isRecursiveReferenceAllowed: Boolean = true): Boolean = plan
match {
+ case UnresolvedRecursiveReference(name) =>
+ if (!isRecursiveReferenceAllowed) {
+ throw new AnalysisException(s"Wrong usage of recursive
reference ${name}")
+ }
+ true
+ case Join(left, right, Inner, _, _) =>
+ val l = traversePlanAndCheck(left, isRecursiveReferenceAllowed)
+ val r = traversePlanAndCheck(right,
isRecursiveReferenceAllowed)
+ if (l && r) {
Review comment:
Basically only those operations are allowed where you can compute the
current set of rows from the result of the previous iteration.
When it comes to join I think this rule implies that the recursive reference
cannot be used on both sides of any kind of join because otherwise if you don't
join by the same columns then you need the cumulated result of recursion to
compute the result of the current iteration. For example if `t` is recursive
the statement `FROM t AS a [? OUTER ] JOIN t AS b on b.col1 = a.col2` can't be
computed based on the result of the previous iteration only.
We could allow FROM t AS a [? OUTER ] JOIN t AS b on b.col1 = a.col1` kind
of statements, but IMHO that doesn't make much sense to make as an exception.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
