You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "srielau (via GitHub)" <gi...@apache.org> on 2026/08/12 12:31:14 UTC
[PR] [SPARK-58738][SQL] Add parse_command function for SQL statement JSON analysis [spark]
srielau opened a new pull request, #57962:
URL: https://github.com/apache/spark/pull/57962
### What changes were proposed in this pull request?
Add a built-in scalar SQL function `parse_command(sqlStmt)` that parse-only analyzes a SQL statement string and returns a compact JSON description:
- `parse_success` and, on failure, a nested STANDARD-format error object (`errorClass`, `messageTemplate`, `sqlState`, `messageParameters`) so batch evaluation never fails a query on bad SQL
- Statement classification from ISO/IEC 9075-2:2023 Table 39 (`statement_identifier`, `statement_code`, `statement_type`, `statement_class`); Spark-only statements use product-specific identifiers with negative implementation-defined codes
- Multipart `table_references` and `function_references` (identifier part sequences)
- `select_list` entries with name parts and expression text
- `parameter_markers` (`named` and `unnamed_count`)
- `as_subquery` for `CREATE TABLE ... AS SELECT` (still Table 39 `CREATE TABLE` / code 77)
### Why are the changes needed?
Dialect-migration and SQL analysis tooling need an efficient, batchable way to verify that translated SQL parses and to extract statement kind / references without requiring catalog objects to exist.
### Does this PR introduce _any_ user-facing change?
Yes. New SQL function:
```sql
SELECT parse_command('SELECT a FROM t');
-- {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,...}
SELECT parse_command('SELEC');
-- {"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR",...}}
```
### How was this patch tested?
Added `ParseCommandResultSuite` and `ParseCommandSuite` (catalyst). Ran:
```
build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.parser.ParseCommandResultSuite org.apache.spark.sql.catalyst.expressions.ParseCommandSuite'
```
All 13 tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.5
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3775129900
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
Review Comment:
Addressed: moved `ParseSql` / `ParseSqlResult` / `SqlStatementCodes` to `sql/core` and parse with a ThreadLocal `SparkSqlParser` (same surface as the default session parser). Registered via `BaseSessionStateBuilder`. Added goldens/unit coverage for EXPLAIN / SET / ADD JAR / CREATE VIEW.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
Review Comment:
Addressed (with a deliberate choice on versioning): keeping the JSON string return type without a `format_version` field — new fields can be added additively. Dropped `select_list[].expression` (`Expression.sql`) entirely so we do not freeze the pretty-printer as API; `select_list` is name parts only.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.apache.spark.sql.catalyst.analysis.UnresolvedExecuteImmediate
+import org.apache.spark.sql.catalyst.plans.logical._
+
+/**
+ * Classification of a parsed SQL statement using ISO/IEC 9075-2:2023 Table 39,
+ * "SQL-statement codes" (clause 23.1 <get diagnostics statement>).
+ *
+ * @param statementIdentifier Table 39 Identifier column (or Spark product name)
+ * @param statementCode Table 39 Code column; Spark-only statements use negative
+ * implementation-defined codes (Table 39 IE005 / IV190)
+ */
+case class SqlStatementClassification(
+ statementIdentifier: String,
+ statementCode: Int)
+
+/**
+ * Maps unresolved [[LogicalPlan]]s to Table 39 statement codes.
+ *
+ * Spark-only statements use the standard's implementation-defined escape hatch:
+ * a product-specific identifier and a distinct negative code. Codes are
+ * append-only and must never be renumbered.
+ */
+object SqlStatementCodes {
+
+ // Standard Table 39 entries used by Spark SQL (ISO/IEC 9075-2:2023).
+ val Select: SqlStatementClassification = SqlStatementClassification("SELECT", 21)
+ val Insert: SqlStatementClassification = SqlStatementClassification("INSERT", 50)
+ val DeleteWhere: SqlStatementClassification = SqlStatementClassification("DELETE WHERE", 19)
+ val UpdateWhere: SqlStatementClassification = SqlStatementClassification("UPDATE WHERE", 82)
+ val Merge: SqlStatementClassification = SqlStatementClassification("MERGE", 128)
+ val CreateTable: SqlStatementClassification = SqlStatementClassification("CREATE TABLE", 77)
+ val CreateView: SqlStatementClassification = SqlStatementClassification("CREATE VIEW", 84)
+ val DropTable: SqlStatementClassification = SqlStatementClassification("DROP TABLE", 32)
+ val DropView: SqlStatementClassification = SqlStatementClassification("DROP VIEW", 36)
+ val AlterTable: SqlStatementClassification = SqlStatementClassification("ALTER TABLE", 4)
+ val CreateSchema: SqlStatementClassification = SqlStatementClassification("CREATE SCHEMA", 64)
+ val DropSchema: SqlStatementClassification = SqlStatementClassification("DROP SCHEMA", 31)
+ val SetSchema: SqlStatementClassification = SqlStatementClassification("SET SCHEMA", 74)
+ val TruncateTable: SqlStatementClassification =
+ SqlStatementClassification("TRUNCATE TABLE", 139)
+ val CreateRoutine: SqlStatementClassification = SqlStatementClassification("CREATE ROUTINE", 14)
+ val DropRoutine: SqlStatementClassification = SqlStatementClassification("DROP ROUTINE", 30)
+ val ExecuteImmediate: SqlStatementClassification =
+ SqlStatementClassification("EXECUTE IMMEDIATE", 43)
+ val Call: SqlStatementClassification = SqlStatementClassification("CALL", 7)
+
+ // Table 39 "Unrecognized statements": empty identifier, code 0.
+ val Unrecognized: SqlStatementClassification = SqlStatementClassification("", 0)
+
+ // Spark product-specific identifiers with append-only negative codes
+ // (Table 39 implementation-defined / IE005 row: negative Code values).
+ val CacheTable: SqlStatementClassification = spark("CACHE TABLE", -1)
+ val CacheTableAsSelect: SqlStatementClassification = spark("CACHE TABLE AS SELECT", -2)
+ val UncacheTable: SqlStatementClassification = spark("UNCACHE TABLE", -3)
+ val RefreshTable: SqlStatementClassification = spark("REFRESH TABLE", -4)
+ val ShowTables: SqlStatementClassification = spark("SHOW TABLES", -5)
+ val DescribeTable: SqlStatementClassification = spark("DESCRIBE TABLE", -6)
+ val AnalyzeTable: SqlStatementClassification = spark("ANALYZE TABLE", -7)
+ val DeclareVariable: SqlStatementClassification = spark("DECLARE VARIABLE", -8)
+ val SetVariable: SqlStatementClassification = spark("SET VARIABLE", -9)
+ val DropVariable: SqlStatementClassification = spark("DROP VARIABLE", -10)
+ val ShowTableProperties: SqlStatementClassification = spark("SHOW TBLPROPERTIES", -11)
+ val DescribeNamespace: SqlStatementClassification = spark("DESCRIBE NAMESPACE", -12)
+ val ShowFunctions: SqlStatementClassification = spark("SHOW FUNCTIONS", -13)
+ val DescribeFunction: SqlStatementClassification = spark("DESCRIBE FUNCTION", -14)
+ val ShowCreateTable: SqlStatementClassification = spark("SHOW CREATE TABLE", -15)
+ val ShowColumns: SqlStatementClassification = spark("SHOW COLUMNS", -16)
+ val ShowPartitions: SqlStatementClassification = spark("SHOW PARTITIONS", -17)
+ val ShowViews: SqlStatementClassification = spark("SHOW VIEWS", -18)
+ val RefreshFunction: SqlStatementClassification = spark("REFRESH FUNCTION", -19)
+ val CommentOnNamespace: SqlStatementClassification = spark("COMMENT ON NAMESPACE", -20)
+ val CommentOnTable: SqlStatementClassification = spark("COMMENT ON TABLE", -21)
+ // SQL/PSM-style scripting (9075-4); not in Foundation Table 39.
+ val BeginEnd: SqlStatementClassification = spark("BEGIN END", -22)
+
+ private def spark(identifier: String, code: Int): SqlStatementClassification = {
+ assert(code < 0, s"Spark statement codes must be negative, got $code")
+ SqlStatementClassification(statementIdentifier = identifier, statementCode = code)
+ }
+
+ /** Classify an unresolved logical plan. */
+ def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
+ case UnresolvedWith(child, _, _) => classify(child)
+ case _: CompoundBody => BeginEnd
+ case _: InsertIntoStatement => Insert
+ case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
+ case _: UpdateTable => UpdateWhere
+ case _: MergeIntoTable => Merge
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect => CreateTable
+ case _: CreateTable | _: CreateTableLike | _: ReplaceTable => CreateTable
+ case _: CreateView => CreateView
+ case _: DropTable => DropTable
+ case _: DropView => DropView
+ case _: CreateNamespace => CreateSchema
+ case _: DropNamespace => DropSchema
+ case _: SetCatalogAndNamespace => SetSchema
+ case _: TruncateTable => TruncateTable
+ case _: CreateFunction => CreateRoutine
+ case _: DropFunction => DropRoutine
+ case _: UnresolvedExecuteImmediate => ExecuteImmediate
+ case _: Call => Call
+ case _: CommentOnTable => CommentOnTable
+ case _: AlterTableCommand | _: RenameTable => AlterTable
+ case _: CacheTable => CacheTable
+ case _: CacheTableAsSelect => CacheTableAsSelect
+ case _: UncacheTable => UncacheTable
+ case _: RefreshTable => RefreshTable
+ case _: ShowTables | _: ShowTablesExtended => ShowTables
+ case _: DescribeRelation | _: DescribeTablePartition | _: DescribeColumn =>
+ DescribeTable
+ case _: AnalyzeTable | _: AnalyzeTables | _: AnalyzeColumn => AnalyzeTable
+ case _: CreateVariable => DeclareVariable
+ case _: SetVariable => SetVariable
+ case _: DropVariable => DropVariable
+ case _: ShowTableProperties => ShowTableProperties
+ case _: DescribeNamespace => DescribeNamespace
+ case _: ShowFunctions => ShowFunctions
+ case _: DescribeFunction => DescribeFunction
+ case _: ShowCreateTable => ShowCreateTable
+ case _: ShowColumns => ShowColumns
+ case _: ShowPartitions | _: ShowTablePartition => ShowPartitions
+ case _: ShowViews => ShowViews
+ case _: RefreshFunction => RefreshFunction
+ case _: CommentOnNamespace => CommentOnNamespace
+ case _: Command => Unrecognized
+ case _ => Select
+ }
Review Comment:
Addressed: filled known SparkSqlParser gaps with append-only negative codes (-23…-36 for EXPLAIN/SET/ADD JAR/…). Query shapes are allowlisted as SELECT; unknown plans (including unknown Commands) map to Unrecognized (code 0). No more `_ => Select`. PR description synced.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ locationFields)
+ )))
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Collect multipart table/view identifiers as written in the SQL.
+ * Deduplicates while preserving first-seen order.
+ */
+ def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty) seen += parts
+ }
+ foreachPlanDeep(plan) {
+ case u: UnresolvedRelation => add(u.multipartIdentifier)
+ case u: UnresolvedTable => add(u.multipartIdentifier)
+ case u: UnresolvedView => add(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => add(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => add(u.nameParts)
+ case _ =>
+ }
+ seen.toSeq
+ }
+
+ /** Collect multipart function names, including table-valued functions. */
+ def collectFunctionReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty) seen += parts
+ }
+ def collectInExpression(e: Expression): Unit = e.foreach {
+ case f: UnresolvedFunction => add(f.nameParts)
+ case _ =>
+ }
+ foreachPlanDeep(plan) { p =>
+ foreachExpressionDeep(p)(collectInExpression)
+ p match {
+ case u: UnresolvedTableValuedFunction => add(u.name)
+ case _ =>
+ }
+ }
+ seen.toSeq
+ }
+
+ /**
+ * Collect the primary select list as `{name, expression}` objects.
+ * Empty for non-query statements without a projected query body.
+ */
+ def collectSelectList(plan: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(plan)
+ val named: Seq[NamedExpression] = query match {
+ case p: Project => p.projectList
+ case a: Aggregate => a.aggregateExpressions
+ case _ => Nil
+ }
+ named.map(selectListItem)
+ }
+
+ private def primaryQueryPlan(plan: LogicalPlan): LogicalPlan = plan match {
+ case UnresolvedWith(child, _, _) => primaryQueryPlan(child)
+ case InsertIntoStatement(_, _, _, query, _, _, _, _, _) =>
+ primaryQueryPlan(query)
+ case c: CreateTableAsSelect => primaryQueryPlan(c.query)
+ case r: ReplaceTableAsSelect => primaryQueryPlan(r.query)
+ case SubqueryAlias(_, child) => primaryQueryPlan(child)
+ case other => other
+ }
Review Comment:
Addressed: `primaryQueryPlan` unwraps `CreateView` / `CreateViewCommand` / `CacheTableAsSelect` (and EXPLAIN / DESCRIBE QUERY). `table_references` is lineage-oriented — CTE definition names and correlation aliases are omitted; tables inside CTE bodies are still collected. Added CREATE VIEW coverage.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseCommand.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseCommandResult
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, references, select list, parameters),
+ * or a STANDARD-format error object when the statement does not parse.
+ *
+ * Designed for batch evaluation over DataFrames of SQL text; never throws on
+ * syntax errors so a single bad row does not fail the query.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` and returns a JSON string describing the
+ statement (parse success, Table 39 statement identifier/code, table and function
+ references, select-list columns, and parameter markers). On syntax error returns
+ JSON with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
+ arguments = """
+ Arguments:
+ * sqlStmt - A SQL statement string to parse.
+ An expression that evaluates to a string.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_('SELECT a, b FROM t');
+ {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"table_references":[["t"]],"function_references":[],"select_list":[{"name":["a"],"expression":"a"},{"name":["b"],"expression":"b"}],"parameter_markers":{"named":[],"unnamed_count":0}}
+ > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+ PARSE_SYNTAX_ERROR
+ """,
+ group = "misc_funcs",
+ since = "4.3.0")
+// scalastyle:on line.size.limit
+case class ParseCommand(child: Expression)
+ extends UnaryExpression
+ with ImplicitCastInputTypes
+ with CodegenFallback {
+
+ override def prettyName: String = "parse_command"
Review Comment:
Addressed: renamed to `parse_sql`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseCommand.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseCommandResult
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, references, select list, parameters),
+ * or a STANDARD-format error object when the statement does not parse.
+ *
+ * Designed for batch evaluation over DataFrames of SQL text; never throws on
+ * syntax errors so a single bad row does not fail the query.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` and returns a JSON string describing the
+ statement (parse success, Table 39 statement identifier/code, table and function
+ references, select-list columns, and parameter markers). On syntax error returns
+ JSON with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
+ arguments = """
+ Arguments:
+ * sqlStmt - A SQL statement string to parse.
+ An expression that evaluates to a string.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_('SELECT a, b FROM t');
+ {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"table_references":[["t"]],"function_references":[],"select_list":[{"name":["a"],"expression":"a"},{"name":["b"],"expression":"b"}],"parameter_markers":{"named":[],"unnamed_count":0}}
+ > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+ PARSE_SYNTAX_ERROR
+ """,
+ group = "misc_funcs",
+ since = "4.3.0")
Review Comment:
Addressed: this change is master-only, so `@since` is now `5.0.0`.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3775833756
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedExecuteImmediate, UnresolvedHaving}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.execution.command._
+import org.apache.spark.sql.execution.datasources.{CreateTempViewUsing, RefreshResource}
+
+/**
+ * Classification of a parsed SQL statement using ISO/IEC 9075-2:2023 Table 39,
+ * "SQL-statement codes" (clause 23.1 <get diagnostics statement>).
+ *
+ * @param statementIdentifier Table 39 Identifier column (or Spark product name)
+ * @param statementCode Table 39 Code column; Spark-only statements use negative
+ * implementation-defined codes (Table 39 IE005 / IV190)
+ */
+case class SqlStatementClassification(
+ statementIdentifier: String,
+ statementCode: Int)
+
+/**
+ * Maps unresolved [[LogicalPlan]]s to Table 39 statement codes.
+ *
+ * Spark-only statements use the standard's implementation-defined escape hatch:
+ * a product-specific identifier and a distinct negative code. Codes are
+ * append-only and must never be renumbered.
+ *
+ * Unknown plans map to [[Unrecognized]] (empty identifier, code 0). Query
+ * shapes are allowlisted; unknown non-commands are not assumed to be SELECT.
+ */
+object SqlStatementCodes {
+
+ // Standard Table 39 entries used by Spark SQL (ISO/IEC 9075-2:2023).
+ val Select: SqlStatementClassification = SqlStatementClassification("SELECT", 21)
+ val Insert: SqlStatementClassification = SqlStatementClassification("INSERT", 50)
+ val DeleteWhere: SqlStatementClassification = SqlStatementClassification("DELETE WHERE", 19)
+ val UpdateWhere: SqlStatementClassification = SqlStatementClassification("UPDATE WHERE", 82)
+ val Merge: SqlStatementClassification = SqlStatementClassification("MERGE", 128)
+ val CreateTable: SqlStatementClassification = SqlStatementClassification("CREATE TABLE", 77)
+ val CreateView: SqlStatementClassification = SqlStatementClassification("CREATE VIEW", 84)
+ val DropTable: SqlStatementClassification = SqlStatementClassification("DROP TABLE", 32)
+ val DropView: SqlStatementClassification = SqlStatementClassification("DROP VIEW", 36)
+ val AlterTable: SqlStatementClassification = SqlStatementClassification("ALTER TABLE", 4)
+ val CreateSchema: SqlStatementClassification = SqlStatementClassification("CREATE SCHEMA", 64)
+ val DropSchema: SqlStatementClassification = SqlStatementClassification("DROP SCHEMA", 31)
+ val SetSchema: SqlStatementClassification = SqlStatementClassification("SET SCHEMA", 74)
+ val TruncateTable: SqlStatementClassification =
+ SqlStatementClassification("TRUNCATE TABLE", 139)
+ val CreateRoutine: SqlStatementClassification = SqlStatementClassification("CREATE ROUTINE", 14)
+ val DropRoutine: SqlStatementClassification = SqlStatementClassification("DROP ROUTINE", 30)
+ val ExecuteImmediate: SqlStatementClassification =
+ SqlStatementClassification("EXECUTE IMMEDIATE", 43)
+ val Call: SqlStatementClassification = SqlStatementClassification("CALL", 7)
+
+ // Table 39 "Unrecognized statements": empty identifier, code 0.
+ val Unrecognized: SqlStatementClassification = SqlStatementClassification("", 0)
+
+ // Spark product-specific identifiers with append-only negative codes
+ // (Table 39 implementation-defined / IE005 row: negative Code values).
+ val CacheTable: SqlStatementClassification = spark("CACHE TABLE", -1)
+ val CacheTableAsSelect: SqlStatementClassification = spark("CACHE TABLE AS SELECT", -2)
+ val UncacheTable: SqlStatementClassification = spark("UNCACHE TABLE", -3)
+ val RefreshTable: SqlStatementClassification = spark("REFRESH TABLE", -4)
+ val ShowTables: SqlStatementClassification = spark("SHOW TABLES", -5)
+ val DescribeTable: SqlStatementClassification = spark("DESCRIBE TABLE", -6)
+ val AnalyzeTable: SqlStatementClassification = spark("ANALYZE TABLE", -7)
+ val DeclareVariable: SqlStatementClassification = spark("DECLARE VARIABLE", -8)
+ val SetVariable: SqlStatementClassification = spark("SET VARIABLE", -9)
+ val DropVariable: SqlStatementClassification = spark("DROP VARIABLE", -10)
+ val ShowTableProperties: SqlStatementClassification = spark("SHOW TBLPROPERTIES", -11)
+ val DescribeNamespace: SqlStatementClassification = spark("DESCRIBE NAMESPACE", -12)
+ val ShowFunctions: SqlStatementClassification = spark("SHOW FUNCTIONS", -13)
+ val DescribeFunction: SqlStatementClassification = spark("DESCRIBE FUNCTION", -14)
+ val ShowCreateTable: SqlStatementClassification = spark("SHOW CREATE TABLE", -15)
+ val ShowColumns: SqlStatementClassification = spark("SHOW COLUMNS", -16)
+ val ShowPartitions: SqlStatementClassification = spark("SHOW PARTITIONS", -17)
+ val ShowViews: SqlStatementClassification = spark("SHOW VIEWS", -18)
+ val RefreshFunction: SqlStatementClassification = spark("REFRESH FUNCTION", -19)
+ val CommentOnNamespace: SqlStatementClassification = spark("COMMENT ON NAMESPACE", -20)
+ val CommentOnTable: SqlStatementClassification = spark("COMMENT ON TABLE", -21)
+ // SQL/PSM-style scripting (9075-4); not in Foundation Table 39.
+ val BeginEnd: SqlStatementClassification = spark("BEGIN END", -22)
+ // SparkSqlParser-only session / resource commands (append-only).
+ val Explain: SqlStatementClassification = spark("EXPLAIN", -23)
+ val Set: SqlStatementClassification = spark("SET", -24)
+ val Reset: SqlStatementClassification = spark("RESET", -25)
+ val AddJar: SqlStatementClassification = spark("ADD JAR", -26)
+ val AddFile: SqlStatementClassification = spark("ADD FILE", -27)
+ val AddArchive: SqlStatementClassification = spark("ADD ARCHIVE", -28)
+ val ListJar: SqlStatementClassification = spark("LIST JAR", -29)
+ val ListFile: SqlStatementClassification = spark("LIST FILE", -30)
+ val ClearCache: SqlStatementClassification = spark("CLEAR CACHE", -31)
+ val RefreshResourceCmd: SqlStatementClassification = spark("REFRESH RESOURCE", -32)
+ val DescribeQuery: SqlStatementClassification = spark("DESCRIBE QUERY", -33)
+ val ShowCatalogs: SqlStatementClassification = spark("SHOW CATALOGS", -34)
+ val ShowCurrentNamespace: SqlStatementClassification =
+ spark("SHOW CURRENT NAMESPACE", -35)
+ val SetCatalog: SqlStatementClassification = spark("SET CATALOG", -36)
+
+ private def spark(identifier: String, code: Int): SqlStatementClassification = {
+ assert(code < 0, s"Spark statement codes must be negative, got $code")
+ SqlStatementClassification(statementIdentifier = identifier, statementCode = code)
+ }
+
+ /** Classify an unresolved logical plan. */
+ def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
+ case UnresolvedWith(child, _, _) => classify(child)
+ case _: CompoundBody => BeginEnd
+ case _: InsertIntoStatement => Insert
+ case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
+ case _: UpdateTable => UpdateWhere
+ case _: MergeIntoTable => Merge
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect => CreateTable
+ case _: CreateTable | _: CreateTableLike | _: ReplaceTable => CreateTable
+ case _: CreateView | _: CreateViewCommand | _: CreateTempViewUsing => CreateView
+ case _: DropTable => DropTable
+ case _: DropView => DropView
+ case _: CreateNamespace => CreateSchema
+ case _: DropNamespace => DropSchema
+ case _: SetCatalogAndNamespace | _: SetNamespaceCommand => SetSchema
+ case _: SetCatalogCommand => SetCatalog
+ case _: TruncateTable => TruncateTable
+ case _: CreateFunction | _: CreateFunctionCommand |
+ _: CreateUserDefinedFunction | _: CreateUserDefinedFunctionCommand =>
+ CreateRoutine
+ case _: DropFunction | _: DropFunctionCommand => DropRoutine
+ case _: UnresolvedExecuteImmediate => ExecuteImmediate
+ case _: Call => Call
+ case _: CommentOnTable => CommentOnTable
+ case _: AlterTableCommand | _: RenameTable => AlterTable
+ case _: CacheTable => CacheTable
+ case _: CacheTableAsSelect => CacheTableAsSelect
+ case _: UncacheTable => UncacheTable
+ case _: RefreshTable => RefreshTable
+ case _: ShowTables | _: ShowTablesExtended => ShowTables
+ case _: DescribeRelation | _: DescribeTablePartition | _: DescribeColumn =>
+ DescribeTable
+ case _: DescribeQueryCommand => DescribeQuery
+ case _: AnalyzeTable | _: AnalyzeTables | _: AnalyzeColumn => AnalyzeTable
+ case _: CreateVariable => DeclareVariable
+ case _: SetVariable => SetVariable
+ case _: DropVariable => DropVariable
+ case _: ShowTableProperties => ShowTableProperties
+ case _: DescribeNamespace => DescribeNamespace
+ case _: ShowFunctions => ShowFunctions
+ case _: DescribeFunction => DescribeFunction
+ case _: ShowCreateTable => ShowCreateTable
+ case _: ShowColumns => ShowColumns
+ case _: ShowPartitions | _: ShowTablePartition => ShowPartitions
+ case _: ShowViews => ShowViews
+ case _: RefreshFunction => RefreshFunction
+ case _: CommentOnNamespace => CommentOnNamespace
+ case _: ExplainCommand => Explain
+ case _: SetCommand => Set
+ case _: ResetCommand => Reset
+ case _: AddJarsCommand => AddJar
+ case _: AddFilesCommand => AddFile
+ case _: AddArchivesCommand => AddArchive
+ case _: ListJarsCommand => ListJar
+ case _: ListFilesCommand => ListFile
+ case ClearCacheCommand => ClearCache
+ case _: RefreshResource => RefreshResourceCmd
+ case _: ShowCatalogsCommand => ShowCatalogs
+ case _: ShowCurrentNamespaceCommand => ShowCurrentNamespace
+ case _: Command => Unrecognized
+ case p if isQueryPlan(p) => Select
+ case _ => Unrecognized
Review Comment:
Addressed: added Spark code `-37` (`CREATE METRIC VIEW`) for `CreateMetricView` / `CreateMetricViewCommand`.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57962:
URL: https://github.com/apache/spark/pull/57962#issuecomment-5351008617
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/ebb49417ecdf34ac59c54023309c18af4155bf3c
- merged into branch-4.x https://github.com/apache/spark/commit/ba13ea68e51cd217afa3bc8d50c9932582232ae5
*Posted by `merge_spark_pr.py`*
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #57962: [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis
URL: https://github.com/apache/spark/pull/57962
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3775131003
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
Review Comment:
Addressed: removed the `NonFatal` → LEGACY catch-all. User-facing parse/scripting failures (`ParseException`, `SqlScriptingException`, other `SparkThrowable`) return error JSON; unexpected/internal failures and non-recoverable errors (OOM/SOE) propagate and fail the function.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResultSuite.scala:
##########
@@ -0,0 +1,446 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.SparkFunSuite
+
+class ParseCommandResultSuite extends SparkFunSuite {
Review Comment:
Addressed: behavioral coverage lives in `sql-tests/inputs/parse-sql.sql`. Unit suite keeps Table 39 pin tests plus a few parser-surface / lineage contracts (CREATE VIEW select_list, CTE exclusion, SparkSqlParser-only codes).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ locationFields)
+ )))
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Collect multipart table/view identifiers as written in the SQL.
+ * Deduplicates while preserving first-seen order.
+ */
+ def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
Review Comment:
Addressed: `collectTableReferences` / `collectFunctionReferences` / `collectSelectList` are now private; coverage is via `fromSql` and goldens.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseCommand.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseCommandResult
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, references, select list, parameters),
+ * or a STANDARD-format error object when the statement does not parse.
+ *
+ * Designed for batch evaluation over DataFrames of SQL text; never throws on
+ * syntax errors so a single bad row does not fail the query.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` and returns a JSON string describing the
+ statement (parse success, Table 39 statement identifier/code, table and function
+ references, select-list columns, and parameter markers). On syntax error returns
+ JSON with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
Review Comment:
Addressed: PR title/description and `@ExpressionDescription` usage/examples updated to the slim `parse_sql` contract (no `statement_type` / `statement_class` / `as_subquery` / expression text).
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_command function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3774376964
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
Review Comment:
**[Critical] Unversioned JSON + `Expression.sql` as public contract**
Shipping analysis results as an unversioned JSON string with `expression` = `Expression.sql` freezes an unstable pretty-printer as API. Everything exposed here is exploitable by scripts.
Can we use a versioned struct (or at least `format_version`), and treat expression text as non-contractual unless we explicitly commit to stability?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ locationFields)
+ )))
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Collect multipart table/view identifiers as written in the SQL.
+ * Deduplicates while preserving first-seen order.
+ */
+ def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty) seen += parts
+ }
+ foreachPlanDeep(plan) {
+ case u: UnresolvedRelation => add(u.multipartIdentifier)
+ case u: UnresolvedTable => add(u.multipartIdentifier)
+ case u: UnresolvedView => add(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => add(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => add(u.nameParts)
+ case _ =>
+ }
+ seen.toSeq
+ }
+
+ /** Collect multipart function names, including table-valued functions. */
+ def collectFunctionReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty) seen += parts
+ }
+ def collectInExpression(e: Expression): Unit = e.foreach {
+ case f: UnresolvedFunction => add(f.nameParts)
+ case _ =>
+ }
+ foreachPlanDeep(plan) { p =>
+ foreachExpressionDeep(p)(collectInExpression)
+ p match {
+ case u: UnresolvedTableValuedFunction => add(u.name)
+ case _ =>
+ }
+ }
+ seen.toSeq
+ }
+
+ /**
+ * Collect the primary select list as `{name, expression}` objects.
+ * Empty for non-query statements without a projected query body.
+ */
+ def collectSelectList(plan: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(plan)
+ val named: Seq[NamedExpression] = query match {
+ case p: Project => p.projectList
+ case a: Aggregate => a.aggregateExpressions
+ case _ => Nil
+ }
+ named.map(selectListItem)
+ }
+
+ private def primaryQueryPlan(plan: LogicalPlan): LogicalPlan = plan match {
+ case UnresolvedWith(child, _, _) => primaryQueryPlan(child)
+ case InsertIntoStatement(_, _, _, query, _, _, _, _, _) =>
+ primaryQueryPlan(query)
+ case c: CreateTableAsSelect => primaryQueryPlan(c.query)
+ case r: ReplaceTableAsSelect => primaryQueryPlan(r.query)
+ case SubqueryAlias(_, child) => primaryQueryPlan(child)
+ case other => other
+ }
Review Comment:
**[High] `primaryQueryPlan` / deep-walk gaps**
`primaryQueryPlan` doesn’t unwrap `CreateView` / `CacheTableAsSelect`, so `select_list` is empty for `CREATE VIEW v AS SELECT a, b FROM t` even though table refs are collected. `foreachPlanDeep`’s special cases will bitrot.
Can we cover view/CTAS-like wrappers in `primaryQueryPlan`, add goldens, and clarify whether CTE names belong in `table_references`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
Review Comment:
**[Critical] Session parser divergence**
`ParseCommandResult` uses a ThreadLocal `CatalystSqlParser`, but sessions parse with `SparkSqlParser`/`SparkSqlAstBuilder` (`EXPLAIN`/`SET`/`ADD JAR`/`USING` DDL/variable substitution, plus parser extensions). That makes `parse_command` disagree with `spark.sql` for a large statement class.
Can we parse with the same parser surface production uses (likely move the expression to `sql/core`), and add goldens that currently fail under Catalyst-only parsing?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.apache.spark.sql.catalyst.analysis.UnresolvedExecuteImmediate
+import org.apache.spark.sql.catalyst.plans.logical._
+
+/**
+ * Classification of a parsed SQL statement using ISO/IEC 9075-2:2023 Table 39,
+ * "SQL-statement codes" (clause 23.1 <get diagnostics statement>).
+ *
+ * @param statementIdentifier Table 39 Identifier column (or Spark product name)
+ * @param statementCode Table 39 Code column; Spark-only statements use negative
+ * implementation-defined codes (Table 39 IE005 / IV190)
+ */
+case class SqlStatementClassification(
+ statementIdentifier: String,
+ statementCode: Int)
+
+/**
+ * Maps unresolved [[LogicalPlan]]s to Table 39 statement codes.
+ *
+ * Spark-only statements use the standard's implementation-defined escape hatch:
+ * a product-specific identifier and a distinct negative code. Codes are
+ * append-only and must never be renumbered.
+ */
+object SqlStatementCodes {
+
+ // Standard Table 39 entries used by Spark SQL (ISO/IEC 9075-2:2023).
+ val Select: SqlStatementClassification = SqlStatementClassification("SELECT", 21)
+ val Insert: SqlStatementClassification = SqlStatementClassification("INSERT", 50)
+ val DeleteWhere: SqlStatementClassification = SqlStatementClassification("DELETE WHERE", 19)
+ val UpdateWhere: SqlStatementClassification = SqlStatementClassification("UPDATE WHERE", 82)
+ val Merge: SqlStatementClassification = SqlStatementClassification("MERGE", 128)
+ val CreateTable: SqlStatementClassification = SqlStatementClassification("CREATE TABLE", 77)
+ val CreateView: SqlStatementClassification = SqlStatementClassification("CREATE VIEW", 84)
+ val DropTable: SqlStatementClassification = SqlStatementClassification("DROP TABLE", 32)
+ val DropView: SqlStatementClassification = SqlStatementClassification("DROP VIEW", 36)
+ val AlterTable: SqlStatementClassification = SqlStatementClassification("ALTER TABLE", 4)
+ val CreateSchema: SqlStatementClassification = SqlStatementClassification("CREATE SCHEMA", 64)
+ val DropSchema: SqlStatementClassification = SqlStatementClassification("DROP SCHEMA", 31)
+ val SetSchema: SqlStatementClassification = SqlStatementClassification("SET SCHEMA", 74)
+ val TruncateTable: SqlStatementClassification =
+ SqlStatementClassification("TRUNCATE TABLE", 139)
+ val CreateRoutine: SqlStatementClassification = SqlStatementClassification("CREATE ROUTINE", 14)
+ val DropRoutine: SqlStatementClassification = SqlStatementClassification("DROP ROUTINE", 30)
+ val ExecuteImmediate: SqlStatementClassification =
+ SqlStatementClassification("EXECUTE IMMEDIATE", 43)
+ val Call: SqlStatementClassification = SqlStatementClassification("CALL", 7)
+
+ // Table 39 "Unrecognized statements": empty identifier, code 0.
+ val Unrecognized: SqlStatementClassification = SqlStatementClassification("", 0)
+
+ // Spark product-specific identifiers with append-only negative codes
+ // (Table 39 implementation-defined / IE005 row: negative Code values).
+ val CacheTable: SqlStatementClassification = spark("CACHE TABLE", -1)
+ val CacheTableAsSelect: SqlStatementClassification = spark("CACHE TABLE AS SELECT", -2)
+ val UncacheTable: SqlStatementClassification = spark("UNCACHE TABLE", -3)
+ val RefreshTable: SqlStatementClassification = spark("REFRESH TABLE", -4)
+ val ShowTables: SqlStatementClassification = spark("SHOW TABLES", -5)
+ val DescribeTable: SqlStatementClassification = spark("DESCRIBE TABLE", -6)
+ val AnalyzeTable: SqlStatementClassification = spark("ANALYZE TABLE", -7)
+ val DeclareVariable: SqlStatementClassification = spark("DECLARE VARIABLE", -8)
+ val SetVariable: SqlStatementClassification = spark("SET VARIABLE", -9)
+ val DropVariable: SqlStatementClassification = spark("DROP VARIABLE", -10)
+ val ShowTableProperties: SqlStatementClassification = spark("SHOW TBLPROPERTIES", -11)
+ val DescribeNamespace: SqlStatementClassification = spark("DESCRIBE NAMESPACE", -12)
+ val ShowFunctions: SqlStatementClassification = spark("SHOW FUNCTIONS", -13)
+ val DescribeFunction: SqlStatementClassification = spark("DESCRIBE FUNCTION", -14)
+ val ShowCreateTable: SqlStatementClassification = spark("SHOW CREATE TABLE", -15)
+ val ShowColumns: SqlStatementClassification = spark("SHOW COLUMNS", -16)
+ val ShowPartitions: SqlStatementClassification = spark("SHOW PARTITIONS", -17)
+ val ShowViews: SqlStatementClassification = spark("SHOW VIEWS", -18)
+ val RefreshFunction: SqlStatementClassification = spark("REFRESH FUNCTION", -19)
+ val CommentOnNamespace: SqlStatementClassification = spark("COMMENT ON NAMESPACE", -20)
+ val CommentOnTable: SqlStatementClassification = spark("COMMENT ON TABLE", -21)
+ // SQL/PSM-style scripting (9075-4); not in Foundation Table 39.
+ val BeginEnd: SqlStatementClassification = spark("BEGIN END", -22)
+
+ private def spark(identifier: String, code: Int): SqlStatementClassification = {
+ assert(code < 0, s"Spark statement codes must be negative, got $code")
+ SqlStatementClassification(statementIdentifier = identifier, statementCode = code)
+ }
+
+ /** Classify an unresolved logical plan. */
+ def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
+ case UnresolvedWith(child, _, _) => classify(child)
+ case _: CompoundBody => BeginEnd
+ case _: InsertIntoStatement => Insert
+ case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
+ case _: UpdateTable => UpdateWhere
+ case _: MergeIntoTable => Merge
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect => CreateTable
+ case _: CreateTable | _: CreateTableLike | _: ReplaceTable => CreateTable
+ case _: CreateView => CreateView
+ case _: DropTable => DropTable
+ case _: DropView => DropView
+ case _: CreateNamespace => CreateSchema
+ case _: DropNamespace => DropSchema
+ case _: SetCatalogAndNamespace => SetSchema
+ case _: TruncateTable => TruncateTable
+ case _: CreateFunction => CreateRoutine
+ case _: DropFunction => DropRoutine
+ case _: UnresolvedExecuteImmediate => ExecuteImmediate
+ case _: Call => Call
+ case _: CommentOnTable => CommentOnTable
+ case _: AlterTableCommand | _: RenameTable => AlterTable
+ case _: CacheTable => CacheTable
+ case _: CacheTableAsSelect => CacheTableAsSelect
+ case _: UncacheTable => UncacheTable
+ case _: RefreshTable => RefreshTable
+ case _: ShowTables | _: ShowTablesExtended => ShowTables
+ case _: DescribeRelation | _: DescribeTablePartition | _: DescribeColumn =>
+ DescribeTable
+ case _: AnalyzeTable | _: AnalyzeTables | _: AnalyzeColumn => AnalyzeTable
+ case _: CreateVariable => DeclareVariable
+ case _: SetVariable => SetVariable
+ case _: DropVariable => DropVariable
+ case _: ShowTableProperties => ShowTableProperties
+ case _: DescribeNamespace => DescribeNamespace
+ case _: ShowFunctions => ShowFunctions
+ case _: DescribeFunction => DescribeFunction
+ case _: ShowCreateTable => ShowCreateTable
+ case _: ShowColumns => ShowColumns
+ case _: ShowPartitions | _: ShowTablePartition => ShowPartitions
+ case _: ShowViews => ShowViews
+ case _: RefreshFunction => RefreshFunction
+ case _: CommentOnNamespace => CommentOnNamespace
+ case _: Command => Unrecognized
+ case _ => Select
+ }
Review Comment:
**[High] Unknown plans should not become SELECT**
`classify` ends with `Command => Unrecognized` and `_ => Select`. Unknown non-commands become SELECT/21 — the opposite of an allowlist.
Can unknown plans default to Unrecognized, and can we either extend the Spark-code table or document that code `0` is expected for many Spark statements? Also the PR body still advertises `statement_type` / `statement_class` / `as_subquery` which are gone.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResultSuite.scala:
##########
@@ -0,0 +1,446 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.SparkFunSuite
+
+class ParseCommandResultSuite extends SparkFunSuite {
Review Comment:
**[Medium] Prefer goldens for the user contract**
A lot of `ParseCommandResultSuite` duplicates the golden file. Can we keep Table 39 pin tests in-unit and move behavioral cases to `parse-command.sql`, including `CREATE VIEW`, and statements that only `SparkSqlParser` accepts?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseCommand.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseCommandResult
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, references, select list, parameters),
+ * or a STANDARD-format error object when the statement does not parse.
+ *
+ * Designed for batch evaluation over DataFrames of SQL text; never throws on
+ * syntax errors so a single bad row does not fail the query.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` and returns a JSON string describing the
+ statement (parse success, Table 39 statement identifier/code, table and function
+ references, select-list columns, and parameter markers). On syntax error returns
+ JSON with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
+ arguments = """
+ Arguments:
+ * sqlStmt - A SQL statement string to parse.
+ An expression that evaluates to a string.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_('SELECT a, b FROM t');
+ {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"table_references":[["t"]],"function_references":[],"select_list":[{"name":["a"],"expression":"a"},{"name":["b"],"expression":"b"}],"parameter_markers":{"named":[],"unnamed_count":0}}
+ > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+ PARSE_SYNTAX_ERROR
+ """,
+ group = "misc_funcs",
+ since = "4.3.0")
Review Comment:
**[High] Wrong `@since` for branch-4.x**
`since = "4.3.0"` doesn’t match `branch-4.x` (`4.4.0-SNAPSHOT`; master is `5.0.0-SNAPSHOT`). For a normal master+branch-4.x backport this should be `4.4.0` (or `5.0.0` if master-only).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ locationFields)
+ )))
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Collect multipart table/view identifiers as written in the SQL.
+ * Deduplicates while preserving first-seen order.
+ */
+ def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
Review Comment:
**[Low] Public `collect_*` helpers**
Are `collectTableReferences` / `collectFunctionReferences` / `collectSelectList` intentionally public? If only tests need them, prefer private + golden/`fromSql` coverage so we don’t freeze helpers as API.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseCommand.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseCommandResult
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, references, select list, parameters),
+ * or a STANDARD-format error object when the statement does not parse.
+ *
+ * Designed for batch evaluation over DataFrames of SQL text; never throws on
+ * syntax errors so a single bad row does not fail the query.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` and returns a JSON string describing the
+ statement (parse success, Table 39 statement identifier/code, table and function
+ references, select-list columns, and parameter markers). On syntax error returns
+ JSON with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
Review Comment:
**[Low] PR description / docs drift**
Please sync the PR description with the actual JSON (drop `statement_type` / `statement_class` / `as_subquery`) so the review contract matches what we’re freezing. Same for any usage text that still implies those fields.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseCommandResult.scala:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list items, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not throw.
+ */
+object ParseCommandResult {
+
+ private val parser: ThreadLocal[CatalystSqlParser] =
+ ThreadLocal.withInitial(() => new CatalystSqlParser())
+
+ /** Parse `sql` and render the JSON result string. Never throws for bad SQL. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ case e: ParseException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ case NonFatal(e) =>
+ // Unexpected failures still must not fail a batch row.
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(
+ "errorClass" -> JString("LEGACY"),
+ "messageParameters" -> JObject(
+ "message" -> JString(Option(e.getMessage).getOrElse(e.toString))
+ )
+ )
+ )))
Review Comment:
**[Medium] Broad NonFatal → LEGACY catch-all**
The `NonFatal` → `errorClass: LEGACY` path makes unexpected failures part of the public JSON and can leak internal messages. Pathological SQL can still DoS a batch (`StackOverflowError` / OOM are not `NonFatal`).
Can we limit catches to known parse/scripting failures, and document that untrusted SQL text can still exhaust task resources?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseCommand.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseCommandResult
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, references, select list, parameters),
+ * or a STANDARD-format error object when the statement does not parse.
+ *
+ * Designed for batch evaluation over DataFrames of SQL text; never throws on
+ * syntax errors so a single bad row does not fail the query.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` and returns a JSON string describing the
+ statement (parse success, Table 39 statement identifier/code, table and function
+ references, select-list columns, and parameter markers). On syntax error returns
+ JSON with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
+ arguments = """
+ Arguments:
+ * sqlStmt - A SQL statement string to parse.
+ An expression that evaluates to a string.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_('SELECT a, b FROM t');
+ {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"table_references":[["t"]],"function_references":[],"select_list":[{"name":["a"],"expression":"a"},{"name":["b"],"expression":"b"}],"parameter_markers":{"named":[],"unnamed_count":0}}
+ > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+ PARSE_SYNTAX_ERROR
+ """,
+ group = "misc_funcs",
+ since = "4.3.0")
+// scalastyle:on line.size.limit
+case class ParseCommand(child: Expression)
+ extends UnaryExpression
+ with ImplicitCastInputTypes
+ with CodegenFallback {
+
+ override def prettyName: String = "parse_command"
Review Comment:
**[High] Misleading name `parse_command`**
`parse_command` will be read as “parse a Catalyst `Command`,” but this covers queries too.
Can we pick a name that matches the contract (`parse_sql` / `sql_statement_info`) before this ships?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57962:
URL: https://github.com/apache/spark/pull/57962#issuecomment-5340401034
The test failure is real
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3798479114
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val refs = collectPlanReferences(plan)
+ if (refs.tables.nonEmpty) {
+ fields += "table_references" -> JArray(refs.tables.map(partsToJArray).toList)
+ }
+ if (refs.functions.nonEmpty) {
+ fields += "function_references" -> JArray(refs.functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ refs.parameterMarkers.foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan root (children expose only nested
+ * children; visit the root without re-walking those children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ visitNonChildSlots(p)(f)
+ }
+ }
+
+ /**
+ * Visit plan slots that are not exposed via `children` / expression
+ * subqueries. For [[SingleStatement]], only the wrapped root (and its
+ * subqueries / non-child slots) is visited -- its children were already
+ * walked by the outer `foreachWithSubqueries` via
+ * `SingleStatement.children`.
+ */
+ private def visitNonChildSlots(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ visitSingleStatementRoot(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case ExplainCommand(logicalPlan, _) =>
+ foreachPlanDeep(logicalPlan)(f)
+ case DescribeQueryCommand(_, queryPlan) =>
+ foreachPlanDeep(queryPlan)(f)
+ case _ =>
+ }
+ }
+
+ /**
+ * Visit a [[SingleStatement]] wrapped root without re-traversing its
+ * children (already exposed by `SingleStatement.children`).
+ */
+ private def visitSingleStatementRoot(root: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ f(root)
+ root.subqueries.foreach(sq => foreachPlanDeep(sq)(f))
+ visitNonChildSlots(root)(f)
+ }
+
+ /** Multipart name from a table/view-shaped plan node, if any. */
+ private def tableOrViewParts(plan: LogicalPlan): Option[Seq[String]] = plan match {
+ case u: UnresolvedRelation => Some(u.multipartIdentifier)
+ case u: UnresolvedTable => Some(u.multipartIdentifier)
+ case u: UnresolvedView => Some(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => Some(u.nameParts)
+ case _ => None
+ }
+
+ private def tableIdentifierParts(id: org.apache.spark.sql.catalyst.TableIdentifier): Seq[String] =
+ id.catalog.toSeq ++ id.database.toSeq :+ id.table
+
+ private final case class PlanReferences(
+ tables: Seq[Seq[String]],
+ functions: Seq[Seq[String]],
+ parameterMarkers: Option[JObject])
+
+ /**
+ * Collect multipart table/view identifiers, function names, and parameter
+ * markers for lineage. CTE definition names and correlation aliases are
+ * omitted from tables; tables referenced inside CTE bodies are still
+ * included. Function / variable identifiers are not collected as tables.
+ * Deduplicates while preserving first-seen order. Uses one CTE-name pass
+ * then one combined deep walk for tables, functions, and parameters.
+ */
+ private def collectPlanReferences(plan: LogicalPlan): PlanReferences = {
+ val tables = mutable.LinkedHashSet.empty[Seq[String]]
+ val functions = mutable.LinkedHashSet.empty[Seq[String]]
+ val namedParams = mutable.LinkedHashSet.empty[String]
+ var unnamedCount = 0
+ // CTE names are always single-part in the grammar; UnresolvedRelation refs
+ // to CTEs are likewise single-part, so filtering matches that shape.
+ val cteNames = mutable.HashSet.empty[String]
Review Comment:
Addressed: the lineage walk now carries the active CTE aliases instead of using a statement-wide set. `UnresolvedWith` adds its aliases only for its own body, and each definition sees only the aliases declared before it (plus its own name when the clause is `RECURSIVE`), matching `CTESubstitution`.
Your example now reports both tables:
```
SELECT parse_sql('SELECT * FROM real_t WHERE EXISTS (WITH real_t AS (SELECT * FROM inner_base) SELECT * FROM real_t)');
-- "table_references":[["inner_base"],["real_t"]]
```
Added both the shadowing case and a forward-reference case (`WITH a AS (SELECT * FROM b), b AS (SELECT 1 AS x) SELECT * FROM a` -> `b` is the real table) to `ParseSqlResultSuite` and `parse-sql.sql`.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3775696642
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ // User-facing parse / scripting failures become JSON; internal errors fail.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ }
Review Comment:
**[High] Broad `SparkThrowable` catch still swallows internals**
The `SparkThrowable` catch-all still turns internal failures (e.g. `SparkException.internalError` from `QueryParsingErrors`) into `parse_success=false` JSON, which conflicts with the stated contract that unexpected failures propagate.
Please keep only `ParseException` / `SqlScriptingException` (or an explicit allowlist) and rethrow the rest.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ // User-facing parse / scripting failures become JSON; internal errors fail.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ locationFields)
+ )))
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case ExplainCommand(logicalPlan, _) =>
+ foreachPlanDeep(logicalPlan)(f)
+ case DescribeQueryCommand(_, queryPlan) =>
+ foreachPlanDeep(queryPlan)(f)
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Collect multipart table/view identifiers for lineage (as written in the
+ * SQL). CTE definition names and correlation aliases are omitted; tables
+ * referenced inside CTE bodies are still included. Deduplicates while
+ * preserving first-seen order.
+ */
+ private def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ val cteNames = mutable.HashSet.empty[String]
+
+ def addCteNames(w: UnresolvedWith): Unit = {
+ w.cteRelations.foreach { case (name, _, _) =>
+ cteNames += name.toLowerCase(java.util.Locale.ROOT)
+ }
+ }
+
+ def isCteName(parts: Seq[String]): Boolean = parts match {
+ case Seq(name) => cteNames.contains(name.toLowerCase(java.util.Locale.ROOT))
+ case _ => false
+ }
+
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty && !isCteName(parts)) seen += parts
+ }
+
+ // First pass: gather CTE names in scope (including nested).
+ foreachPlanDeep(plan) {
+ case w: UnresolvedWith => addCteNames(w)
+ case _ =>
+ }
+
+ foreachPlanDeep(plan) {
+ case u: UnresolvedRelation => add(u.multipartIdentifier)
+ case u: UnresolvedTable => add(u.multipartIdentifier)
+ case u: UnresolvedView => add(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => add(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => add(u.nameParts)
Review Comment:
**[High] `UnresolvedIdentifier` pollutes `table_references`**
Matching `UnresolvedIdentifier` for `table_references` pulls in function/variable identifiers (`CreateFunction` / `CreateVariable` children). Please collect only table/view-shaped nodes (and intentional DDL table/view targets), and add negative tests for `CREATE FUNCTION` / `DECLARE VARIABLE`. CTE filtering is still single-part-only.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedExecuteImmediate, UnresolvedHaving}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.execution.command._
+import org.apache.spark.sql.execution.datasources.{CreateTempViewUsing, RefreshResource}
+
+/**
+ * Classification of a parsed SQL statement using ISO/IEC 9075-2:2023 Table 39,
+ * "SQL-statement codes" (clause 23.1 <get diagnostics statement>).
+ *
+ * @param statementIdentifier Table 39 Identifier column (or Spark product name)
+ * @param statementCode Table 39 Code column; Spark-only statements use negative
+ * implementation-defined codes (Table 39 IE005 / IV190)
+ */
+case class SqlStatementClassification(
+ statementIdentifier: String,
+ statementCode: Int)
+
+/**
+ * Maps unresolved [[LogicalPlan]]s to Table 39 statement codes.
+ *
+ * Spark-only statements use the standard's implementation-defined escape hatch:
+ * a product-specific identifier and a distinct negative code. Codes are
+ * append-only and must never be renumbered.
+ *
+ * Unknown plans map to [[Unrecognized]] (empty identifier, code 0). Query
+ * shapes are allowlisted; unknown non-commands are not assumed to be SELECT.
+ */
+object SqlStatementCodes {
+
+ // Standard Table 39 entries used by Spark SQL (ISO/IEC 9075-2:2023).
+ val Select: SqlStatementClassification = SqlStatementClassification("SELECT", 21)
+ val Insert: SqlStatementClassification = SqlStatementClassification("INSERT", 50)
+ val DeleteWhere: SqlStatementClassification = SqlStatementClassification("DELETE WHERE", 19)
+ val UpdateWhere: SqlStatementClassification = SqlStatementClassification("UPDATE WHERE", 82)
+ val Merge: SqlStatementClassification = SqlStatementClassification("MERGE", 128)
+ val CreateTable: SqlStatementClassification = SqlStatementClassification("CREATE TABLE", 77)
+ val CreateView: SqlStatementClassification = SqlStatementClassification("CREATE VIEW", 84)
+ val DropTable: SqlStatementClassification = SqlStatementClassification("DROP TABLE", 32)
+ val DropView: SqlStatementClassification = SqlStatementClassification("DROP VIEW", 36)
+ val AlterTable: SqlStatementClassification = SqlStatementClassification("ALTER TABLE", 4)
+ val CreateSchema: SqlStatementClassification = SqlStatementClassification("CREATE SCHEMA", 64)
+ val DropSchema: SqlStatementClassification = SqlStatementClassification("DROP SCHEMA", 31)
+ val SetSchema: SqlStatementClassification = SqlStatementClassification("SET SCHEMA", 74)
+ val TruncateTable: SqlStatementClassification =
+ SqlStatementClassification("TRUNCATE TABLE", 139)
+ val CreateRoutine: SqlStatementClassification = SqlStatementClassification("CREATE ROUTINE", 14)
+ val DropRoutine: SqlStatementClassification = SqlStatementClassification("DROP ROUTINE", 30)
+ val ExecuteImmediate: SqlStatementClassification =
+ SqlStatementClassification("EXECUTE IMMEDIATE", 43)
+ val Call: SqlStatementClassification = SqlStatementClassification("CALL", 7)
+
+ // Table 39 "Unrecognized statements": empty identifier, code 0.
+ val Unrecognized: SqlStatementClassification = SqlStatementClassification("", 0)
+
+ // Spark product-specific identifiers with append-only negative codes
+ // (Table 39 implementation-defined / IE005 row: negative Code values).
+ val CacheTable: SqlStatementClassification = spark("CACHE TABLE", -1)
+ val CacheTableAsSelect: SqlStatementClassification = spark("CACHE TABLE AS SELECT", -2)
+ val UncacheTable: SqlStatementClassification = spark("UNCACHE TABLE", -3)
+ val RefreshTable: SqlStatementClassification = spark("REFRESH TABLE", -4)
+ val ShowTables: SqlStatementClassification = spark("SHOW TABLES", -5)
+ val DescribeTable: SqlStatementClassification = spark("DESCRIBE TABLE", -6)
+ val AnalyzeTable: SqlStatementClassification = spark("ANALYZE TABLE", -7)
+ val DeclareVariable: SqlStatementClassification = spark("DECLARE VARIABLE", -8)
+ val SetVariable: SqlStatementClassification = spark("SET VARIABLE", -9)
+ val DropVariable: SqlStatementClassification = spark("DROP VARIABLE", -10)
+ val ShowTableProperties: SqlStatementClassification = spark("SHOW TBLPROPERTIES", -11)
+ val DescribeNamespace: SqlStatementClassification = spark("DESCRIBE NAMESPACE", -12)
+ val ShowFunctions: SqlStatementClassification = spark("SHOW FUNCTIONS", -13)
+ val DescribeFunction: SqlStatementClassification = spark("DESCRIBE FUNCTION", -14)
+ val ShowCreateTable: SqlStatementClassification = spark("SHOW CREATE TABLE", -15)
+ val ShowColumns: SqlStatementClassification = spark("SHOW COLUMNS", -16)
+ val ShowPartitions: SqlStatementClassification = spark("SHOW PARTITIONS", -17)
+ val ShowViews: SqlStatementClassification = spark("SHOW VIEWS", -18)
+ val RefreshFunction: SqlStatementClassification = spark("REFRESH FUNCTION", -19)
+ val CommentOnNamespace: SqlStatementClassification = spark("COMMENT ON NAMESPACE", -20)
+ val CommentOnTable: SqlStatementClassification = spark("COMMENT ON TABLE", -21)
+ // SQL/PSM-style scripting (9075-4); not in Foundation Table 39.
+ val BeginEnd: SqlStatementClassification = spark("BEGIN END", -22)
+ // SparkSqlParser-only session / resource commands (append-only).
+ val Explain: SqlStatementClassification = spark("EXPLAIN", -23)
+ val Set: SqlStatementClassification = spark("SET", -24)
+ val Reset: SqlStatementClassification = spark("RESET", -25)
+ val AddJar: SqlStatementClassification = spark("ADD JAR", -26)
+ val AddFile: SqlStatementClassification = spark("ADD FILE", -27)
+ val AddArchive: SqlStatementClassification = spark("ADD ARCHIVE", -28)
+ val ListJar: SqlStatementClassification = spark("LIST JAR", -29)
+ val ListFile: SqlStatementClassification = spark("LIST FILE", -30)
+ val ClearCache: SqlStatementClassification = spark("CLEAR CACHE", -31)
+ val RefreshResourceCmd: SqlStatementClassification = spark("REFRESH RESOURCE", -32)
+ val DescribeQuery: SqlStatementClassification = spark("DESCRIBE QUERY", -33)
+ val ShowCatalogs: SqlStatementClassification = spark("SHOW CATALOGS", -34)
+ val ShowCurrentNamespace: SqlStatementClassification =
+ spark("SHOW CURRENT NAMESPACE", -35)
+ val SetCatalog: SqlStatementClassification = spark("SET CATALOG", -36)
+
+ private def spark(identifier: String, code: Int): SqlStatementClassification = {
+ assert(code < 0, s"Spark statement codes must be negative, got $code")
+ SqlStatementClassification(statementIdentifier = identifier, statementCode = code)
+ }
+
+ /** Classify an unresolved logical plan. */
+ def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
+ case UnresolvedWith(child, _, _) => classify(child)
+ case _: CompoundBody => BeginEnd
+ case _: InsertIntoStatement => Insert
+ case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
+ case _: UpdateTable => UpdateWhere
+ case _: MergeIntoTable => Merge
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect => CreateTable
+ case _: CreateTable | _: CreateTableLike | _: ReplaceTable => CreateTable
+ case _: CreateView | _: CreateViewCommand | _: CreateTempViewUsing => CreateView
+ case _: DropTable => DropTable
+ case _: DropView => DropView
+ case _: CreateNamespace => CreateSchema
+ case _: DropNamespace => DropSchema
+ case _: SetCatalogAndNamespace | _: SetNamespaceCommand => SetSchema
+ case _: SetCatalogCommand => SetCatalog
+ case _: TruncateTable => TruncateTable
+ case _: CreateFunction | _: CreateFunctionCommand |
+ _: CreateUserDefinedFunction | _: CreateUserDefinedFunctionCommand =>
+ CreateRoutine
+ case _: DropFunction | _: DropFunctionCommand => DropRoutine
+ case _: UnresolvedExecuteImmediate => ExecuteImmediate
+ case _: Call => Call
+ case _: CommentOnTable => CommentOnTable
+ case _: AlterTableCommand | _: RenameTable => AlterTable
+ case _: CacheTable => CacheTable
+ case _: CacheTableAsSelect => CacheTableAsSelect
+ case _: UncacheTable => UncacheTable
+ case _: RefreshTable => RefreshTable
+ case _: ShowTables | _: ShowTablesExtended => ShowTables
+ case _: DescribeRelation | _: DescribeTablePartition | _: DescribeColumn =>
+ DescribeTable
+ case _: DescribeQueryCommand => DescribeQuery
+ case _: AnalyzeTable | _: AnalyzeTables | _: AnalyzeColumn => AnalyzeTable
+ case _: CreateVariable => DeclareVariable
+ case _: SetVariable => SetVariable
+ case _: DropVariable => DropVariable
+ case _: ShowTableProperties => ShowTableProperties
+ case _: DescribeNamespace => DescribeNamespace
+ case _: ShowFunctions => ShowFunctions
+ case _: DescribeFunction => DescribeFunction
+ case _: ShowCreateTable => ShowCreateTable
+ case _: ShowColumns => ShowColumns
+ case _: ShowPartitions | _: ShowTablePartition => ShowPartitions
+ case _: ShowViews => ShowViews
+ case _: RefreshFunction => RefreshFunction
+ case _: CommentOnNamespace => CommentOnNamespace
+ case _: ExplainCommand => Explain
+ case _: SetCommand => Set
+ case _: ResetCommand => Reset
+ case _: AddJarsCommand => AddJar
+ case _: AddFilesCommand => AddFile
+ case _: AddArchivesCommand => AddArchive
+ case _: ListJarsCommand => ListJar
+ case _: ListFilesCommand => ListFile
+ case ClearCacheCommand => ClearCache
+ case _: RefreshResource => RefreshResourceCmd
+ case _: ShowCatalogsCommand => ShowCatalogs
+ case _: ShowCurrentNamespaceCommand => ShowCurrentNamespace
+ case _: Command => Unrecognized
+ case p if isQueryPlan(p) => Select
+ case _ => Unrecognized
+ }
+
+ /**
+ * Allowlisted query-shaped plans. Unknown non-command plans are not assumed
+ * to be SELECT.
+ */
+ private def isQueryPlan(plan: LogicalPlan): Boolean = plan match {
+ case _: Project | _: Aggregate | _: Distinct | _: Filter | _: Sort |
+ _: GlobalLimit | _: LocalLimit | _: Join | _: Union | _: Except |
+ _: Intersect | _: SubqueryAlias | _: Repartition |
+ _: RepartitionByExpression | _: Sample | _: Range |
+ _: OneRowRelation | _: LocalRelation | _: Deduplicate |
+ _: Expand | _: Generate | _: Window | _: Tail | _: Offset |
+ _: LateralJoin | _: UnresolvedHaving | _: CollectMetrics |
+ _: WithCTE => true
+ case _ => false
Review Comment:
**[High] `isQueryPlan` allowlist still misses common query roots**
`isQueryPlan` still misses `UnresolvedRelation` (`TABLE …`), `UnresolvedInlineTable`/`ResolvedInlineTable` (`VALUES` when not eagerly evaluated), and `RelationTimeTravel`. `TABLE …` currently classifies as Unrecognized; `VALUES` can flip between SELECT and Unrecognized based on eager-eval conf.
Please extend the allowlist and pin these in tests.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.SparkFunSuite
+
+/**
+ * Pin Table 39 codes and parser-surface contracts that goldens do not cover.
+ * Behavioral coverage lives in sql-tests/inputs/parse-sql.sql.
+ */
Review Comment:
**[Medium] Unit suite still overlaps goldens**
`ParseSqlResultSuite` still overlaps goldens (SELECT/CTE/CREATE VIEW/EXPLAIN). Prefer goldens for behavior; keep the suite for Table 39 code pins and cases goldens can’t cover (`TABLE`/`VALUES` classification, `CREATE FUNCTION` not in `table_references`, internal error propagation).
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedExecuteImmediate, UnresolvedHaving}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.execution.command._
+import org.apache.spark.sql.execution.datasources.{CreateTempViewUsing, RefreshResource}
+
+/**
+ * Classification of a parsed SQL statement using ISO/IEC 9075-2:2023 Table 39,
+ * "SQL-statement codes" (clause 23.1 <get diagnostics statement>).
+ *
+ * @param statementIdentifier Table 39 Identifier column (or Spark product name)
+ * @param statementCode Table 39 Code column; Spark-only statements use negative
+ * implementation-defined codes (Table 39 IE005 / IV190)
+ */
+case class SqlStatementClassification(
+ statementIdentifier: String,
+ statementCode: Int)
+
+/**
+ * Maps unresolved [[LogicalPlan]]s to Table 39 statement codes.
+ *
+ * Spark-only statements use the standard's implementation-defined escape hatch:
+ * a product-specific identifier and a distinct negative code. Codes are
+ * append-only and must never be renumbered.
+ *
+ * Unknown plans map to [[Unrecognized]] (empty identifier, code 0). Query
+ * shapes are allowlisted; unknown non-commands are not assumed to be SELECT.
+ */
+object SqlStatementCodes {
+
+ // Standard Table 39 entries used by Spark SQL (ISO/IEC 9075-2:2023).
+ val Select: SqlStatementClassification = SqlStatementClassification("SELECT", 21)
+ val Insert: SqlStatementClassification = SqlStatementClassification("INSERT", 50)
+ val DeleteWhere: SqlStatementClassification = SqlStatementClassification("DELETE WHERE", 19)
+ val UpdateWhere: SqlStatementClassification = SqlStatementClassification("UPDATE WHERE", 82)
+ val Merge: SqlStatementClassification = SqlStatementClassification("MERGE", 128)
+ val CreateTable: SqlStatementClassification = SqlStatementClassification("CREATE TABLE", 77)
+ val CreateView: SqlStatementClassification = SqlStatementClassification("CREATE VIEW", 84)
+ val DropTable: SqlStatementClassification = SqlStatementClassification("DROP TABLE", 32)
+ val DropView: SqlStatementClassification = SqlStatementClassification("DROP VIEW", 36)
+ val AlterTable: SqlStatementClassification = SqlStatementClassification("ALTER TABLE", 4)
+ val CreateSchema: SqlStatementClassification = SqlStatementClassification("CREATE SCHEMA", 64)
+ val DropSchema: SqlStatementClassification = SqlStatementClassification("DROP SCHEMA", 31)
+ val SetSchema: SqlStatementClassification = SqlStatementClassification("SET SCHEMA", 74)
+ val TruncateTable: SqlStatementClassification =
+ SqlStatementClassification("TRUNCATE TABLE", 139)
+ val CreateRoutine: SqlStatementClassification = SqlStatementClassification("CREATE ROUTINE", 14)
+ val DropRoutine: SqlStatementClassification = SqlStatementClassification("DROP ROUTINE", 30)
+ val ExecuteImmediate: SqlStatementClassification =
+ SqlStatementClassification("EXECUTE IMMEDIATE", 43)
+ val Call: SqlStatementClassification = SqlStatementClassification("CALL", 7)
+
+ // Table 39 "Unrecognized statements": empty identifier, code 0.
+ val Unrecognized: SqlStatementClassification = SqlStatementClassification("", 0)
+
+ // Spark product-specific identifiers with append-only negative codes
+ // (Table 39 implementation-defined / IE005 row: negative Code values).
+ val CacheTable: SqlStatementClassification = spark("CACHE TABLE", -1)
+ val CacheTableAsSelect: SqlStatementClassification = spark("CACHE TABLE AS SELECT", -2)
+ val UncacheTable: SqlStatementClassification = spark("UNCACHE TABLE", -3)
+ val RefreshTable: SqlStatementClassification = spark("REFRESH TABLE", -4)
+ val ShowTables: SqlStatementClassification = spark("SHOW TABLES", -5)
+ val DescribeTable: SqlStatementClassification = spark("DESCRIBE TABLE", -6)
+ val AnalyzeTable: SqlStatementClassification = spark("ANALYZE TABLE", -7)
+ val DeclareVariable: SqlStatementClassification = spark("DECLARE VARIABLE", -8)
+ val SetVariable: SqlStatementClassification = spark("SET VARIABLE", -9)
+ val DropVariable: SqlStatementClassification = spark("DROP VARIABLE", -10)
+ val ShowTableProperties: SqlStatementClassification = spark("SHOW TBLPROPERTIES", -11)
+ val DescribeNamespace: SqlStatementClassification = spark("DESCRIBE NAMESPACE", -12)
+ val ShowFunctions: SqlStatementClassification = spark("SHOW FUNCTIONS", -13)
+ val DescribeFunction: SqlStatementClassification = spark("DESCRIBE FUNCTION", -14)
+ val ShowCreateTable: SqlStatementClassification = spark("SHOW CREATE TABLE", -15)
+ val ShowColumns: SqlStatementClassification = spark("SHOW COLUMNS", -16)
+ val ShowPartitions: SqlStatementClassification = spark("SHOW PARTITIONS", -17)
+ val ShowViews: SqlStatementClassification = spark("SHOW VIEWS", -18)
+ val RefreshFunction: SqlStatementClassification = spark("REFRESH FUNCTION", -19)
+ val CommentOnNamespace: SqlStatementClassification = spark("COMMENT ON NAMESPACE", -20)
+ val CommentOnTable: SqlStatementClassification = spark("COMMENT ON TABLE", -21)
+ // SQL/PSM-style scripting (9075-4); not in Foundation Table 39.
+ val BeginEnd: SqlStatementClassification = spark("BEGIN END", -22)
+ // SparkSqlParser-only session / resource commands (append-only).
+ val Explain: SqlStatementClassification = spark("EXPLAIN", -23)
+ val Set: SqlStatementClassification = spark("SET", -24)
+ val Reset: SqlStatementClassification = spark("RESET", -25)
+ val AddJar: SqlStatementClassification = spark("ADD JAR", -26)
+ val AddFile: SqlStatementClassification = spark("ADD FILE", -27)
+ val AddArchive: SqlStatementClassification = spark("ADD ARCHIVE", -28)
+ val ListJar: SqlStatementClassification = spark("LIST JAR", -29)
+ val ListFile: SqlStatementClassification = spark("LIST FILE", -30)
+ val ClearCache: SqlStatementClassification = spark("CLEAR CACHE", -31)
+ val RefreshResourceCmd: SqlStatementClassification = spark("REFRESH RESOURCE", -32)
+ val DescribeQuery: SqlStatementClassification = spark("DESCRIBE QUERY", -33)
+ val ShowCatalogs: SqlStatementClassification = spark("SHOW CATALOGS", -34)
+ val ShowCurrentNamespace: SqlStatementClassification =
+ spark("SHOW CURRENT NAMESPACE", -35)
+ val SetCatalog: SqlStatementClassification = spark("SET CATALOG", -36)
+
+ private def spark(identifier: String, code: Int): SqlStatementClassification = {
+ assert(code < 0, s"Spark statement codes must be negative, got $code")
+ SqlStatementClassification(statementIdentifier = identifier, statementCode = code)
+ }
+
+ /** Classify an unresolved logical plan. */
+ def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
+ case UnresolvedWith(child, _, _) => classify(child)
+ case _: CompoundBody => BeginEnd
+ case _: InsertIntoStatement => Insert
+ case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
+ case _: UpdateTable => UpdateWhere
+ case _: MergeIntoTable => Merge
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect => CreateTable
+ case _: CreateTable | _: CreateTableLike | _: ReplaceTable => CreateTable
+ case _: CreateView | _: CreateViewCommand | _: CreateTempViewUsing => CreateView
+ case _: DropTable => DropTable
+ case _: DropView => DropView
+ case _: CreateNamespace => CreateSchema
+ case _: DropNamespace => DropSchema
+ case _: SetCatalogAndNamespace | _: SetNamespaceCommand => SetSchema
+ case _: SetCatalogCommand => SetCatalog
+ case _: TruncateTable => TruncateTable
+ case _: CreateFunction | _: CreateFunctionCommand |
+ _: CreateUserDefinedFunction | _: CreateUserDefinedFunctionCommand =>
+ CreateRoutine
+ case _: DropFunction | _: DropFunctionCommand => DropRoutine
+ case _: UnresolvedExecuteImmediate => ExecuteImmediate
+ case _: Call => Call
+ case _: CommentOnTable => CommentOnTable
+ case _: AlterTableCommand | _: RenameTable => AlterTable
+ case _: CacheTable => CacheTable
+ case _: CacheTableAsSelect => CacheTableAsSelect
+ case _: UncacheTable => UncacheTable
+ case _: RefreshTable => RefreshTable
+ case _: ShowTables | _: ShowTablesExtended => ShowTables
+ case _: DescribeRelation | _: DescribeTablePartition | _: DescribeColumn =>
+ DescribeTable
+ case _: DescribeQueryCommand => DescribeQuery
+ case _: AnalyzeTable | _: AnalyzeTables | _: AnalyzeColumn => AnalyzeTable
+ case _: CreateVariable => DeclareVariable
+ case _: SetVariable => SetVariable
+ case _: DropVariable => DropVariable
+ case _: ShowTableProperties => ShowTableProperties
+ case _: DescribeNamespace => DescribeNamespace
+ case _: ShowFunctions => ShowFunctions
+ case _: DescribeFunction => DescribeFunction
+ case _: ShowCreateTable => ShowCreateTable
+ case _: ShowColumns => ShowColumns
+ case _: ShowPartitions | _: ShowTablePartition => ShowPartitions
+ case _: ShowViews => ShowViews
+ case _: RefreshFunction => RefreshFunction
+ case _: CommentOnNamespace => CommentOnNamespace
+ case _: ExplainCommand => Explain
+ case _: SetCommand => Set
+ case _: ResetCommand => Reset
+ case _: AddJarsCommand => AddJar
+ case _: AddFilesCommand => AddFile
+ case _: AddArchivesCommand => AddArchive
+ case _: ListJarsCommand => ListJar
+ case _: ListFilesCommand => ListFile
+ case ClearCacheCommand => ClearCache
+ case _: RefreshResource => RefreshResourceCmd
+ case _: ShowCatalogsCommand => ShowCatalogs
+ case _: ShowCurrentNamespaceCommand => ShowCurrentNamespace
+ case _: Command => Unrecognized
+ case p if isQueryPlan(p) => Select
+ case _ => Unrecognized
Review Comment:
**[Low] `CreateMetricView` → Unrecognized**
(nit) `CreateMetricView` falls through to Unrecognized via `Command`. If `parse_sql` is meant to classify Spark DDL comprehensively, add a Spark-specific code; otherwise call out the gap.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
Review Comment:
**[High] ThreadLocal stock parser ≠ session extensions**
ThreadLocal `new SparkSqlParser()` ignores `SparkSessionExtensions` parser wrapping. `VariableSubstitution` / `SQLConf.get` work, but the class doc saying coverage matches the production session parser is overstated.
Please document the intentional limitation (stock parser only under distributed eval) or note why extensions cannot be honored — ideally with a pin that extensions are ignored.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ // User-facing parse / scripting failures become JSON; internal errors fail.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
Review Comment:
**[Medium] JSON result schema still unversioned**
Expression text is gone (good), but the JSON is still unversioned. Please add a `schema_version` field before this ships so clients can evolve safely.
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala:
##########
@@ -96,7 +96,12 @@ abstract class BaseSessionStateBuilder(
*/
protected lazy val functionRegistry: FunctionRegistry = {
parentState.map(_.functionRegistry.clone())
- .getOrElse(extensions.registerFunctions(FunctionRegistry.builtin.clone()))
+ .getOrElse {
+ val registry = FunctionRegistry.builtin.clone()
+ // sql/core-only builtins that need SparkSqlParser.
+ ParseSql.register(registry)
+ extensions.registerFunctions(registry)
Review Comment:
**[Low] Connect / package placement**
(nit) Confirming Connect is covered via `BaseSessionStateBuilder` session registry — LGTM. `catalyst` package under `sql/core` matches existing core-only catalyst types; no change needed.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3775720697
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ // User-facing parse / scripting failures become JSON; internal errors fail.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
Review Comment:
Pushing back on adding `schema_version` for now.
The point of returning JSON (vs a typed struct) is that clients can tolerate additive evolution: new fields appear, old clients ignore them. We are intentionally freezing only the fields we have today as the contract, and we already removed the unstable piece (`select_list[].expression` / `Expression.sql`).
A `schema_version` would help if we expected *breaking* renames/removals or semantic flips of existing keys. That is not the plan — statement codes are append-only, and JSON keys are additive. Introducing a version field now mostly forces every client to branch on a number without buying protection against the kind of change we actually intend to make.
If we later need an incompatible change, we can add `schema_version` (or a new function) at that point. Until then, keeping the payload unversioned matches how we expect this API to evolve.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3805793698
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5610,6 +5610,16 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PARSE_SQL_ENABLED =
+ buildConf("spark.sql.parseSql.enabled")
Review Comment:
**Non-blocking:**
Could we name this as a SQL-function gate, for example `spark.sql.functions.parseSql.enabled`? The current key can read as enabling parser behavior generally rather than making the `parse_sql` function available.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3797665197
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val refs = collectPlanReferences(plan)
+ if (refs.tables.nonEmpty) {
+ fields += "table_references" -> JArray(refs.tables.map(partsToJArray).toList)
+ }
+ if (refs.functions.nonEmpty) {
+ fields += "function_references" -> JArray(refs.functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ refs.parameterMarkers.foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan root (children expose only nested
+ * children; visit the root without re-walking those children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ visitNonChildSlots(p)(f)
+ }
+ }
+
+ /**
+ * Visit plan slots that are not exposed via `children` / expression
+ * subqueries. For [[SingleStatement]], only the wrapped root (and its
+ * subqueries / non-child slots) is visited -- its children were already
+ * walked by the outer `foreachWithSubqueries` via
+ * `SingleStatement.children`.
+ */
+ private def visitNonChildSlots(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ visitSingleStatementRoot(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case ExplainCommand(logicalPlan, _) =>
+ foreachPlanDeep(logicalPlan)(f)
+ case DescribeQueryCommand(_, queryPlan) =>
+ foreachPlanDeep(queryPlan)(f)
+ case _ =>
+ }
+ }
+
+ /**
+ * Visit a [[SingleStatement]] wrapped root without re-traversing its
+ * children (already exposed by `SingleStatement.children`).
+ */
+ private def visitSingleStatementRoot(root: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ f(root)
+ root.subqueries.foreach(sq => foreachPlanDeep(sq)(f))
+ visitNonChildSlots(root)(f)
+ }
+
+ /** Multipart name from a table/view-shaped plan node, if any. */
+ private def tableOrViewParts(plan: LogicalPlan): Option[Seq[String]] = plan match {
+ case u: UnresolvedRelation => Some(u.multipartIdentifier)
+ case u: UnresolvedTable => Some(u.multipartIdentifier)
+ case u: UnresolvedView => Some(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => Some(u.nameParts)
+ case _ => None
+ }
+
+ private def tableIdentifierParts(id: org.apache.spark.sql.catalyst.TableIdentifier): Seq[String] =
+ id.catalog.toSeq ++ id.database.toSeq :+ id.table
+
+ private final case class PlanReferences(
+ tables: Seq[Seq[String]],
+ functions: Seq[Seq[String]],
+ parameterMarkers: Option[JObject])
+
+ /**
+ * Collect multipart table/view identifiers, function names, and parameter
+ * markers for lineage. CTE definition names and correlation aliases are
+ * omitted from tables; tables referenced inside CTE bodies are still
+ * included. Function / variable identifiers are not collected as tables.
+ * Deduplicates while preserving first-seen order. Uses one CTE-name pass
+ * then one combined deep walk for tables, functions, and parameters.
+ */
+ private def collectPlanReferences(plan: LogicalPlan): PlanReferences = {
+ val tables = mutable.LinkedHashSet.empty[Seq[String]]
+ val functions = mutable.LinkedHashSet.empty[Seq[String]]
+ val namedParams = mutable.LinkedHashSet.empty[String]
+ var unnamedCount = 0
+ // CTE names are always single-part in the grammar; UnresolvedRelation refs
+ // to CTEs are likewise single-part, so filtering matches that shape.
+ val cteNames = mutable.HashSet.empty[String]
Review Comment:
**Blocking:**
Keep CTE-name filtering scoped to the `UnresolvedWith` that defines each alias. For example, `SELECT * FROM real_t WHERE EXISTS (WITH real_t AS (SELECT * FROM inner_base) SELECT * FROM real_t)` should report both `real_t` and `inner_base`, but this statement-wide set suppresses the outer real table too. Please carry the active CTE environment during traversal and add this shadowing case to the lineage tests.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3811879086
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5610,6 +5610,16 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PARSE_SQL_ENABLED =
+ buildConf("spark.sql.function.parseSql.enabled")
+ .doc("When true, enables the parse_sql SQL function. This feature is under active " +
+ "development; the JSON contract may change across releases while the flag remains " +
+ "off by default.")
+ .version("5.0.0")
Review Comment:
4.4.0
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, FunctionRegistryBase, TypeCheckResult}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseSqlResult
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, lineage references, select-list names,
+ * parameters), or a STANDARD-format error object when the statement does not
+ * parse.
+ *
+ * Behind [[SQLConf.PARSE_SQL_ENABLED]] while the JSON contract is still
+ * evolving. Designed for batch evaluation over DataFrames of SQL text.
+ * User-facing parse errors become JSON; unexpected internal failures propagate.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` with the stock Spark SQL parser and
+ returns a JSON string describing the statement (parse success, Table 39 statement
+ identifier/code, target and source table references for lineage, select-list column
+ names, and parameter markers). Session parser extensions are not applied.
+ Requires spark.sql.function.parseSql.enabled=true. On syntax / parse error returns JSON
+ with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
+ arguments = """
+ Arguments:
+ * sqlStmt - A SQL statement string to parse.
+ An expression that evaluates to a string.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_('SELECT a, b FROM t');
+ {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}
+ > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+ PARSE_SYNTAX_ERROR
+ """,
+ group = "misc_funcs",
+ since = "5.0.0")
Review Comment:
4.3.0
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3813253070
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5610,6 +5610,16 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PARSE_SQL_ENABLED =
+ buildConf("spark.sql.function.parseSql.enabled")
+ .doc("When true, enables the parse_sql SQL function. This feature is under active " +
+ "development; the JSON contract may change across releases while the flag remains " +
+ "off by default.")
+ .version("5.0.0")
Review Comment:
Addressed: set `.version("4.4.0")` to match `branch-4.x` (`4.4.0-SNAPSHOT`).
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, FunctionRegistryBase, TypeCheckResult}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.parser.ParseSqlResult
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.types.StringTypeWithCollation
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved statement (identifier/code, lineage references, select-list names,
+ * parameters), or a STANDARD-format error object when the statement does not
+ * parse.
+ *
+ * Behind [[SQLConf.PARSE_SQL_ENABLED]] while the JSON contract is still
+ * evolving. Designed for batch evaluation over DataFrames of SQL text.
+ * User-facing parse errors become JSON; unexpected internal failures propagate.
+ */
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` with the stock Spark SQL parser and
+ returns a JSON string describing the statement (parse success, Table 39 statement
+ identifier/code, target and source table references for lineage, select-list column
+ names, and parameter markers). Session parser extensions are not applied.
+ Requires spark.sql.function.parseSql.enabled=true. On syntax / parse error returns JSON
+ with `parse_success` false, source location, and a nested STANDARD error object
+ instead of throwing.""",
+ arguments = """
+ Arguments:
+ * sqlStmt - A SQL statement string to parse.
+ An expression that evaluates to a string.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_('SELECT a, b FROM t');
+ {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}
+ > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+ PARSE_SYNTAX_ERROR
+ """,
+ group = "misc_funcs",
+ since = "5.0.0")
Review Comment:
Addressed: set `since` / `FunctionRegistryBase.build` to `4.4.0` to match `branch-4.x` (`4.4.0-SNAPSHOT`). Used `4.4.0` rather than `4.3.0` so the annotation stays consistent with the config version and the current feature branch.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3797669762
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
Review Comment:
Thanks, the root-only traversal and focused positional-marker coverage address this.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
Review Comment:
Thanks, the combined reference walk resolves the per-row traversal concern.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3775832304
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ // User-facing parse / scripting failures become JSON; internal errors fail.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ }
Review Comment:
Addressed: catch is now only `ParseException` / `SqlScriptingException`. Other `SparkThrowable`s (including `SparkException.internalError`) propagate and fail the function.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedExecuteImmediate, UnresolvedHaving}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.execution.command._
+import org.apache.spark.sql.execution.datasources.{CreateTempViewUsing, RefreshResource}
+
+/**
+ * Classification of a parsed SQL statement using ISO/IEC 9075-2:2023 Table 39,
+ * "SQL-statement codes" (clause 23.1 <get diagnostics statement>).
+ *
+ * @param statementIdentifier Table 39 Identifier column (or Spark product name)
+ * @param statementCode Table 39 Code column; Spark-only statements use negative
+ * implementation-defined codes (Table 39 IE005 / IV190)
+ */
+case class SqlStatementClassification(
+ statementIdentifier: String,
+ statementCode: Int)
+
+/**
+ * Maps unresolved [[LogicalPlan]]s to Table 39 statement codes.
+ *
+ * Spark-only statements use the standard's implementation-defined escape hatch:
+ * a product-specific identifier and a distinct negative code. Codes are
+ * append-only and must never be renumbered.
+ *
+ * Unknown plans map to [[Unrecognized]] (empty identifier, code 0). Query
+ * shapes are allowlisted; unknown non-commands are not assumed to be SELECT.
+ */
+object SqlStatementCodes {
+
+ // Standard Table 39 entries used by Spark SQL (ISO/IEC 9075-2:2023).
+ val Select: SqlStatementClassification = SqlStatementClassification("SELECT", 21)
+ val Insert: SqlStatementClassification = SqlStatementClassification("INSERT", 50)
+ val DeleteWhere: SqlStatementClassification = SqlStatementClassification("DELETE WHERE", 19)
+ val UpdateWhere: SqlStatementClassification = SqlStatementClassification("UPDATE WHERE", 82)
+ val Merge: SqlStatementClassification = SqlStatementClassification("MERGE", 128)
+ val CreateTable: SqlStatementClassification = SqlStatementClassification("CREATE TABLE", 77)
+ val CreateView: SqlStatementClassification = SqlStatementClassification("CREATE VIEW", 84)
+ val DropTable: SqlStatementClassification = SqlStatementClassification("DROP TABLE", 32)
+ val DropView: SqlStatementClassification = SqlStatementClassification("DROP VIEW", 36)
+ val AlterTable: SqlStatementClassification = SqlStatementClassification("ALTER TABLE", 4)
+ val CreateSchema: SqlStatementClassification = SqlStatementClassification("CREATE SCHEMA", 64)
+ val DropSchema: SqlStatementClassification = SqlStatementClassification("DROP SCHEMA", 31)
+ val SetSchema: SqlStatementClassification = SqlStatementClassification("SET SCHEMA", 74)
+ val TruncateTable: SqlStatementClassification =
+ SqlStatementClassification("TRUNCATE TABLE", 139)
+ val CreateRoutine: SqlStatementClassification = SqlStatementClassification("CREATE ROUTINE", 14)
+ val DropRoutine: SqlStatementClassification = SqlStatementClassification("DROP ROUTINE", 30)
+ val ExecuteImmediate: SqlStatementClassification =
+ SqlStatementClassification("EXECUTE IMMEDIATE", 43)
+ val Call: SqlStatementClassification = SqlStatementClassification("CALL", 7)
+
+ // Table 39 "Unrecognized statements": empty identifier, code 0.
+ val Unrecognized: SqlStatementClassification = SqlStatementClassification("", 0)
+
+ // Spark product-specific identifiers with append-only negative codes
+ // (Table 39 implementation-defined / IE005 row: negative Code values).
+ val CacheTable: SqlStatementClassification = spark("CACHE TABLE", -1)
+ val CacheTableAsSelect: SqlStatementClassification = spark("CACHE TABLE AS SELECT", -2)
+ val UncacheTable: SqlStatementClassification = spark("UNCACHE TABLE", -3)
+ val RefreshTable: SqlStatementClassification = spark("REFRESH TABLE", -4)
+ val ShowTables: SqlStatementClassification = spark("SHOW TABLES", -5)
+ val DescribeTable: SqlStatementClassification = spark("DESCRIBE TABLE", -6)
+ val AnalyzeTable: SqlStatementClassification = spark("ANALYZE TABLE", -7)
+ val DeclareVariable: SqlStatementClassification = spark("DECLARE VARIABLE", -8)
+ val SetVariable: SqlStatementClassification = spark("SET VARIABLE", -9)
+ val DropVariable: SqlStatementClassification = spark("DROP VARIABLE", -10)
+ val ShowTableProperties: SqlStatementClassification = spark("SHOW TBLPROPERTIES", -11)
+ val DescribeNamespace: SqlStatementClassification = spark("DESCRIBE NAMESPACE", -12)
+ val ShowFunctions: SqlStatementClassification = spark("SHOW FUNCTIONS", -13)
+ val DescribeFunction: SqlStatementClassification = spark("DESCRIBE FUNCTION", -14)
+ val ShowCreateTable: SqlStatementClassification = spark("SHOW CREATE TABLE", -15)
+ val ShowColumns: SqlStatementClassification = spark("SHOW COLUMNS", -16)
+ val ShowPartitions: SqlStatementClassification = spark("SHOW PARTITIONS", -17)
+ val ShowViews: SqlStatementClassification = spark("SHOW VIEWS", -18)
+ val RefreshFunction: SqlStatementClassification = spark("REFRESH FUNCTION", -19)
+ val CommentOnNamespace: SqlStatementClassification = spark("COMMENT ON NAMESPACE", -20)
+ val CommentOnTable: SqlStatementClassification = spark("COMMENT ON TABLE", -21)
+ // SQL/PSM-style scripting (9075-4); not in Foundation Table 39.
+ val BeginEnd: SqlStatementClassification = spark("BEGIN END", -22)
+ // SparkSqlParser-only session / resource commands (append-only).
+ val Explain: SqlStatementClassification = spark("EXPLAIN", -23)
+ val Set: SqlStatementClassification = spark("SET", -24)
+ val Reset: SqlStatementClassification = spark("RESET", -25)
+ val AddJar: SqlStatementClassification = spark("ADD JAR", -26)
+ val AddFile: SqlStatementClassification = spark("ADD FILE", -27)
+ val AddArchive: SqlStatementClassification = spark("ADD ARCHIVE", -28)
+ val ListJar: SqlStatementClassification = spark("LIST JAR", -29)
+ val ListFile: SqlStatementClassification = spark("LIST FILE", -30)
+ val ClearCache: SqlStatementClassification = spark("CLEAR CACHE", -31)
+ val RefreshResourceCmd: SqlStatementClassification = spark("REFRESH RESOURCE", -32)
+ val DescribeQuery: SqlStatementClassification = spark("DESCRIBE QUERY", -33)
+ val ShowCatalogs: SqlStatementClassification = spark("SHOW CATALOGS", -34)
+ val ShowCurrentNamespace: SqlStatementClassification =
+ spark("SHOW CURRENT NAMESPACE", -35)
+ val SetCatalog: SqlStatementClassification = spark("SET CATALOG", -36)
+
+ private def spark(identifier: String, code: Int): SqlStatementClassification = {
+ assert(code < 0, s"Spark statement codes must be negative, got $code")
+ SqlStatementClassification(statementIdentifier = identifier, statementCode = code)
+ }
+
+ /** Classify an unresolved logical plan. */
+ def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
+ case UnresolvedWith(child, _, _) => classify(child)
+ case _: CompoundBody => BeginEnd
+ case _: InsertIntoStatement => Insert
+ case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
+ case _: UpdateTable => UpdateWhere
+ case _: MergeIntoTable => Merge
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect => CreateTable
+ case _: CreateTable | _: CreateTableLike | _: ReplaceTable => CreateTable
+ case _: CreateView | _: CreateViewCommand | _: CreateTempViewUsing => CreateView
+ case _: DropTable => DropTable
+ case _: DropView => DropView
+ case _: CreateNamespace => CreateSchema
+ case _: DropNamespace => DropSchema
+ case _: SetCatalogAndNamespace | _: SetNamespaceCommand => SetSchema
+ case _: SetCatalogCommand => SetCatalog
+ case _: TruncateTable => TruncateTable
+ case _: CreateFunction | _: CreateFunctionCommand |
+ _: CreateUserDefinedFunction | _: CreateUserDefinedFunctionCommand =>
+ CreateRoutine
+ case _: DropFunction | _: DropFunctionCommand => DropRoutine
+ case _: UnresolvedExecuteImmediate => ExecuteImmediate
+ case _: Call => Call
+ case _: CommentOnTable => CommentOnTable
+ case _: AlterTableCommand | _: RenameTable => AlterTable
+ case _: CacheTable => CacheTable
+ case _: CacheTableAsSelect => CacheTableAsSelect
+ case _: UncacheTable => UncacheTable
+ case _: RefreshTable => RefreshTable
+ case _: ShowTables | _: ShowTablesExtended => ShowTables
+ case _: DescribeRelation | _: DescribeTablePartition | _: DescribeColumn =>
+ DescribeTable
+ case _: DescribeQueryCommand => DescribeQuery
+ case _: AnalyzeTable | _: AnalyzeTables | _: AnalyzeColumn => AnalyzeTable
+ case _: CreateVariable => DeclareVariable
+ case _: SetVariable => SetVariable
+ case _: DropVariable => DropVariable
+ case _: ShowTableProperties => ShowTableProperties
+ case _: DescribeNamespace => DescribeNamespace
+ case _: ShowFunctions => ShowFunctions
+ case _: DescribeFunction => DescribeFunction
+ case _: ShowCreateTable => ShowCreateTable
+ case _: ShowColumns => ShowColumns
+ case _: ShowPartitions | _: ShowTablePartition => ShowPartitions
+ case _: ShowViews => ShowViews
+ case _: RefreshFunction => RefreshFunction
+ case _: CommentOnNamespace => CommentOnNamespace
+ case _: ExplainCommand => Explain
+ case _: SetCommand => Set
+ case _: ResetCommand => Reset
+ case _: AddJarsCommand => AddJar
+ case _: AddFilesCommand => AddFile
+ case _: AddArchivesCommand => AddArchive
+ case _: ListJarsCommand => ListJar
+ case _: ListFilesCommand => ListFile
+ case ClearCacheCommand => ClearCache
+ case _: RefreshResource => RefreshResourceCmd
+ case _: ShowCatalogsCommand => ShowCatalogs
+ case _: ShowCurrentNamespaceCommand => ShowCurrentNamespace
+ case _: Command => Unrecognized
+ case p if isQueryPlan(p) => Select
+ case _ => Unrecognized
+ }
+
+ /**
+ * Allowlisted query-shaped plans. Unknown non-command plans are not assumed
+ * to be SELECT.
+ */
+ private def isQueryPlan(plan: LogicalPlan): Boolean = plan match {
+ case _: Project | _: Aggregate | _: Distinct | _: Filter | _: Sort |
+ _: GlobalLimit | _: LocalLimit | _: Join | _: Union | _: Except |
+ _: Intersect | _: SubqueryAlias | _: Repartition |
+ _: RepartitionByExpression | _: Sample | _: Range |
+ _: OneRowRelation | _: LocalRelation | _: Deduplicate |
+ _: Expand | _: Generate | _: Window | _: Tail | _: Offset |
+ _: LateralJoin | _: UnresolvedHaving | _: CollectMetrics |
+ _: WithCTE => true
+ case _ => false
Review Comment:
Addressed: `isQueryPlan` now includes `UnresolvedRelation` (`TABLE`), `UnresolvedInlineTable` / `ResolvedInlineTable` (`VALUES`), `RelationTimeTravel`, and `UnresolvedTableValuedFunction`. Unit test pins `TABLE` / `VALUES` (both eager-inline conf settings) as SELECT/21.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
Review Comment:
Addressed: documented on `ParseSqlResult` / `@ExpressionDescription` that we intentionally use a stock ThreadLocal `SparkSqlParser` because `parse_sql` must evaluate on executors without a session — `SparkSessionExtensions` parser wrappers are not applied under distributed eval.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses [[SparkSqlParser]] so coverage matches the production session parser
+ * (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ *
+ * On success the JSON includes the statement identifier/code (ISO/IEC
+ * 9075-2:2023 Table 39), table/function references, select-list names, and
+ * parameter markers. On parse failure it returns `parse_success: false` with
+ * source location and a nested STANDARD-format error object, and does not
+ * throw. Unexpected / internal failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ } catch {
+ // User-facing parse / scripting failures become JSON; internal errors fail.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ case e: SparkThrowable with Throwable =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ fields += "table_references" -> JArray(
+ collectTableReferences(plan).map(partsToJArray).toList)
+ fields += "function_references" -> JArray(
+ collectFunctionReferences(plan).map(partsToJArray).toList)
+ fields += "select_list" -> JArray(collectSelectList(plan).toList)
+ fields += "parameter_markers" -> parameterMarkersJson(plan)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ locationFields)
+ )))
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case ExplainCommand(logicalPlan, _) =>
+ foreachPlanDeep(logicalPlan)(f)
+ case DescribeQueryCommand(_, queryPlan) =>
+ foreachPlanDeep(queryPlan)(f)
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Collect multipart table/view identifiers for lineage (as written in the
+ * SQL). CTE definition names and correlation aliases are omitted; tables
+ * referenced inside CTE bodies are still included. Deduplicates while
+ * preserving first-seen order.
+ */
+ private def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ val cteNames = mutable.HashSet.empty[String]
+
+ def addCteNames(w: UnresolvedWith): Unit = {
+ w.cteRelations.foreach { case (name, _, _) =>
+ cteNames += name.toLowerCase(java.util.Locale.ROOT)
+ }
+ }
+
+ def isCteName(parts: Seq[String]): Boolean = parts match {
+ case Seq(name) => cteNames.contains(name.toLowerCase(java.util.Locale.ROOT))
+ case _ => false
+ }
+
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty && !isCteName(parts)) seen += parts
+ }
+
+ // First pass: gather CTE names in scope (including nested).
+ foreachPlanDeep(plan) {
+ case w: UnresolvedWith => addCteNames(w)
+ case _ =>
+ }
+
+ foreachPlanDeep(plan) {
+ case u: UnresolvedRelation => add(u.multipartIdentifier)
+ case u: UnresolvedTable => add(u.multipartIdentifier)
+ case u: UnresolvedView => add(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => add(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => add(u.nameParts)
Review Comment:
Addressed: dropped the blanket `UnresolvedIdentifier` collector. Table/view lineage comes from relation/table/view nodes plus explicit DDL target slots (`CreateView` / `CreateTable` / `DropTable` / …). `CREATE FUNCTION` / `DECLARE VARIABLE` names are excluded (pinned in unit + golden). CTE filtering remains single-part because CTE names and CTE references are single-part in the grammar.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.SparkFunSuite
+
+/**
+ * Pin Table 39 codes and parser-surface contracts that goldens do not cover.
+ * Behavioral coverage lives in sql-tests/inputs/parse-sql.sql.
+ */
Review Comment:
Addressed: slimmed `ParseSqlResultSuite` to Table 39 pins plus cases goldens are awkward for (`TABLE`/`VALUES` classification under both eager-inline settings, `CREATE FUNCTION`/`DECLARE VARIABLE` not in `table_references`). Overlapping SELECT/CTE/CREATE VIEW/EXPLAIN cases removed from the unit suite (covered by `parse-sql.sql`).
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala:
##########
@@ -96,7 +96,12 @@ abstract class BaseSessionStateBuilder(
*/
protected lazy val functionRegistry: FunctionRegistry = {
parentState.map(_.functionRegistry.clone())
- .getOrElse(extensions.registerFunctions(FunctionRegistry.builtin.clone()))
+ .getOrElse {
+ val registry = FunctionRegistry.builtin.clone()
+ // sql/core-only builtins that need SparkSqlParser.
+ ParseSql.register(registry)
+ extensions.registerFunctions(registry)
Review Comment:
Ack — no change needed.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3796062810
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
Review Comment:
Addressed: `visitSingleStatementRoot` now visits only the wrapped root (plus its expression subqueries / non-child slots). Children are left to the outer `foreachWithSubqueries` walk via `SingleStatement.children`, so positional markers are counted once. Added unit + golden coverage for `BEGIN SELECT * FROM t WHERE a = ?; END` (`unnamed_count: 1`).
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
Review Comment:
Addressed: kept the CTE-name pass, then collect tables, functions, and parameter markers together in one subsequent `foreachPlanDeep` walk (`collectPlanReferences`), preserving first-seen order.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case ExplainCommand(logicalPlan, _) =>
+ foreachPlanDeep(logicalPlan)(f)
+ case DescribeQueryCommand(_, queryPlan) =>
+ foreachPlanDeep(queryPlan)(f)
+ case _ =>
+ }
+ }
+ }
+
+ /** Multipart name from a table/view-shaped plan node, if any. */
+ private def tableOrViewParts(plan: LogicalPlan): Option[Seq[String]] = plan match {
+ case u: UnresolvedRelation => Some(u.multipartIdentifier)
+ case u: UnresolvedTable => Some(u.multipartIdentifier)
+ case u: UnresolvedView => Some(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => Some(u.nameParts)
+ case _ => None
+ }
+
+ private def tableIdentifierParts(id: org.apache.spark.sql.catalyst.TableIdentifier): Seq[String] =
+ id.catalog.toSeq ++ id.database.toSeq :+ id.table
+
+ /**
+ * Collect multipart table/view identifiers for lineage (as written in the
+ * SQL). CTE definition names and correlation aliases are omitted; tables
+ * referenced inside CTE bodies are still included. Function / variable
+ * identifiers are not collected. Deduplicates while preserving first-seen
+ * order.
+ */
+ private def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ // CTE names are always single-part in the grammar; UnresolvedRelation refs
+ // to CTEs are likewise single-part, so filtering matches that shape.
+ val cteNames = mutable.HashSet.empty[String]
+
+ def addCteNames(w: UnresolvedWith): Unit = {
+ w.cteRelations.foreach { case (name, _, _) =>
+ cteNames += name.toLowerCase(java.util.Locale.ROOT)
+ }
+ }
+
+ def isCteName(parts: Seq[String]): Boolean = parts match {
+ case Seq(name) => cteNames.contains(name.toLowerCase(java.util.Locale.ROOT))
+ case _ => false
+ }
+
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty && !isCteName(parts)) seen += parts
+ }
+
+ // First pass: gather CTE names in scope (including nested).
+ foreachPlanDeep(plan) {
+ case w: UnresolvedWith => addCteNames(w)
+ case _ =>
+ }
+
+ foreachPlanDeep(plan) {
+ case u: UnresolvedRelation => add(u.multipartIdentifier)
+ case u: UnresolvedTable => add(u.multipartIdentifier)
+ case u: UnresolvedView => add(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => add(u.multipartIdentifier)
+ // Table/view DDL targets only — not CreateFunction / CreateVariable names.
Review Comment:
Addressed: replaced the em-dash with ASCII `--` in that comment.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3806639264
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5610,6 +5610,16 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PARSE_SQL_ENABLED =
+ buildConf("spark.sql.parseSql.enabled")
Review Comment:
Addressed: renamed to `spark.sql.function.parseSql.enabled` (and updated docs / gating goldens). Used the singular `spark.sql.function.*` prefix already used by nearby function configs (`concatBinaryAsString`, `eltOutputAsString`, `protobufExtensions.enabled`) rather than `functions`.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58738][SQL] Add parse_sql function for SQL statement JSON analysis [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57962:
URL: https://github.com/apache/spark/pull/57962#discussion_r3784772747
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
Review Comment:
**Non-blocking:**
Each input row currently performs four full deep-plan walks: two for tables, one for functions, and one here for parameters. Since this API is intended for batch evaluation, please keep the CTE-name pass and collect tables, functions, and parameters together in one subsequent walk while preserving first-seen order.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
Review Comment:
**Blocking:**
This re-traverses children that `SingleStatement.children` already exposed to the outer `foreachWithSubqueries` walk. Named markers hide the duplicate visit in a set, but every `PosParameter` increments `unnamedCount`, so `BEGIN SELECT * FROM t WHERE a = ?; END` reports too many markers. Please visit the wrapped root without revisiting its children, and add a scripting test with `?` below the root.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.parser
+
+import scala.collection.mutable
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.{compact, parse => parseJson, render}
+
+import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext}
+import org.apache.spark.sql.exceptions.SqlScriptingException
+import org.apache.spark.sql.execution.SparkSqlParser
+import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand}
+import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
+
+/**
+ * Parses a SQL statement string and returns a compact JSON description of the
+ * unresolved plan (parse-only; no catalog resolution).
+ *
+ * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
+ * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
+ * Session-specific [[org.apache.spark.sql.SparkSessionExtensions]] parser
+ * wrappers are intentionally not applied: `parse_sql` must evaluate on
+ * executors without a session, so only the stock parser is available under
+ * distributed eval.
+ *
+ * On success the JSON always includes `parse_success`, the statement
+ * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
+ * fields (`table_references`, `function_references`, `select_list`,
+ * `parameter_markers`) when empty. On parse failure it returns
+ * `parse_success: false` with source location and a nested STANDARD-format
+ * error object, and does not throw. Only [[ParseException]] /
+ * [[SqlScriptingException]] are converted to JSON; unexpected / internal
+ * failures propagate so the function fails.
+ */
+object ParseSqlResult {
+
+ private val parser: ThreadLocal[SparkSqlParser] =
+ ThreadLocal.withInitial(() => new SparkSqlParser())
+
+ /** Parse `sql` and render the JSON result string. */
+ def fromSql(sql: String): String = {
+ try {
+ // Do not inherit the outer query's origin from the parse_sql expression.
+ // Errors and parsed nodes must refer to the SQL string passed to this function.
+ val origin = if (sql.nonEmpty) {
+ Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql))
+ } else {
+ Origin(sqlText = Some(sql))
+ }
+ CurrentOrigin.withOrigin(origin) {
+ val plan = parser.get().parsePlan(sql)
+ fromPlan(plan)
+ }
+ } catch {
+ // User-facing parse / scripting failures become JSON; everything else fails.
+ case e: ParseException =>
+ errorJson(e)
+ case e: SqlScriptingException =>
+ errorJson(e)
+ }
+ }
+
+ /** Build success JSON from an already-parsed unresolved plan. */
+ def fromPlan(plan: LogicalPlan): String = {
+ val classification = SqlStatementCodes.classify(plan)
+ val fields = mutable.ListBuffer.empty[JField]
+ fields += "parse_success" -> JBool(true)
+ fields += "statement_identifier" -> JString(classification.statementIdentifier)
+ fields += "statement_code" -> JInt(classification.statementCode)
+ // Omit unused collections / markers so consumers can treat absence as empty.
+ val tables = collectTableReferences(plan)
+ if (tables.nonEmpty) {
+ fields += "table_references" -> JArray(tables.map(partsToJArray).toList)
+ }
+ val functions = collectFunctionReferences(plan)
+ if (functions.nonEmpty) {
+ fields += "function_references" -> JArray(functions.map(partsToJArray).toList)
+ }
+ val selectList = collectSelectList(plan)
+ if (selectList.nonEmpty) {
+ fields += "select_list" -> JArray(selectList.toList)
+ }
+ parameterMarkersJson(plan).foreach(markers => fields += "parameter_markers" -> markers)
+ compact(render(JObject(fields.toList)))
+ }
+
+ private def errorJson(e: SparkThrowable with Throwable): String = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ val origin = e match {
+ case p: ParseException => Some(p.start)
+ case s: SqlScriptingException => Some(s.origin)
+ case _ => None
+ }
+ val locationFields = origin.toSeq.flatMap(originFields)
+ val contextFields = if (errorObj.obj.exists(_._1 == "queryContext")) {
+ Nil
+ } else {
+ origin.toSeq.flatMap(queryContextField)
+ }
+ compact(render(JObject(
+ "parse_success" -> JBool(false),
+ "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
+ )))
+ }
+
+ private def queryContextField(origin: Origin): Option[JField] = origin.context match {
+ case context: SQLQueryContext if context.isValid =>
+ Some("queryContext" -> JArray(List(JObject(
+ "objectType" -> JString(context.objectType),
+ "objectName" -> JString(context.objectName),
+ "startIndex" -> JInt(context.startIndex + 1),
+ "stopIndex" -> JInt(context.stopIndex + 1),
+ "fragment" -> JString(context.fragment)
+ ))))
+ case _ => None
+ }
+
+ private def originFields(origin: Origin): Seq[JField] = Seq(
+ origin.line.map(line => "line" -> JInt(line)),
+ origin.startPosition.map(position => "position" -> JInt(position))).flatten
+
+ private def partsToJArray(parts: Seq[String]): JArray =
+ JArray(parts.map(JString).toList)
+
+ /**
+ * Walk expressions in all product fields, including wrappers such as column
+ * definitions that [[LogicalPlan.expressions]] does not descend into.
+ */
+ private def foreachExpressionDeep(plan: LogicalPlan)(f: Expression => Unit): Unit = {
+ def visit(value: Any): Unit = value match {
+ case e: Expression => f(e)
+ case _: LogicalPlan =>
+ case values: Iterable[_] => values.foreach(visit)
+ case value: Product => value.productIterator.foreach(visit)
+ case _ =>
+ }
+ plan.productIterator.foreach(visit)
+ }
+
+ /**
+ * Deep plan walk covering tree slots that standard `collect` /
+ * `collectWithSubqueries` miss:
+ * - [[UnresolvedWith]] CTE definitions (`innerChildren`, not `children`)
+ * - [[InsertIntoStatement]].table (non-child plan slot)
+ * - [[SingleStatement]].parsedPlan (children expose only nested children)
+ * - [[CompoundBody]].handlers (not in `children`)
+ * - [[SimpleCaseStatement]].elseBody (not in `children`)
+ * Nested expression subqueries are still covered by `foreachWithSubqueries`.
+ */
+ private def foreachPlanDeep(plan: LogicalPlan)(f: LogicalPlan => Unit): Unit = {
+ plan.foreachWithSubqueries { p =>
+ f(p)
+ p match {
+ case w: UnresolvedWith =>
+ w.cteRelations.foreach { case (_, ctePlan, _) =>
+ foreachPlanDeep(ctePlan)(f)
+ }
+ case InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>
+ foreachPlanDeep(table)(f)
+ case s: SingleStatement =>
+ // Root of the wrapped statement is skipped by SingleStatement.children.
+ foreachPlanDeep(s.parsedPlan)(f)
+ case c: CompoundBody =>
+ c.handlers.foreach(h => foreachPlanDeep(h)(f))
+ case s: SimpleCaseStatement =>
+ s.elseBody.foreach(b => foreachPlanDeep(b)(f))
+ case ExplainCommand(logicalPlan, _) =>
+ foreachPlanDeep(logicalPlan)(f)
+ case DescribeQueryCommand(_, queryPlan) =>
+ foreachPlanDeep(queryPlan)(f)
+ case _ =>
+ }
+ }
+ }
+
+ /** Multipart name from a table/view-shaped plan node, if any. */
+ private def tableOrViewParts(plan: LogicalPlan): Option[Seq[String]] = plan match {
+ case u: UnresolvedRelation => Some(u.multipartIdentifier)
+ case u: UnresolvedTable => Some(u.multipartIdentifier)
+ case u: UnresolvedView => Some(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+ case u: UnresolvedIdentifier => Some(u.nameParts)
+ case _ => None
+ }
+
+ private def tableIdentifierParts(id: org.apache.spark.sql.catalyst.TableIdentifier): Seq[String] =
+ id.catalog.toSeq ++ id.database.toSeq :+ id.table
+
+ /**
+ * Collect multipart table/view identifiers for lineage (as written in the
+ * SQL). CTE definition names and correlation aliases are omitted; tables
+ * referenced inside CTE bodies are still included. Function / variable
+ * identifiers are not collected. Deduplicates while preserving first-seen
+ * order.
+ */
+ private def collectTableReferences(plan: LogicalPlan): Seq[Seq[String]] = {
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ // CTE names are always single-part in the grammar; UnresolvedRelation refs
+ // to CTEs are likewise single-part, so filtering matches that shape.
+ val cteNames = mutable.HashSet.empty[String]
+
+ def addCteNames(w: UnresolvedWith): Unit = {
+ w.cteRelations.foreach { case (name, _, _) =>
+ cteNames += name.toLowerCase(java.util.Locale.ROOT)
+ }
+ }
+
+ def isCteName(parts: Seq[String]): Boolean = parts match {
+ case Seq(name) => cteNames.contains(name.toLowerCase(java.util.Locale.ROOT))
+ case _ => false
+ }
+
+ def add(parts: Seq[String]): Unit = {
+ if (parts.nonEmpty && !isCteName(parts)) seen += parts
+ }
+
+ // First pass: gather CTE names in scope (including nested).
+ foreachPlanDeep(plan) {
+ case w: UnresolvedWith => addCteNames(w)
+ case _ =>
+ }
+
+ foreachPlanDeep(plan) {
+ case u: UnresolvedRelation => add(u.multipartIdentifier)
+ case u: UnresolvedTable => add(u.multipartIdentifier)
+ case u: UnresolvedView => add(u.multipartIdentifier)
+ case u: UnresolvedTableOrView => add(u.multipartIdentifier)
+ // Table/view DDL targets only — not CreateFunction / CreateVariable names.
Review Comment:
**Nit:**
The repository guidance requires ASCII punctuation in code comments.
```suggestion
// Table/view DDL targets only -- not CreateFunction / CreateVariable names.
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org