srielau commented on code in PR #59873: URL: https://github.com/apache/spark/pull/59873#discussion_r4429938127
########## sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala: ########## @@ -0,0 +1,412 @@ +/* + * 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") { Review Comment: Added coverage for global temp views, DROP of a local temp view, path-based parquet tables, CREATE TABLE vs an existing temp view, and EXPLAIN / DESCRIBE QUERY select-list types. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala: ########## @@ -79,6 +79,20 @@ private[sql] trait CatalogManager extends SQLConfHelper with Logging { // ---- Namespace ---- def currentNamespace: Array[String] def setCurrentNamespace(namespace: Array[String]): Unit + /** + * Explicit current catalog name, or `None` if the session still follows + * `spark.sql.defaultCatalog`. Used to restore ANALYZE overrides without + * materializing the implicit default as a stored catalog. + */ + private[sql] def storedCurrentCatalogName: Option[String] Review Comment: Reverted. CatalogManager has no ANALYZE-specific API in this PR. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -64,70 +122,325 @@ object ParseSqlResult { ThreadLocal.withInitial(() => new SparkSqlParser()) /** Parse `sql` as a batch and render the JSON result array. */ - def fromSql(sql: String): String = { + def fromSql( + sql: String, + mode: String = ParseSqlMode.Parse, + context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String = { val split = parser.get().splitStatementsWithPositions(sql) val statements = split.completeStatements ++ split.partialStatement - compact(render(JArray(statements.map(parseStatement).toList))) + val spans = statementSpans(sql, statements) + val json = if (mode == ParseSqlMode.Analyze) { + analyzeBatch(sql, statements, spans, context) + } else { + JArray(statements.zip(spans).map { case (segment, span) => + parsePlanOrError(segment, span) match { + case ParseFailed(errorObj) => errorObj + case Parsed(_, plan) => + successJson(plan, analyzed = None, analyzeError = None, None, span) + } + }.toList) + } + compact(render(json)) + } + + private sealed trait ParsedStatement + private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan) + extends ParsedStatement + private case class ParseFailed(errorObj: JObject) extends ParsedStatement + + /** + * Parse every statement, then analyze each with optional context overrides. + * Statements are analyzed against the session as it is, without being + * executed, so one does not see the effect of an earlier one. + * Analysis / context failures stay JSON (`analyze_success` false). + * After a context override failure, references stay unqualified parse names + * rather than the invoker or the attempted context. + */ + private def analyzeBatch( + batch: String, + statements: Seq[PositionedSqlStatement], + spans: Seq[StatementSpan], + context: ParseSqlAnalyzeContext): JArray = { + val parsed = statements.zip(spans).map { case (segment, span) => + parsePlanOrError(segment, span) + } + val session = ClassicSparkSession.getActiveSession + .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession()) + // The batch is analyzed on its own: a SQL script that calls parse_sql does + // not lend it its locals, and CALL is not run. + SqlScriptingContextManager.create(null).runWith { + InvokeProcedures.withoutInvocation { + try { + withAnalyzeContext(session, context) { + JArray(parsed.zip(spans).map { + case (ParseFailed(errorObj), _) => errorObj + case (Parsed(_, plan), span) => + val (analyzed, error) = analyze(session, plan) + successJson(plan, analyzed, error, Some(session), span) + }.toList) + } + } catch { + case e: SparkThrowable with Throwable if !isInternalError(e) => + JArray(parsed.zip(spans).map { + case (ParseFailed(errorObj), _) => errorObj + case (Parsed(_, plan), span) => + successJson(plan, None, Some(e), Some(session), span, qualify = false) + }.toList) + } + } + } + } + + /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the call. */ + private def isInternalError(e: SparkThrowable): Boolean = + Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR")) + + /** Character span of a statement object in the original batch. */ + private case class StatementSpan(start: Int, length: Int) + + /** + * Code-point `start` for each statement in one forward pass over `batch`. + * Statements are in source order, so prefix counts are not repeated. + */ + private def statementSpans( + batch: String, + statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = { + var lastUtf16 = 0 + var lastCp = 0 + statements.map { segment => + val utf16 = math.max(0, math.min(segment.start, batch.length)) + if (utf16 >= lastUtf16) { + lastCp += batch.codePointCount(lastUtf16, utf16) + lastUtf16 = utf16 + } else { + lastCp = batch.codePointCount(0, utf16) + lastUtf16 = utf16 + } + StatementSpan(lastCp + 1, charLength(segment)) + } } - private def parseStatement(segment: PositionedSqlStatement): JObject = { + /** Parsed plan, or parse-failure JSON for this statement. */ + private def parsePlanOrError( + segment: PositionedSqlStatement, + span: StatementSpan): ParsedStatement = { val sql = segment.statement try { // Do not inherit the outer query's origin from the parse_sql expression. // Errors and parsed nodes refer to the individual statement. val origin = if (sql.nonEmpty) { - Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql)) + Origin( + startIndex = Some(0), + stopIndex = Some(sql.codePointCount(0, sql.length) - 1), Review Comment: Origin stopIndex is now sql.length - 1 (UTF-16) until SPARK-60672. JSON start / length remain Unicode code points. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -407,19 +720,167 @@ object ParseSqlResult { named.map(selectListItem) } + /** Select-list items with SQL types from an analyzed query's output. */ + private def collectTypedSelectList( + analyzed: LogicalPlan, + parsed: LogicalPlan): Seq[JObject] = { + val query = primaryQueryPlan(analyzed) + if (query.resolved && query.output.nonEmpty) { + query.output.map { attr => + JObject( + "name" -> partsToJArray(Seq(attr.name)), + "type" -> DescribeRelationJsonCommand.jsonType( + CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType))) + } + } else { + collectSelectList(parsed) + } + } + + private def uniqueIdentities( + ids: Seq[Seq[String]], + session: ClassicSparkSession): Seq[Seq[String]] = { + val sensitive = session.sessionState.conf.caseSensitiveAnalysis + val seen = mutable.LinkedHashSet.empty[Seq[String]] + ids.filter { parts => + val key = if (sensitive) parts else parts.map(_.toLowerCase(Locale.ROOT)) + seen.add(key) + } + } + + private def sessionIdent(name: String): Seq[String] = + Seq(CatalogManager.SYSTEM_CATALOG_NAME, CatalogManager.SESSION_NAMESPACE, name) + + private def isTempViewCreate(plan: LogicalPlan): Boolean = plan match { + case c: CreateViewCommand => + c.viewType == LocalTempView || c.viewType == GlobalTempView + case _: CreateTempViewUsing => true + case _ => false + } + + private def qualifyTarget( + parts: Seq[String], + plan: LogicalPlan, + session: ClassicSparkSession, + dml: Boolean): Seq[String] = { + if (isTempViewCreate(plan)) { + sessionIdent(parts.last) + } else if (dml) { + qualifyRelation(parts, session) + } else { + qualifyWithCurrentNamespace(parts, session) + } + } + + private def qualifyRelation( + parts: Seq[String], + session: ClassicSparkSession): Seq[String] = { + val catalog = session.sessionState.catalog + if (CatalogManager.isSessionQualifiedViewName(parts) && catalog.isTempView(parts)) { + sessionIdent(parts.last) + } else { + resolvedRelationIdentity(parts, session) + .getOrElse(qualifyWithCurrentNamespace(parts, session)) + } + } + + /** + * Identity of the relation `parts` resolves to, so a name found through the + * path is reported where it was found. None when it does not resolve. + */ + private def resolvedRelationIdentity( + parts: Seq[String], + session: ClassicSparkSession): Option[Seq[String]] = { + try { + val catalog = session.sessionState.catalog + session.sessionState.analyzer.execute(UnresolvedRelation(parts)) match { + case s: SubqueryAlias => + val name = s.identifier.name + val qualifier = s.identifier.qualifier + if (catalog.isTempView(qualifier :+ name) || + (parts.size == 1 && catalog.getRawTempView(name).isDefined)) { + Some(sessionIdent(name)) + } else if (qualifier.nonEmpty) { + Some(qualifier :+ name) + } else { + None + } + case _ => None + } + } catch { + case NonFatal(_) => None + } + } + + private def qualifyRoutine( + parts: Seq[String], + session: ClassicSparkSession): Seq[String] = { + val catalog = session.sessionState.catalog + val tempFirst = + session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER) == "first" + def isTemp(name: String): Boolean = + catalog.isTemporaryFunction(FunctionIdentifier(name)) + def isBuiltin(name: String): Boolean = { + val id = FunctionIdentifier(name) + FunctionRegistry.builtin.functionExists(id) || + TableFunctionRegistry.builtin.functionExists(id) + } + def builtinIdent(name: String): Seq[String] = + FunctionRegistry.builtinFunctionIdentifier(name).nameParts + parts match { + case Seq(name) if tempFirst && isTemp(name) => + sessionIdent(name) + case Seq(name) if isBuiltin(name) => + builtinIdent(name) + case Seq(name) if isTemp(name) => + sessionIdent(name) + case Seq(ns, name) if isSystemNamespace(ns) => + Seq(CatalogManager.SYSTEM_CATALOG_NAME, ns.toLowerCase(Locale.ROOT), name) + case Seq(cat, ns, name) + if cat.equalsIgnoreCase(CatalogManager.SYSTEM_CATALOG_NAME) && + isSystemNamespace(ns) => + Seq(CatalogManager.SYSTEM_CATALOG_NAME, ns.toLowerCase(Locale.ROOT), name) + case other => + qualifyWithCurrentNamespace(other, session) + } + } + + private def qualifyWithCurrentNamespace( + parts: Seq[String], + session: ClassicSparkSession): Seq[String] = { + val mgr = session.sessionState.catalogManager + val catalogName = mgr.currentCatalog.name() + val namespace = mgr.currentNamespace.toSeq + parts match { + case Seq(name) => Seq(catalogName) ++ namespace :+ name + case other if mgr.isCatalogRegistered(other.head) => other + case other => catalogName +: other + } + } + + private def isSystemNamespace(ns: String): Boolean = + ns.equalsIgnoreCase(CatalogManager.BUILTIN_NAMESPACE) || + ns.equalsIgnoreCase(CatalogManager.SESSION_NAMESPACE) + private def primaryQueryPlan(plan: LogicalPlan): LogicalPlan = plan match { case UnresolvedWith(child, _, _) => primaryQueryPlan(child) + case WithCTE(child, _) => primaryQueryPlan(child) case UnresolvedInsert(_, _, _, query, _, _, _, _, _) => primaryQueryPlan(query) - case InsertIntoStatement(_, _, _, query, _, _, _, _, _) => - primaryQueryPlan(query) + case i: InsertIntoStatement => primaryQueryPlan(i.query) + case d: DataWritingCommand => primaryQueryPlan(d.query) Review Comment: primaryQueryPlan now unwraps V2WriteCommand. EXPLAIN / DESCRIBE QUERY get a typed select_list from the inner analyzed query. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -64,70 +122,325 @@ object ParseSqlResult { ThreadLocal.withInitial(() => new SparkSqlParser()) /** Parse `sql` as a batch and render the JSON result array. */ - def fromSql(sql: String): String = { + def fromSql( + sql: String, + mode: String = ParseSqlMode.Parse, + context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String = { val split = parser.get().splitStatementsWithPositions(sql) val statements = split.completeStatements ++ split.partialStatement - compact(render(JArray(statements.map(parseStatement).toList))) + val spans = statementSpans(sql, statements) + val json = if (mode == ParseSqlMode.Analyze) { + analyzeBatch(sql, statements, spans, context) + } else { + JArray(statements.zip(spans).map { case (segment, span) => + parsePlanOrError(segment, span) match { + case ParseFailed(errorObj) => errorObj + case Parsed(_, plan) => + successJson(plan, analyzed = None, analyzeError = None, None, span) + } + }.toList) + } + compact(render(json)) + } + + private sealed trait ParsedStatement + private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan) + extends ParsedStatement + private case class ParseFailed(errorObj: JObject) extends ParsedStatement + + /** + * Parse every statement, then analyze each with optional context overrides. + * Statements are analyzed against the session as it is, without being + * executed, so one does not see the effect of an earlier one. + * Analysis / context failures stay JSON (`analyze_success` false). + * After a context override failure, references stay unqualified parse names + * rather than the invoker or the attempted context. + */ + private def analyzeBatch( + batch: String, + statements: Seq[PositionedSqlStatement], + spans: Seq[StatementSpan], + context: ParseSqlAnalyzeContext): JArray = { + val parsed = statements.zip(spans).map { case (segment, span) => + parsePlanOrError(segment, span) + } + val session = ClassicSparkSession.getActiveSession + .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession()) + // The batch is analyzed on its own: a SQL script that calls parse_sql does + // not lend it its locals, and CALL is not run. + SqlScriptingContextManager.create(null).runWith { + InvokeProcedures.withoutInvocation { + try { + withAnalyzeContext(session, context) { + JArray(parsed.zip(spans).map { + case (ParseFailed(errorObj), _) => errorObj + case (Parsed(_, plan), span) => + val (analyzed, error) = analyze(session, plan) + successJson(plan, analyzed, error, Some(session), span) + }.toList) + } + } catch { + case e: SparkThrowable with Throwable if !isInternalError(e) => + JArray(parsed.zip(spans).map { + case (ParseFailed(errorObj), _) => errorObj + case (Parsed(_, plan), span) => + successJson(plan, None, Some(e), Some(session), span, qualify = false) + }.toList) + } + } + } + } + + /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the call. */ + private def isInternalError(e: SparkThrowable): Boolean = + Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR")) + + /** Character span of a statement object in the original batch. */ + private case class StatementSpan(start: Int, length: Int) + + /** + * Code-point `start` for each statement in one forward pass over `batch`. + * Statements are in source order, so prefix counts are not repeated. + */ + private def statementSpans( + batch: String, + statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = { + var lastUtf16 = 0 + var lastCp = 0 + statements.map { segment => + val utf16 = math.max(0, math.min(segment.start, batch.length)) + if (utf16 >= lastUtf16) { + lastCp += batch.codePointCount(lastUtf16, utf16) + lastUtf16 = utf16 + } else { + lastCp = batch.codePointCount(0, utf16) + lastUtf16 = utf16 + } + StatementSpan(lastCp + 1, charLength(segment)) + } } - private def parseStatement(segment: PositionedSqlStatement): JObject = { + /** Parsed plan, or parse-failure JSON for this statement. */ + private def parsePlanOrError( + segment: PositionedSqlStatement, + span: StatementSpan): ParsedStatement = { val sql = segment.statement try { // Do not inherit the outer query's origin from the parse_sql expression. // Errors and parsed nodes refer to the individual statement. val origin = if (sql.nonEmpty) { - Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText = Some(sql)) + Origin( + startIndex = Some(0), + stopIndex = Some(sql.codePointCount(0, sql.length) - 1), + sqlText = Some(sql)) } else { Origin(sqlText = Some(sql)) } CurrentOrigin.withOrigin(origin) { - val plan = parser.get().parsePlan(sql) - successJson(plan, segment) + Parsed(segment, parser.get().parsePlan(sql)) } } catch { - // User-facing parse / scripting failures become JSON; everything else fails. case e: ParseException => - errorJson(e, segment) + ParseFailed(errorJson(e, span)) case e: SqlScriptingException => - errorJson(e, segment) + ParseFailed(errorJson(e, span)) + } + } + + /** Analyze `plan` without executing it. Scripts never reach QueryExecution. */ Review Comment: Time-travel still analyzes as the relation; we report the resolved table identity, not a VERSION AS OF suffix. Happy to add a JSON field later if we want the spec in the contract. ########## sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala: ########## @@ -0,0 +1,412 @@ +/* + * 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 invoker schema and can override current_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")))))) + + val json = spark.sql( + """SELECT parse_sql( + | 'SELECT a FROM t', + | 'ANALYZE', + | current_schema => 'parse_sql_ctx_a')""".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 \ "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_b") + } 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 and current_catalog inherit from invoker") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val json = spark.sql( + """SELECT parse_sql( + | 'SELECT 1 AS a', + | mode => 'ANALYZE', + | current_catalog => current_catalog())""".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 current_path override is restored on the invoker") { + withSQLConf( + SQLConf.PARSE_SQL_ENABLED.key -> "true", + SQLConf.PATH_ENABLED.key -> "true") { + sql("SET PATH = spark_catalog.default, system.builtin") + try { + val before = spark.sql("SELECT current_path()").head().getString(0) + val json = spark.sql( + """SELECT parse_sql( + | 'SELECT 1 AS a', + | 'ANALYZE', + | current_path => 'system.builtin, spark_catalog.default')""".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(spark.sql("SELECT current_path()").head().getString(0) === before) + } finally { + sql("SET PATH = DEFAULT_PATH") + } + } + } + + test("ANALYZE reports invalid current_schema as JSON instead of failing the query") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val json = spark.sql( + "SELECT parse_sql('SELECT 1 AS a', 'ANALYZE', current_schema => '!!!')") + .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)) + assert(analyzed \ "select_list" === JArray(List( + JObject("name" -> JArray(List(JString("a"))))))) + assert((analyzed \ "error" \ "errorClass").asInstanceOf[JString].s === + "INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT") + } + } + + 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 errorClass = (statement \ "error" \ "errorClass").asInstanceOf[JString].s + assert(errorClass.contains("UNRESOLVED") || errorClass.contains("PARSE_SQL_OUTER") || Review Comment: Tightened the script-locals assertion to UNRESOLVED_COLUMN*. CREATE TEMP VIEW now asserts the view does not exist after ANALYZE. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -17,24 +17,75 @@ package org.apache.spark.sql.catalyst.parser +import java.util.Locale + 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.{FunctionIdentifier, SqlScriptingContextManager, TableIdentifier} import org.apache.spark.sql.catalyst.analysis._ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, SQLQueryContext} +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.classic.{SparkSession => ClassicSparkSession} +import org.apache.spark.sql.connector.catalog.{CatalogManager, PathElement} +import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.exceptions.SqlScriptingException -import org.apache.spark.sql.execution.SparkSqlParser -import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryCommand, ExplainCommand} -import org.apache.spark.sql.execution.datasources.CreateTempViewUsing +import org.apache.spark.sql.execution.{CommandExecutionMode, SparkSqlParser} +import org.apache.spark.sql.execution.command.{ + CreateDataSourceTableAsSelectCommand, + CreateViewCommand, + DataWritingCommand, + DescribeQueryCommand, + DescribeRelationJsonCommand, + ExplainCommand, + LeafRunnableCommand +} +import org.apache.spark.sql.execution.datasources.{CreateTempViewUsing, InsertIntoDataSourceCommand} +import org.apache.spark.sql.internal.SQLConf + +/** + * Mode for [[ParseSqlResult.fromSql]]. `PARSE` is unresolved. `ANALYZE` types + * the select list and fully qualifies resolved relations and routines. + */ +object ParseSqlMode { Review Comment: Leaving ParseSqlResult in one file for this PR, and mode as a string parse. Happy to split in a follow-up once the contract settles. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala: ########## @@ -18,71 +18,172 @@ 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.{ + ParseSqlAnalyzeContext, 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. + * Optional named `current_catalog`, `current_schema`, and `current_path` + * override that context for the duration of analysis. */ // 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 - UTF-16 code-unit `start` and UTF-16 code-unit `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. An empty or closed-comment-only batch returns `[]`.""", + usage = """_FUNC_(sqlStmt[, mode[, current_catalog[, current_schema[, current_path]]]]) - Review Comment: ExpressionDescription is updated for the two-arg signature. sql-expression-schema.md and the parse-sql goldens were regenerated for the new analyzed signature. -- 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]
