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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
