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: [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]
