You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2022/03/28 03:49:37 UTC
[spark] branch master updated: [SPARK-38616][SQL] Keep track of SQL query text in Catalyst TreeNode
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 ecfe049 [SPARK-38616][SQL] Keep track of SQL query text in Catalyst TreeNode
ecfe049 is described below
commit ecfe049cc271a3037c5e697be1a1e6af8f61f854
Author: Gengliang Wang <ge...@apache.org>
AuthorDate: Mon Mar 28 11:47:49 2022 +0800
[SPARK-38616][SQL] Keep track of SQL query text in Catalyst TreeNode
### What changes were proposed in this pull request?
Spark SQL uses the class Origin for tracking the position of each TreeNode in the SQL query text. When there is a parser error, we can show the position info in the error message:
```
> sql("create tabe foo(i int)")
org.apache.spark.sql.catalyst.parser.ParseException:
no viable alternative at input 'create tabe'(line 1, pos 7)
== SQL ==
create tabe foo(i int)
-------^^^
```
It contains two fields: line and startPosition. This is enough for the parser since the SQL query text is known.
However, the SQL query text is unknown in the execution phase. Spark SQL can't show the problematic SQL clause on ANSI runtime failures.
This PR is to include the query text in Origin. After this, we can provide details in the error messages of Expressions which can throw runtime exceptions when ANSI mode is on.
### Why are the changes needed?
Currently, there is not enough error context for runtime ANSI failures.
In the following example, the error message only tells that there is a "divide by zero" error, without pointing out where the exact SQL statement is.
```
SELECT
ws2.web_sales / ws1.web_sales web_q1_q2, ss2.store_sales / ss1.store_sales store_q1_q2,
ws3.web_sales / ws2.web_sales web_q2_q3, ss3.store_sales / ss2.store_sales store_q2_q3
FROM
ss ss1, ss ss2, ss ss3, ws ws1, ws ws2, ws ws3
WHERE
....
AND CASE WHEN ws1.web_sales > 0
THEN ws2.web_sales / ws1.web_sales
ELSE NULL END
> CASE WHEN ss1.store_sales > 0
THEN ss2.store_sales / ss1.store_sales
ELSE NULL END
AND CASE WHEN ws2.web_sales > 0
THEN ws3.web_sales / ws2.web_sales
ELSE NULL END
> CASE WHEN ss2.store_sales > 0
THEN ss3.store_sales / ss2.store_sales
ELSE NULL END
ORDER BY ss1.ca_county
```
```
org.apache.spark.SparkArithmeticException: divide by zero at
org.apache.spark.sql.errors.QueryExecutionErrors$.divideByZeroError(QueryExecutionErrors.scala:140) at
org.apache.spark.sql.catalyst.expressions.DivModLike.eval(arithmetic.scala:437) at
org.apache.spark.sql.catalyst.expressions.DivModLike.eval$(arithmetic.scala:425) at
org.apache.spark.sql.catalyst.expressions.Divide.eval(arithmetic.scala:534)
```
This PR is the initial PR for the project https://issues.apache.org/jira/browse/SPARK-38615
The project is able to provide such error context:
```
org.apache.spark.SparkArithmeticException: divide by zero
SparkArithmeticException: divide by zero. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
== SQL(line 2, position 43) ==
ws2.web_sales / ws1.web_sales web_q1_q2, ss2.store_sales / ss1.store_sales store_q1_q2
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT
Closes #35936 from gengliangwang/trackText.
Lead-authored-by: Gengliang Wang <ge...@apache.org>
Co-authored-by: Gengliang Wang <lt...@gmail.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../sql/catalyst/catalog/SessionCatalog.scala | 9 +++-
.../spark/sql/catalyst/parser/ParseDriver.scala | 26 ++++++----
.../spark/sql/catalyst/parser/ParserUtils.scala | 28 ++++++++++-
.../apache/spark/sql/catalyst/trees/TreeNode.scala | 12 ++++-
.../sql/catalyst/parser/ParserUtilsSuite.scala | 56 ++++++++++++++++++++++
.../apache/spark/sql/execution/SQLViewSuite.scala | 55 +++++++++++++++++++++
6 files changed, 174 insertions(+), 12 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 5872f2a..322302e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder
import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, ExpressionInfo, UpCast}
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException, ParserInterface}
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, SubqueryAlias, View}
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, StringUtils}
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -872,9 +873,15 @@ class SessionCatalog(
throw new IllegalStateException("Invalid view without text.")
}
val viewConfigs = metadata.viewSQLConfigs
+ val origin = Origin(
+ objectType = Some("VIEW"),
+ objectName = Some(metadata.qualifiedName)
+ )
val parsedPlan = SQLConf.withExistingConf(View.effectiveSQLConf(viewConfigs, isTempView)) {
try {
- parser.parseQuery(viewText)
+ CurrentOrigin.withOrigin(origin) {
+ parser.parseQuery(viewText)
+ }
} catch {
case _: ParseException =>
throw QueryCompilationErrors.invalidViewText(viewText, metadata.qualifiedName)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala
index 82be4d6..bf0ee9c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala
@@ -26,6 +26,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.{FunctionIdentifier, SQLConfHelper, TableIdentifier}
import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.parser.ParserUtils.withOrigin
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.trees.Origin
import org.apache.spark.sql.errors.QueryParsingErrors
@@ -43,7 +44,10 @@ abstract class AbstractSqlParser extends ParserInterface with SQLConfHelper with
/** Creates Expression for a given SQL string. */
override def parseExpression(sqlText: String): Expression = parse(sqlText) { parser =>
- astBuilder.visitSingleExpression(parser.singleExpression())
+ val ctx = parser.singleExpression()
+ withOrigin(ctx, Some(sqlText)) {
+ astBuilder.visitSingleExpression(ctx)
+ }
}
/** Creates TableIdentifier for a given SQL string. */
@@ -75,16 +79,22 @@ abstract class AbstractSqlParser extends ParserInterface with SQLConfHelper with
/** Creates LogicalPlan for a given SQL string of query. */
override def parseQuery(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
- astBuilder.visitQuery(parser.query())
+ val ctx = parser.query()
+ withOrigin(ctx, Some(sqlText)) {
+ astBuilder.visitQuery(ctx)
+ }
}
/** Creates LogicalPlan for a given SQL string. */
override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
- astBuilder.visitSingleStatement(parser.singleStatement()) match {
- case plan: LogicalPlan => plan
- case _ =>
- val position = Origin(None, None)
- throw QueryParsingErrors.sqlStatementUnsupportedError(sqlText, position)
+ val ctx = parser.singleStatement()
+ withOrigin(ctx, Some(sqlText)) {
+ astBuilder.visitSingleStatement(ctx) match {
+ case plan: LogicalPlan => plan
+ case _ =>
+ val position = Origin(None, None)
+ throw QueryParsingErrors.sqlStatementUnsupportedError(sqlText, position)
+ }
}
}
@@ -271,7 +281,7 @@ class ParseException(
val builder = new StringBuilder
builder ++= "\n" ++= message
start match {
- case Origin(Some(l), Some(p)) =>
+ case Origin(Some(l), Some(p), _, _, _, _, _) =>
builder ++= s"(line $l, pos $p)\n"
command.foreach { cmd =>
val (above, below) = cmd.split("\n").splitAt(l)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala
index 694ddfc..048012a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala
@@ -114,6 +114,24 @@ object ParserUtils {
Origin(opt.map(_.getLine), opt.map(_.getCharPositionInLine))
}
+ def positionAndText(
+ startToken: Token,
+ stopToken: Token,
+ sqlText: String,
+ objectType: Option[String],
+ objectName: Option[String]): Origin = {
+ val startOpt = Option(startToken)
+ val stopOpt = Option(stopToken)
+ Origin(
+ line = startOpt.map(_.getLine),
+ startPosition = startOpt.map(_.getCharPositionInLine),
+ startIndex = startOpt.map(_.getStartIndex),
+ stopIndex = stopOpt.map(_.getStopIndex),
+ sqlText = Some(sqlText),
+ objectType = objectType,
+ objectName = objectName)
+ }
+
/** Validate the condition. If it doesn't throw a parse exception. */
def validate(f: => Boolean, message: String, ctx: ParserRuleContext): Unit = {
if (!f) {
@@ -126,9 +144,15 @@ object ParserUtils {
* registered origin. This method restores the previously set origin after completion of the
* closure.
*/
- def withOrigin[T](ctx: ParserRuleContext)(f: => T): T = {
+ def withOrigin[T](ctx: ParserRuleContext, sqlText: Option[String] = None)(f: => T): T = {
val current = CurrentOrigin.get
- CurrentOrigin.set(position(ctx.getStart))
+ val text = sqlText.orElse(current.sqlText)
+ if (text.isEmpty) {
+ CurrentOrigin.set(position(ctx.getStart))
+ } else {
+ CurrentOrigin.set(positionAndText(ctx.getStart, ctx.getStop, text.get,
+ current.objectType, current.objectName))
+ }
try {
f
} finally {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
index 83814b0..1b0d408 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
@@ -52,9 +52,19 @@ import org.apache.spark.util.collection.BitSet
/** Used by [[TreeNode.getNodeNumbered]] when traversing the tree for a given number */
private class MutableInt(var i: Int)
+/**
+ * Contexts of TreeNodes, including location, SQL text, object type and object name.
+ * The only supported object type is "VIEW" now. In the future, we may support SQL UDF or other
+ * objects which contain SQL text.
+ */
case class Origin(
line: Option[Int] = None,
- startPosition: Option[Int] = None)
+ startPosition: Option[Int] = None,
+ startIndex: Option[Int] = None,
+ stopIndex: Option[Int] = None,
+ sqlText: Option[String] = None,
+ objectType: Option[String] = None,
+ objectName: Option[String] = None)
/**
* Provides a location for TreeNodes to ask about the context of their origin. For example, which
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ParserUtilsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ParserUtilsSuite.scala
index 35f0900..818ddb6 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ParserUtilsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ParserUtilsSuite.scala
@@ -53,6 +53,24 @@ class ParserUtilsSuite extends SparkFunSuite {
parser.statement().asInstanceOf[CreateNamespaceContext]
}
+ val castClause =
+ """
+ |CAST(1 /* Convert
+ | INT
+ | AS
+ | String */ as STRING)""".stripMargin.trim
+
+ val castQuery =
+ s"""
+ |SELECT
+ |$castClause /* SHOULD NOT INCLUDE THIS */
+ | AS s
+ |""".stripMargin
+
+ val castQueryContext = buildContext(castQuery) { parser =>
+ parser.statement().asInstanceOf[StatementDefaultContext]
+ }
+
val emptyContext = buildContext("") { parser =>
parser.statement
}
@@ -199,4 +217,42 @@ class ParserUtilsSuite extends SparkFunSuite {
assert(origin == Origin(Some(3), Some(27)))
assert(CurrentOrigin.get == current)
}
+
+ private def findCastContext(ctx: ParserRuleContext): Option[CastContext] = {
+ ctx match {
+ case context: CastContext =>
+ Some(context)
+ case _ =>
+ val it = ctx.children.iterator()
+ while(it.hasNext) {
+ it.next() match {
+ case p: ParserRuleContext =>
+ val childResult = findCastContext(p)
+ if (childResult.isDefined) {
+ return childResult
+ }
+ case _ =>
+ }
+ }
+ None
+ }
+ }
+
+ test("withOrigin: setting SQL text") {
+ withOrigin(castQueryContext, Some(castQuery)) {
+ assert(CurrentOrigin.get.sqlText.contains(castQuery))
+ val castContext = findCastContext(castQueryContext)
+ assert(castContext.isDefined)
+ withOrigin(castContext.get) {
+ val current = CurrentOrigin.get
+ assert(current.sqlText.contains(castQuery))
+ assert(current.startIndex.isDefined)
+ assert(current.stopIndex.isDefined)
+ // With sqlText, startIndex, stopIndex, we can get the corresponding SQL text of the
+ // Cast clause.
+ assert(current.sqlText.get.substring(current.startIndex.get, current.stopIndex.get + 1) ==
+ castClause)
+ }
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala
index 68eb15b..77513c5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala
@@ -20,7 +20,10 @@ package org.apache.spark.sql.execution
import org.apache.spark.SparkException
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.expressions.{Add, Alias, Divide}
import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.trees.Origin
import org.apache.spark.sql.internal.SQLConf._
import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}
@@ -909,6 +912,58 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
}
}
+ test("CurrentOrigin is correctly set in and out of the View") {
+ withTable("t") {
+ Seq((1, 1), (2, 2)).toDF("a", "b").write.format("parquet").saveAsTable("t")
+ Seq("VIEW", "TEMPORARY VIEW").foreach { viewType =>
+ val viewId = "v"
+ withView(viewId) {
+ val viewText = "SELECT a + b c FROM t"
+ sql(
+ s"""
+ |CREATE $viewType $viewId AS
+ |-- the body of the view
+ |$viewText
+ |""".stripMargin)
+ val plan = sql("select c / 2.0D d from v").logicalPlan
+ val add = plan.collectFirst {
+ case Project(Seq(Alias(a: Add, _)), _) => a
+ }
+ assert(add.isDefined)
+ val qualifiedName = if (viewType == "VIEW") {
+ s"default.$viewId"
+ } else {
+ viewId
+ }
+ val expectedAddOrigin = Origin(
+ line = Some(1),
+ startPosition = Some(7),
+ startIndex = Some(7),
+ stopIndex = Some(11),
+ sqlText = Some("SELECT a + b c FROM t"),
+ objectType = Some("VIEW"),
+ objectName = Some(qualifiedName)
+ )
+ assert(add.get.origin == expectedAddOrigin)
+
+ val divide = plan.collectFirst {
+ case Project(Seq(Alias(d: Divide, _)), _) => d
+ }
+ assert(divide.isDefined)
+ val expectedDivideOrigin = Origin(
+ line = Some(1),
+ startPosition = Some(7),
+ startIndex = Some(7),
+ stopIndex = Some(14),
+ sqlText = Some("select c / 2.0D d from v"),
+ objectType = None,
+ objectName = None)
+ assert(divide.get.origin == expectedDivideOrigin)
+ }
+ }
+ }
+ }
+
test("SPARK-37932: view join with same view") {
withTable("t") {
withView("v1") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org