[WIP][SPARK-24497][SQL] Support recursive SQL query with adaptive replanning #23531

peter-toth · 2019-01-13T12:00:07Z

What changes were proposed in this pull request?

This PR adds recursive query feature to Spark SQL.

A recursive query is defined using the WITH RECURSIVE keywords and referring the name of the common table expression within the query.
The implementation complies with SQL standard and follows similar rules to other relational databases:

A query is made of an anchor followed by a recursive term.
The anchor terms doesn't contain self reference and it is used to initialize the query.
The recursive term contains a self reference and it is used to expand the current set of rows with new ones.
The anchor and recursive terms must be joined with each other by UNION or UNION ALL operators.
New rows can only be derived from the newly added rows of the previous iteration (or from the initial set of rows of anchor terms). This limitation implies that recursive references can't be used with some of the joins, aggregations or subqueries.

Please see cte-recursive.sql and with.sql for some examples.

Please note that this PR focuses on the minimal working implementation which means:

SQL recursion is actually loop where the current iteration is computed based on the previous one's result and when an iteration returns no rows the loop is over. The final result is the union of all iteration results. This means that caching intermediate results could speed up the process, but caching was removed from this PR to reduce complexity and can be added back in a follow-up PR.
A common way to stop SQL recursion is using the LIMIT operator to stop computing more than the required number of rows. LIMIT support was removed from this PR to reduce complexity and can be added back in a follow-up PR.
Some relational databases are more relaxed in terms how many anchor and recursive terms can be in a recursion. This PR allows the most simple case and allows only 1-1 of them. A follow-up PR can target to relax this limitation.

Why are the changes needed?

Recursive query is an ANSI SQL feature that is useful to process hierarchical data.

Does this PR introduce any user-facing change?

Yes, adds recursive query feature.

How was this patch tested?

Added new UTs and tests in cte-recursion.sql and with.sql.

gatorsmile · 2019-01-13T23:32:39Z

ok to test

gatorsmile · 2019-01-13T23:34:49Z

cc @maryannxue

SparkQA · 2019-01-14T03:42:02Z

Test build #101149 has finished for PR 23531 at commit 1c81909.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class PlanTraverseStatus(
case class UnresolvedRecursiveReference(name: String) extends LeafNode
case class RecursiveTable(
case class RecursiveReference(name: String, output: Seq[Attribute]) extends LeafNode
case class With(
case class RecursiveTableExec(
case class RecursiveReferenceExec(name: String, output: Seq[Attribute]) extends LeafExecNode

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

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala

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

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

SparkQA · 2019-01-14T15:25:02Z

Test build #101190 has started for PR 23531 at commit a69473f.

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala

SparkQA · 2019-01-14T15:47:37Z

Test build #101191 has started for PR 23531 at commit f6d8c17.

shaneknapp · 2019-01-14T18:59:35Z

test this please

shaneknapp · 2019-01-14T19:09:59Z

test this please

SparkQA · 2019-01-14T23:14:59Z

Test build #101197 has finished for PR 23531 at commit f6d8c17.

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

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

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

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

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

SparkQA · 2019-01-16T13:30:20Z

Test build #101306 has finished for PR 23531 at commit 5f69d60.

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

SparkQA · 2019-01-16T13:59:47Z

Test build #101308 has finished for PR 23531 at commit d9cc864.

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

SparkQA · 2019-01-16T14:14:22Z

Test build #101304 has finished for PR 23531 at commit 4ea3a4a.

This patch passes all tests.
This patch does not merge cleanly.
This patch adds no public classes.

SparkQA · 2019-01-16T14:45:40Z

Test build #101310 has finished for PR 23531 at commit 8f9a673.

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

SparkQA · 2019-01-16T23:21:55Z

Test build #101327 has finished for PR 23531 at commit f5feb63.

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

sql/core/src/test/resources/sql-tests/inputs/recursion.sql

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

SparkQA · 2019-01-22T19:24:39Z

Test build #101546 has finished for PR 23531 at commit 24af7b2.

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

SparkQA · 2019-01-23T00:28:46Z

Test build #101555 has finished for PR 23531 at commit 0c6e957.

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

SparkQA · 2019-01-23T01:29:03Z

Test build #101559 has finished for PR 23531 at commit 31574d7.

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

SparkQA · 2019-01-23T12:43:49Z

Test build #101585 has finished for PR 23531 at commit aab0886.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

peter-toth · 2020-03-26T08:37:33Z

retest this please

SparkQA · 2020-03-26T13:39:41Z

Test build #120411 has finished for PR 23531 at commit 656995b.

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

HyukjinKwon · 2020-04-03T12:10:23Z

retest this please

SparkQA · 2020-04-03T16:54:19Z

Test build #120774 has finished for PR 23531 at commit 656995b.

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

dmateusp · 2020-04-08T08:57:02Z

seems like the PR addressed the feedback, can someone review?

…497-recursive-sql

SparkQA · 2020-04-14T21:02:11Z

Test build #121283 has finished for PR 23531 at commit 8205e61.

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

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

maryannxue · 2020-04-20T17:15:01Z

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

+
+  override def children: Seq[SparkPlan] = anchorTerm :: Nil
+
+  override def innerChildren: Seq[QueryPlan[_]] = logicalRecursiveTerm +: super.innerChildren


Any specific reason we need to include logicalResursiveTerm as innerChildren other than for display?

No specific reason, just for display. I can find a better way to do it.

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

maryannxue · 2020-04-20T17:23:54Z

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

+   * Notify the listeners of the physical plan change.
+   */
+  private def onUpdatePlan(executionId: Long): Unit = {
+    val queryExecution = SQLExecution.getQueryExecution(executionId)


Not sure if we should introduce the idea of iterations in metrics here. Or maybe we need a better way of expressing recursion in UI.

I liked the idea that we can see what happened in each iteration, but I agree that it can be too much information to present on UI, especially when recursion level is high.
Do you have any idea how to present recursion on the UI?

For performance and UI readability reasons, we don't wanna really update all those physical nodes to the UI. AQE has metric-only UI updates, and I assume we can do sth. similar here.
Would you mind doing a little research on how other SQL engines or tools present recursive queries in GUIs?

I see.
Sure, I'm happy to look into it how other engines show recursion on their UI.

Unfortunately, I didn't have time for this yet, but it is still on my list.

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

peter-toth · 2020-04-21T14:27:18Z

Thanks @maryannxue for your comments, really appreciated. I will try to address them soon.

…497-recursive-sql

SparkQA · 2020-06-18T09:50:59Z

Test build #124213 has finished for PR 23531 at commit fc7b04a.

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

…497-recursive-sql

SparkQA · 2020-06-18T12:12:49Z

Test build #124214 has finished for PR 23531 at commit 607aa4d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class TimeFormatters(date: DateFormatter, timestamp: TimestampFormatter)

SparkQA · 2020-06-19T00:01:35Z

Test build #124233 has finished for PR 23531 at commit adf120c.

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

SparkQA · 2020-06-20T00:51:18Z

Test build #124299 has finished for PR 23531 at commit 69202d5.

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

peter-toth · 2020-07-23T19:01:01Z

I've opened a simplified PR (without adaptive replanning in each iteration) here: #29210. That implementation is bit simpler and hopefully will get merged sooner than this PR.

github-actions · 2020-11-01T00:36:56Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

mgaido91 reviewed Jan 14, 2019

View reviewed changes

peter-toth force-pushed the SPARK-24497 branch from 4ea3a4a to 5f69d60 Compare January 16, 2019 09:23