srielau commented on code in PR #59873: URL: https://github.com/apache/spark/pull/59873#discussion_r4467789416
########## sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala: ########## @@ -0,0 +1,414 @@ +/* + * 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 java.util.Collections +import java.util.concurrent.atomic.AtomicInteger + +import org.json4s._ +import org.json4s.jackson.JsonMethods.parse + +import org.apache.spark.sql.{AnalysisException, QueryTest} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog} +import org.apache.spark.sql.connector.catalog.procedures.{ + BoundProcedure, + ProcedureParameter, + UnboundProcedure +} +import org.apache.spark.sql.connector.read.Scan +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.{DataTypes, StructType} + +class ParseSqlAnalyzeSuite extends QueryTest with SharedSparkSession { + + private def evalJson(sql: String, mode: String = "ANALYZE"): JValue = { + val result = spark.sql(s"SELECT parse_sql($sql, '$mode')").head().getString(0) + parse(result) + } + + private def evalStatement(sql: String, mode: String = "ANALYZE"): JObject = { + evalJson("'" + sql.replace("'", "''") + "'", mode) match { + case JArray(List(value: JObject)) => value + case other => fail(s"expected one statement object, got: $other") + } + } + + private def jsonType(name: String): JObject = JObject("name" -> JString(name)) + private val jsonInt: JObject = jsonType("int") + private val jsonBigint: JObject = jsonType("bigint") + private val jsonBoolean: JObject = jsonType("boolean") + private val jsonString: JObject = JObject( + "name" -> JString("string"), + "collation" -> JString("UTF8_BINARY")) + + test("ANALYZE types selected columns and PARSE does not") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val analyzed = evalStatement("SELECT 1 AS a", "ANALYZE") + assert(analyzed \ "parse_success" === JBool(true)) + assert(analyzed \ "analyze_success" === JBool(true)) + assert(analyzed \ "select_list" === JArray(List( + JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt)))) + + val parsed = evalStatement("SELECT 1 AS a", "PARSE") + assert(parsed \ "analyze_success" === JNothing) + assert(parsed \ "select_list" === JArray(List( + JObject("name" -> JArray(List(JString("a"))))))) + } + } + + test("ANALYZE fully qualifies tables and builtin routines") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_analyze_t") { + sql("CREATE TABLE parse_sql_analyze_t (a INT, b STRING) USING parquet") + val j = evalStatement( + "SELECT count(a) AS c FROM parse_sql_analyze_t", "analyze") + assert(j \ "analyze_success" === JBool(true)) + assert(j \ "source_table_references" === JArray(List(JArray(List( + JString("spark_catalog"), JString("default"), JString("parse_sql_analyze_t")))))) + assert(j \ "function_references" === JArray(List(JArray(List( + JString("system"), JString("builtin"), JString("count")))))) + assert(j \ "select_list" === JArray(List( + JObject("name" -> JArray(List(JString("c"))), "type" -> jsonBigint)))) + } + } + } + + test("ANALYZE qualifies local temp views as system.session") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTempView("parse_sql_analyze_v") { + sql("CREATE OR REPLACE TEMP VIEW parse_sql_analyze_v AS SELECT 1 AS a, 'x' AS b") + val j = evalStatement("SELECT a, b FROM parse_sql_analyze_v") + assert(j \ "analyze_success" === JBool(true)) + assert(j \ "source_table_references" === JArray(List(JArray(List( + JString("system"), JString("session"), JString("parse_sql_analyze_v")))))) + assert(j \ "select_list" === JArray(List( + JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt), + JObject("name" -> JArray(List(JString("b"))), "type" -> jsonString)))) + } + } + } + + test("ANALYZE keeps parse metadata when analysis fails") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val j = evalStatement("SELECT a FROM missing_parse_sql_analyze_t") + assert(j \ "parse_success" === JBool(true)) + assert(j \ "analyze_success" === JBool(false)) + assert(j \ "statement_identifier" === JString("SELECT")) + assert(j \ "source_table_references" === JArray(List(JArray(List( + JString("spark_catalog"), + JString("default"), + JString("missing_parse_sql_analyze_t")))))) + val errorClass = (j \ "error" \ "errorClass").asInstanceOf[JString].s + assert(errorClass == "TABLE_OR_VIEW_NOT_FOUND" || + errorClass.contains("TABLE_OR_VIEW_NOT_FOUND")) + } + } + + test("ANALYZE inherits the invoker schema") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + sql("CREATE DATABASE parse_sql_ctx_a") + sql("CREATE DATABASE parse_sql_ctx_b") + try { + sql("CREATE TABLE parse_sql_ctx_a.t (a INT) USING parquet") + sql("USE parse_sql_ctx_b") + try { + val missing = evalStatement("SELECT a FROM t") + assert(missing \ "analyze_success" === JBool(false)) + assert(missing \ "source_table_references" === JArray(List(JArray(List( + JString("spark_catalog"), JString("parse_sql_ctx_b"), JString("t")))))) + } finally { + sql("USE parse_sql_ctx_a") + } + try { + val found = evalStatement("SELECT a FROM t") + assert(found \ "analyze_success" === JBool(true)) + assert(found \ "source_table_references" === JArray(List(JArray(List( + JString("spark_catalog"), JString("parse_sql_ctx_a"), JString("t")))))) + assert(spark.sql("SELECT current_schema()").head().getString(0) === + "parse_sql_ctx_a") + } finally { + sql("USE default") + } + } finally { + sql("DROP TABLE IF EXISTS parse_sql_ctx_a.t") + sql("DROP DATABASE IF EXISTS parse_sql_ctx_a CASCADE") + sql("DROP DATABASE IF EXISTS parse_sql_ctx_b CASCADE") + } + } + } + + test("ANALYZE named mode argument") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val json = spark.sql( + """SELECT parse_sql( + | 'SELECT 1 AS a', + | mode => 'ANALYZE')""".stripMargin).head().getString(0) + val analyzed = parse(json) match { + case JArray(List(value: JObject)) => value + case other => fail(s"expected one statement object, got: $other") + } + assert(analyzed \ "analyze_success" === JBool(true)) + assert(analyzed \ "select_list" === JArray(List( + JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt)))) + } + } + + test("ANALYZE without parameter marker types leaves the query unresolved") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val json = spark.sql("SELECT parse_sql('SELECT abs(?) AS x', 'ANALYZE')") + .head().getString(0) + val analyzed = parse(json) match { + case JArray(List(value: JObject)) => value + case other => fail(s"expected one statement object, got: $other") + } + assert(analyzed \ "parse_success" === JBool(true)) + assert(analyzed \ "analyze_success" === JBool(false)) + } + } + + private def evalBatch(sql: String): List[JObject] = + evalJson("'" + sql.replace("'", "''") + "'") match { + case JArray(values) => values.map(_.asInstanceOf[JObject]) + case other => fail(s"expected a JSON array, got: $other") + } + + private def typedColumns(columns: (String, JObject)*): JArray = JArray(columns.map { + case (name, dataType) => JObject("name" -> JArray(List(JString(name))), "type" -> dataType) + }.toList) + + private def ref(parts: String*): JArray = JArray(parts.map(JString).toList) + + /** Run a full `SELECT parse_sql(...)` and return its statement objects. */ + private def evalCall(query: String): List[JObject] = + parse(spark.sql(query).head().getString(0)) match { + case JArray(values) => values.map(_.asInstanceOf[JObject]) + case other => fail(s"expected a JSON array, got: $other") + } + + private def errorClass(statement: JObject): JValue = statement \ "error" \ "errorClass" + + test("ANALYZE does not run CALL, directly or through EXECUTE IMMEDIATE") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + spark.conf.set("spark.sql.catalog.parse_sql_cat", classOf[InMemoryCatalog].getName) + try { + val catalog = spark.sessionState.catalogManager.catalog("parse_sql_cat") + .asInstanceOf[InMemoryCatalog] + catalog.createProcedure( + Identifier.of(Array("ns"), "record"), ParseSqlAnalyzeSuite.UnboundRecord) + ParseSqlAnalyzeSuite.calls.set(0) + val results = evalBatch( + """CALL parse_sql_cat.ns.record(1); + |EXECUTE IMMEDIATE 'CALL parse_sql_cat.ns.record(2)'""".stripMargin) + assert(results.map(_ \ "parse_success") === List(JBool(true), JBool(true))) + assert(results.map(_ \ "analyze_success") === List(JBool(true), JBool(true))) + assert(ParseSqlAnalyzeSuite.calls.get === 0) + sql("CALL parse_sql_cat.ns.record(3)").collect() + assert(ParseSqlAnalyzeSuite.calls.get === 1) + } finally { + spark.sessionState.catalogManager.reset() + spark.sessionState.conf.unsetConf("spark.sql.catalog.parse_sql_cat") + } + } + } + + test("ANALYZE inside a SQL script does not see the script's locals") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val json = spark.sql( + """BEGIN + | DECLARE parse_sql_outer INT DEFAULT 5; + | SELECT parse_sql('SELECT parse_sql_outer AS x', 'ANALYZE'); + |END""".stripMargin).head().getString(0) + val statement = parse(json) match { + case JArray(List(value: JObject)) => value + case other => fail(s"expected one statement object, got: $other") + } + assert(statement \ "analyze_success" === JBool(false)) + val cls = (statement \ "error" \ "errorClass").asInstanceOf[JString].s + assert(cls === "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION" || + cls === "UNRESOLVED_COLUMN.WITH_SUGGESTION" || + cls.startsWith("UNRESOLVED_COLUMN")) + } + } + + test("ANALYZE requires constant arguments; PARSE does not") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val query = "SELECT parse_sql(c, '%s') FROM VALUES ('SELECT 1') AS t(c)" + val e = intercept[AnalysisException](spark.sql(query.format("ANALYZE")).collect()) + assert(e.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT") + assert(spark.sql(query.format("PARSE")).collect().length === 1) + val modeCol = intercept[AnalysisException]( + spark.sql("SELECT parse_sql(c, m) FROM VALUES ('SELECT 1', 'PARSE') AS t(c, m)") + .collect()) + assert(modeCol.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT") + } + } + + test("ANALYZE does not execute SQL scripts") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_script_t", "parse_sql_script_u") { + sql("CREATE TABLE parse_sql_script_t (a INT) USING parquet") + sql("CREATE TABLE parse_sql_script_u (a INT) USING parquet") + val schema = spark.sql("SELECT current_schema()").head().getString(0) + val j = evalStatement( + """BEGIN + | INSERT INTO parse_sql_script_t VALUES (1); + | DROP TABLE parse_sql_script_u; + |END""".stripMargin) + assert(j \ "parse_success" === JBool(true)) + assert(j \ "analyze_success" === JBool(false)) + assert(errorClass(j) === JString("PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT")) + assert(spark.table("parse_sql_script_t").count() === 0) + assert(spark.catalog.tableExists("parse_sql_script_u")) + assert(spark.sql("SELECT current_schema()").head().getString(0) === schema) + } + } + } + + test("ANALYZE reports range as system.builtin and de-duplicates count") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_dedup_t") { + sql("CREATE TABLE parse_sql_dedup_t (a INT, b INT) USING parquet") + val rangeStmt = evalStatement("SELECT * FROM range(3)") + assert(rangeStmt \ "analyze_success" === JBool(true)) + assert(rangeStmt \ "function_references" === JArray(List( + ref("system", "builtin", "range")))) + val counts = evalStatement( + "SELECT count(a), COUNT(b) FROM parse_sql_dedup_t") + assert(counts \ "analyze_success" === JBool(true)) + assert(counts \ "function_references" === JArray(List( + ref("system", "builtin", "count")))) + } + } + } + + test("ANALYZE qualifies DDL targets by statement kind") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val tempView = evalStatement("CREATE TEMPORARY VIEW parse_sql_ddl_v AS SELECT 1 AS a") + assert(tempView \ "analyze_success" === JBool(true)) + assert(tempView \ "target_table_references" === JArray(List( + ref("system", "session", "parse_sql_ddl_v")))) + assert(!spark.catalog.tableExists("parse_sql_ddl_v")) + withTempView("parse_sql_ddl_t") { + sql("CREATE TEMP VIEW parse_sql_ddl_t AS SELECT 1 AS a") + val table = evalStatement("CREATE TABLE parse_sql_ddl_t (a INT) USING parquet") + assert(table \ "analyze_success" === JBool(true)) + assert(table \ "target_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ddl_t")))) + } + } + } + + test("ANALYZE types session-catalog CTAS query columns") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val j = evalStatement( + "CREATE TABLE parse_sql_ctas_t USING parquet AS SELECT 1 AS a") + assert(j \ "analyze_success" === JBool(true)) + assert(j \ "select_list" === typedColumns("a" -> jsonInt)) + assert(!spark.catalog.tableExists("parse_sql_ctas_t")) + } + } + + test("ANALYZE does not execute a mixed DDL/DML batch") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_batch_t") { + sql("CREATE TABLE parse_sql_batch_t (a INT) USING parquet") + val results = evalBatch( + """INSERT INTO parse_sql_batch_t VALUES (1); + |CREATE TABLE parse_sql_batch_u (a INT) USING parquet; Review Comment: Added in e19b9d04cf1: per-statement isolation (USE / CREATE TEMP VIEW not seen later, failing statement then a passing one), EXECUTE IMMEDIATE DML, `PARSE_SQL_ANALYZE_REQUIRES_SESSION` without a session, and `withTable` around the CTAS / mixed-batch names. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala: ########## @@ -18,43 +18,71 @@ 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.InternalRow +import org.apache.spark.sql.catalyst.analysis.{ + ExpressionBuilder, FunctionRegistry, FunctionRegistryBase, TypeCheckResult +} +import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback -import org.apache.spark.sql.catalyst.parser.ParseSqlResult +import org.apache.spark.sql.catalyst.parser.{ParseSqlMode, ParseSqlResult} +import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature, InputParameter} +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase, QueryExecutionErrors} 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 batch string and returns a compact JSON array describing its - * unresolved statements (source position, identifier/code, lineage references, - * select-list names, parameters). A statement that does not parse is represented - * by a STANDARD-format error object at its position in the array. + * statements (source position, identifier/code, lineage references, + * select-list names, parameters). `PARSE` mode is unresolved. `ANALYZE` + * mode also types the select list and fully qualifies resolved relations + * and routines. A statement that does not parse is represented by a + * STANDARD-format error object at its position in the array. * * 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. + * User-facing parse / analyze errors become JSON; unexpected internal + * failures propagate. `ANALYZE` needs an active Spark session (catalog) + * and inherits the invoker's current catalog, schema, and SQL path. */ // scalastyle:off line.size.limit // scalastyle:off nonascii @ExpressionDescription( - usage = """_FUNC_(sqlStmt) - Splits `sqlStmt` into SQL statements, parses each with - the stock Spark SQL parser, and returns a JSON array describing them (1-based - Unicode code-point `start` and code-point `length`, parse success, Table 39 - statement identifier/code, target and source table references for lineage, - select-list column names, and parameter markers). - Statement length excludes surrounding whitespace and the terminating semicolon. - Session parser extensions are not applied. - Requires spark.sql.function.parseSql.enabled=true. On syntax / parse error returns JSON - for that statement with `parse_success` false, source location, and a nested STANDARD - error object instead of throwing or stopping the remaining statements. Nested error - locations are statement-relative Unicode code points. An empty or closed-comment-only - batch returns `[]`.""", + usage = """_FUNC_(sqlStmt[, mode]) - Review Comment: Fixed in e19b9d04cf1. The usage description now starts on the same line as `) -` so the docs generator split matches, and `mode` is documented as a constant. ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -6922,6 +6927,18 @@ ], "sqlState" : "42601" }, + "PARSE_SQL_ANALYZE_REQUIRES_SESSION" : { + "message" : [ + "parse_sql ANALYZE mode requires an active Spark session with catalog access." Review Comment: Updated in e19b9d04cf1. Both classes take `<functionName>`. REQUIRES_SESSION now says ANALYZE is evaluated on the driver; UNSUPPORTED_SCRIPT points at PARSE for `BEGIN ... END`. -- 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]
