srielau commented on code in PR #59873: URL: https://github.com/apache/spark/pull/59873#discussion_r4487003781
########## sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala: ########## @@ -0,0 +1,793 @@ +/* + * 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, InMemoryRowLevelOperationTableCatalog +} +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} +import org.apache.spark.util.Utils + +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" || Review Comment: Fixed in f9390c2225c. The negative checks now assert the exact error classes, TABLE_OR_VIEW_NOT_FOUND and UNRESOLVED_ROUTINE, so a wrong error cannot pass. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -489,19 +683,307 @@ object ParseSqlResult { named.map(selectListItem) } + /** + * Select-list items with SQL types from the analyzed query output, so `*` + * is expanded and names match analysis. INSERT reports the write projection + * (target columns), not the user's SELECT aliases. + */ + 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 catalogIdentParts(catalog: String, ident: Identifier): Seq[String] = + Seq(catalog) ++ ident.namespace() :+ ident.name() + + private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] = + (r.catalog, r.identifier) match { + case (Some(catalog), Some(ident)) => Some(catalogIdentParts(catalog.name(), ident)) + case _ => None + } + + private def rewriteTargetParts(table: NamedRelation): Option[Seq[String]] = table match { + case d: DataSourceV2Relation => v2RelationParts(d) + case _ => None + } + + private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] = + r.catalogTable.map { t => + val id = t.identifier + if (id.catalog.isEmpty && id.database.isDefined) { + CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id) + } else { + tableIdentifierParts(id) + } + } + + private def viewParts(v: View): Seq[String] = { + val id = v.desc.identifier + if (v.isTempView) { + val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) + if (id.database.exists(_.equalsIgnoreCase(globalTemp))) { + Seq(id.catalog.getOrElse(CatalogManager.SESSION_CATALOG_NAME)) ++ + id.database.toSeq :+ id.table + } else { + sessionIdent(id.table) + } + } else { + tableIdentifierParts(id) + } + } + + private def tempViewParts(ident: Identifier): Seq[String] = { + val parts = ident.namespace().toSeq :+ ident.name() + val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) + val ns = ident.namespace() + if (ns.length == 1 && ns.head.equalsIgnoreCase(globalTemp)) { + Seq(CatalogManager.SESSION_CATALOG_NAME, ns.head, ident.name()) + } else if (ns.isEmpty || CatalogManager.isSessionQualifiedViewName(parts)) { + sessionIdent(ident.name()) + } else { + catalogIdentParts(CatalogManager.SESSION_CATALOG_NAME, ident) + } + } + + private def globalTempIdent(name: String): Seq[String] = + Seq( + CatalogManager.SESSION_CATALOG_NAME, + SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE), + name) + + /** + * Match each parse name to an analyzed identity at most once. Longer parse + * names claim first so `spark_catalog.db.t` is not stolen by a later bare + * `t`. Equal-length matches count: a 3-part parse name is already complete. + */ + private def qualifyParseIdentities( + parseIds: Seq[Seq[String]], + analyzedIds: Seq[Seq[String]], + analysisSucceeded: Boolean, + session: ClassicSparkSession, + fallback: Seq[String] => Seq[String]): Seq[Seq[String]] = { + val remaining = analyzedIds.to(mutable.ArrayBuffer) + val assigned = Array.fill[Option[Seq[String]]](parseIds.length)(None) + parseIds.zipWithIndex.sortBy { case (parts, _) => -parts.length }.foreach { + case (parts, i) => + matchingIdentity(parts, remaining.toSeq) match { + case Some(matched) => + val matchedKey = matched.map(_.toLowerCase(Locale.ROOT)) + val idx = remaining.indexWhere { analyzed => + analyzed.map(_.toLowerCase(Locale.ROOT)) == matchedKey + } + if (idx >= 0) { + remaining.remove(idx) + } + assigned(i) = Some(matched) + case None => + val ident = + if (isPathReference(parts) || + (analysisSucceeded && isFileReference(parts, session))) { + parts + } else { + fallback(parts) + } + assigned(i) = Some(ident) + } + } + assigned.map(_.get).toSeq + } + + private def matchingIdentity( Review Comment: Not addressed in this push. Making IDENTIFIER(...) references work needs the parse walk to evaluate the argument, including session variables, and a rule for which name wins when several are present. I want to settle that design before changing the code, so I am leaving this open. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -43,170 +89,253 @@ import org.apache.spark.sql.execution.datasources.CreateTempViewUsing * executors without a session, so only the stock parser is available under * distributed eval. * - * Every statement object includes its 1-based Unicode code-point `start` in the - * original batch and its code-point `length`, excluding surrounding whitespace - * and the terminating semicolon. On success it also includes + * Every statement object includes its 1-based character `start` in the + * original batch and its character `length`, excluding surrounding + * whitespace and the terminating semicolon. Characters are Unicode code + * points (Spark SQL `length()` / UTF-8 strings): a supplementary-plane + * emoji is one character. Java `String` offsets are UTF-16 (emoji = 2); + * ANTLR `CodePointCharStream` offsets are already code points (emoji = 1). + * The splitter maps ANTLR indexes to UTF-16 for `substring`; JSON converts + * those spans back to code points. On success it also includes * `parse_success`, the statement * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional * fields (`target_table_references`, `source_table_references`, * `function_references`, `select_list`, `parameter_markers`) when empty. On * parse failure the statement object contains `parse_success: false` with * source location and a nested STANDARD-format error object, and parsing - * continues with later statements. Nested error locations are relative to the - * individual statement, while `start` is relative to the original batch; error - * `queryContext` indexes are code points as well. An - * empty or closed-comment-only batch produces an empty array. Only - * [[ParseException]] / [[SqlScriptingException]] are converted to JSON; + * continues with later statements. Nested `queryContext` is the STANDARD + * renderer output for that error. View and EXECUTE IMMEDIATE bodies keep + * their own text; `start` is relative to the original batch. An empty or + * closed-comment-only batch produces an empty array. PARSE converts + * [[ParseException]] / [[SqlScriptingException]] to JSON. + * ANALYZE also reports non-internal analysis errors as JSON; * unexpected / internal failures propagate so the function fails. + * Nested error locations are statement-relative Unicode code points. */ object ParseSqlResult { private val parser: ThreadLocal[SparkSqlParser] = 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): String = { val split = parser.get().splitStatementsWithPositions(sql) val statements = split.completeStatements ++ split.partialStatement - val starts = charStarts(sql, statements) - compact(render(JArray(statements.zip(starts).map { - case (segment, start) => parseStatement(segment, start) - }.toList))) + val spans = statementSpans(sql, statements) + val json = if (mode == ParseSqlMode.Analyze) { + analyzeBatch(statements, spans) + } 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(plan: LogicalPlan) extends ParsedStatement + private case class ParseFailed(errorObj: JObject) extends ParsedStatement + + /** + * Parse every statement, then analyze each against the invoker session. + * Statements are not executed, so one does not see the effect of an earlier + * one. Analysis failures stay JSON (`analyze_success` false) per statement. + */ + private def analyzeBatch( + statements: Seq[PositionedSqlStatement], + spans: Seq[StatementSpan]): JArray = { + val parsed = statements.zip(spans).map { case (segment, span) => + parsePlanOrError(segment, span) + } + val session = ClassicSparkSession.getActiveSession Review Comment: Not addressed in this push. The session clone gap predates this PR: CacheManager and stream sessions clone without the v2 current catalog and namespace. The options are to record the invoker's catalog and namespace at analysis time, or to document the gap with a test. Leaving this open for that decision. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -489,19 +683,307 @@ object ParseSqlResult { named.map(selectListItem) } + /** + * Select-list items with SQL types from the analyzed query output, so `*` + * is expanded and names match analysis. INSERT reports the write projection + * (target columns), not the user's SELECT aliases. + */ + 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 catalogIdentParts(catalog: String, ident: Identifier): Seq[String] = + Seq(catalog) ++ ident.namespace() :+ ident.name() + + private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] = + (r.catalog, r.identifier) match { + case (Some(catalog), Some(ident)) => Some(catalogIdentParts(catalog.name(), ident)) + case _ => None + } + + private def rewriteTargetParts(table: NamedRelation): Option[Seq[String]] = table match { + case d: DataSourceV2Relation => v2RelationParts(d) + case _ => None + } + + private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] = + r.catalogTable.map { t => + val id = t.identifier + if (id.catalog.isEmpty && id.database.isDefined) { + CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id) + } else { + tableIdentifierParts(id) + } + } + + private def viewParts(v: View): Seq[String] = { Review Comment: Not addressed in this push. The identity comes from viewParts, which truncates nested namespaces. The fix is to use CatalogTable.fullIdent in viewParts, HiveTableRelation, and logicalRelationParts, and to check spark.sql.legacy.v1IdentifierNoCatalog. That needs its own tests, so I am deferring it and leaving this open. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -342,10 +468,20 @@ object ParseSqlResult { visitPlan(cts.plan, nextScope, TableRefRole.Source)(f) case _: CacheTable => // Name is taken from multipartIdentifier or the non-child `table` slot. + case _: View | _: ResolvedMetricView => Review Comment: Partly addressed in f9390c2225c. Tests now cover the MERGE source attribution, the table-function body, the relative path, the sessionOrder=last owner, and the insert-only MERGE and row-level subquery cases. There is still no direct test for the isFileReference v2-table case or for isKnownCatalogName. Leaving this open until those are added. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -43,170 +89,253 @@ import org.apache.spark.sql.execution.datasources.CreateTempViewUsing * executors without a session, so only the stock parser is available under * distributed eval. * - * Every statement object includes its 1-based Unicode code-point `start` in the - * original batch and its code-point `length`, excluding surrounding whitespace - * and the terminating semicolon. On success it also includes + * Every statement object includes its 1-based character `start` in the + * original batch and its character `length`, excluding surrounding + * whitespace and the terminating semicolon. Characters are Unicode code + * points (Spark SQL `length()` / UTF-8 strings): a supplementary-plane + * emoji is one character. Java `String` offsets are UTF-16 (emoji = 2); + * ANTLR `CodePointCharStream` offsets are already code points (emoji = 1). + * The splitter maps ANTLR indexes to UTF-16 for `substring`; JSON converts + * those spans back to code points. On success it also includes * `parse_success`, the statement * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional * fields (`target_table_references`, `source_table_references`, * `function_references`, `select_list`, `parameter_markers`) when empty. On * parse failure the statement object contains `parse_success: false` with * source location and a nested STANDARD-format error object, and parsing - * continues with later statements. Nested error locations are relative to the - * individual statement, while `start` is relative to the original batch; error - * `queryContext` indexes are code points as well. An - * empty or closed-comment-only batch produces an empty array. Only - * [[ParseException]] / [[SqlScriptingException]] are converted to JSON; + * continues with later statements. Nested `queryContext` is the STANDARD + * renderer output for that error. View and EXECUTE IMMEDIATE bodies keep + * their own text; `start` is relative to the original batch. An empty or + * closed-comment-only batch produces an empty array. PARSE converts + * [[ParseException]] / [[SqlScriptingException]] to JSON. + * ANALYZE also reports non-internal analysis errors as JSON; * unexpected / internal failures propagate so the function fails. + * Nested error locations are statement-relative Unicode code points. */ object ParseSqlResult { private val parser: ThreadLocal[SparkSqlParser] = 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): String = { val split = parser.get().splitStatementsWithPositions(sql) val statements = split.completeStatements ++ split.partialStatement - val starts = charStarts(sql, statements) - compact(render(JArray(statements.zip(starts).map { - case (segment, start) => parseStatement(segment, start) - }.toList))) + val spans = statementSpans(sql, statements) + val json = if (mode == ParseSqlMode.Analyze) { + analyzeBatch(statements, spans) + } 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(plan: LogicalPlan) extends ParsedStatement + private case class ParseFailed(errorObj: JObject) extends ParsedStatement + + /** + * Parse every statement, then analyze each against the invoker session. + * Statements are not executed, so one does not see the effect of an earlier + * one. Analysis failures stay JSON (`analyze_success` false) per statement. + */ + private def analyzeBatch( + statements: Seq[PositionedSqlStatement], + spans: Seq[StatementSpan]): 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 { + 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) + } + } + } + + /** Analyze `plan`, including nested EXPLAIN / DESCRIBE QUERY payloads. */ + private def analyzePlan( + session: ClassicSparkSession, + plan: LogicalPlan): LogicalPlan = { + val toAnalyze = plan match { + case s: SupervisingCommand => + s.withTransformedSupervisedPlan(inner => analyzePlan(session, inner)) + case other => other + } + session.sessionState.executePlan(toAnalyze, CommandExecutionMode.SKIP).analyzed + } + + /** INTERNAL_ERROR / SQLSTATE XX are Spark bugs, not analysis results. */ + private def isInternalError(e: SparkThrowable): Boolean = + e.isInternalError || Option(e.getSqlState).exists(_.startsWith("XX")) + + /** 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, charStart: Int): JObject = { + /** Parsed plan, or parse-failure JSON for this statement. */ + private def parsePlanOrError( + segment: PositionedSqlStatement, + span: StatementSpan): ParsedStatement = { val sql = segment.statement - val length = sql.codePointCount(0, sql.length) try { // Do not inherit the outer query's origin from the parse_sql expression. - // Errors and parsed nodes refer to the individual statement. Parser - // origins are code-point offsets, so this one is too. + // Errors and parsed nodes refer to the individual statement. val origin = if (sql.nonEmpty) { Origin( startIndex = Some(0), - stopIndex = Some(length - 1), + stopIndex = Some(span.length - 1), sqlText = Some(sql)) } else { Origin(sqlText = Some(sql)) } CurrentOrigin.withOrigin(origin) { - val plan = parser.get().parsePlan(sql) - successJson(plan, charStart, length) + Parsed(parser.get().parsePlan(sql)) } } catch { - // User-facing parse / scripting failures become JSON; everything else fails. case e: ParseException => - errorJson(e, segment, charStart, length) + ParseFailed(errorJson(e, span)) case e: SqlScriptingException => - errorJson(e, segment, charStart, length) + ParseFailed(errorJson(e, span)) + } + } + + /** Analyze `plan` without executing it. Scripts never reach QueryExecution. */ + private def analyze( + session: ClassicSparkSession, + plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with Throwable]) = { + plan match { + case _: CompoundBody => + (None, Some(QueryExecutionErrors.parseSqlAnalyzeUnsupportedScript())) + case _ => + try { + (Some(analyzePlan(session, plan)), None) + } catch { + case e: SparkThrowable with Throwable if !isInternalError(e) => (None, Some(e)) + } } } /** Build success JSON from an already-parsed unresolved plan. */ private def successJson( plan: LogicalPlan, - charStart: Int, - length: Int): JObject = { + analyzed: Option[LogicalPlan], + analyzeError: Option[SparkThrowable with Throwable], + session: Option[ClassicSparkSession], + span: StatementSpan): JObject = { val classification = SqlStatementCodes.classify(plan) val fields = mutable.ListBuffer.empty[JField] - fields += "start" -> JInt(charStart) - fields += "length" -> JInt(length) + fields += "start" -> JInt(span.start) + fields += "length" -> JInt(span.length) fields += "parse_success" -> JBool(true) fields += "statement_identifier" -> JString(classification.statementIdentifier) fields += "statement_code" -> JInt(classification.statementCode) + if (session.nonEmpty) { + fields += "analyze_success" -> JBool(analyzed.isDefined) + } // Omit unused collections / markers so consumers can treat absence as empty. - val refs = collectPlanReferences(plan) - if (refs.targetTables.nonEmpty) { + // Parse names decide which references exist and in which role. ANALYZE + // qualifies those names from the analyzed plan when it can match them. + val parseRefs = collectPlanReferences(plan) + val analyzedRefs = analyzed.map(collectPlanReferences) + val (targetTables, sourceTables, functions) = try { + session.map { s => + val analyzedTargets = analyzedRefs.map(_.targetTables).getOrElse(Nil) + val analyzedSources = analyzedRefs.map(_.sourceTables).getOrElse(Nil) + val analysisSucceeded = analyzed.isDefined + ( + uniqueIdentities( + qualifyParseIdentities( + parseRefs.targetTables, analyzedTargets, analysisSucceeded, s, + qualifyTarget(_, plan, s)), + s), + uniqueIdentities( + qualifyParseIdentities( + parseRefs.sourceTables, analyzedSources, analysisSucceeded, s, + qualifySource(_, s)), + s), + uniqueIdentities(parseRefs.functions.map(qualifyRoutine(_, s)), s) + ) + }.getOrElse( + (parseRefs.targetTables, parseRefs.sourceTables, parseRefs.functions)) + } catch { + case e: SparkThrowable with Throwable if !isInternalError(e) => Review Comment: Not addressed in this push. The catch still reverts all three lists when one lookup fails. A per-name fallback means wrapping each lookup separately, and I want a test that makes one probe throw before changing it. Leaving this open. ########## sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala: ########## @@ -0,0 +1,793 @@ +/* + * 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, InMemoryRowLevelOperationTableCatalog +} +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} +import org.apache.spark.util.Utils + +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") { + withTable("parse_sql_ctas_t") { + 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", "parse_sql_batch_u") { + 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; + |DROP TABLE parse_sql_batch_t""".stripMargin) + assert(results.map(_ \ "analyze_success") === + List(JBool(true), JBool(true), JBool(true))) + assert(spark.table("parse_sql_batch_t").count() === 0) + assert(!spark.catalog.tableExists("parse_sql_batch_u")) + assert(spark.catalog.tableExists("parse_sql_batch_t")) + } + } + } + + test("ANALYZE qualifies global temp views and DROP of local temp views") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withGlobalTempView("parse_sql_global_v") { + sql("CREATE OR REPLACE GLOBAL TEMP VIEW parse_sql_global_v AS SELECT 1 AS a") + val globalDb = spark.conf.get("spark.sql.globalTempDatabase") + val j = evalStatement(s"SELECT a FROM $globalDb.parse_sql_global_v") + assert(j \ "analyze_success" === JBool(true)) + assert(j \ "source_table_references" === JArray(List( + ref("spark_catalog", globalDb, "parse_sql_global_v")))) + } + withTempView("parse_sql_drop_v") { + sql("CREATE TEMP VIEW parse_sql_drop_v AS SELECT 1 AS a") + val drop = evalStatement("DROP VIEW parse_sql_drop_v") + assert(drop \ "analyze_success" === JBool(true)) + assert(drop \ "target_table_references" === JArray(List( + ref("system", "session", "parse_sql_drop_v")))) + assert(spark.catalog.tableExists("parse_sql_drop_v")) + } + } + } + + test("ANALYZE types EXPLAIN and DESCRIBE QUERY inner plans") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_explain_t") { + sql("CREATE TABLE parse_sql_explain_t (a INT) USING parquet") + val explain = evalStatement("EXPLAIN SELECT a FROM parse_sql_explain_t") + assert(explain \ "analyze_success" === JBool(true)) + assert(explain \ "select_list" === typedColumns("a" -> jsonInt)) + assert(explain \ "source_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_explain_t")))) + val missing = evalStatement("EXPLAIN SELECT nope FROM missing_t") + assert(missing \ "analyze_success" === JBool(false)) + val missingClass = (missing \ "error" \ "errorClass").asInstanceOf[JString].s + assert(missingClass == "TABLE_OR_VIEW_NOT_FOUND" || + missingClass.contains("TABLE_OR_VIEW_NOT_FOUND")) + val nested = evalStatement("EXPLAIN EXPLAIN SELECT nope FROM missing_t") + assert(nested \ "analyze_success" === JBool(false)) + } + val describe = evalStatement("DESCRIBE QUERY SELECT 1 AS a") + assert(describe \ "analyze_success" === JBool(true)) + assert(describe \ "select_list" === typedColumns("a" -> jsonInt)) + } + } + + test("ANALYZE reports path-based datasource tables as format plus path") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(1).write.mode("overwrite").parquet(path) + val j = evalStatement(s"SELECT id FROM parquet.`$path`") + assert(j \ "analyze_success" === JBool(true)) + assert(j \ "source_table_references" === JArray(List(ref("parquet", path)))) + } + } + } + + test("ANALYZE unknown functions stay JSON rather than INTERNAL_ERROR") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val j = evalStatement("SELECT no_such_parse_sql_fn(1)") + assert(j \ "parse_success" === JBool(true)) + assert(j \ "analyze_success" === JBool(false)) + val cls = (j \ "error" \ "errorClass").asInstanceOf[JString].s + assert(cls == "UNRESOLVED_ROUTINE" || cls.contains("UNRESOLVED_ROUTINE")) + } + } + + test("ANALYZE keeps distinct tables that share a name suffix") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + sql("CREATE DATABASE parse_sql_zzdb") + try { + withTable("parse_sql_zzdb.t", "t") { + sql("CREATE TABLE parse_sql_zzdb.t (a INT) USING parquet") + sql("CREATE TABLE t (a INT) USING parquet") + val join1 = evalStatement("SELECT * FROM parse_sql_zzdb.t JOIN t") + assert(join1 \ "analyze_success" === JBool(true)) + assert(join1 \ "source_table_references" === JArray(List( + ref("spark_catalog", "parse_sql_zzdb", "t"), + ref("spark_catalog", "default", "t")))) + val join2 = evalStatement("SELECT * FROM t JOIN parse_sql_zzdb.t") + assert(join2 \ "analyze_success" === JBool(true)) + assert(join2 \ "source_table_references" === JArray(List( + ref("spark_catalog", "default", "t"), + ref("spark_catalog", "parse_sql_zzdb", "t")))) + val join3 = evalStatement( + "SELECT * FROM spark_catalog.parse_sql_zzdb.t JOIN t") + assert(join3 \ "analyze_success" === JBool(true)) + assert(join3 \ "source_table_references" === JArray(List( + ref("spark_catalog", "parse_sql_zzdb", "t"), + ref("spark_catalog", "default", "t")))) + } + } finally { + sql("DROP DATABASE IF EXISTS parse_sql_zzdb CASCADE") + } + } + } + + test("ANALYZE persistent CTAS and CREATE VIEW ignore a same-named temp view") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTempView("parse_sql_tv") { + sql("CREATE TEMP VIEW parse_sql_tv AS SELECT 1 AS a") + val ctas = evalStatement( + "CREATE TABLE parse_sql_tv USING parquet AS SELECT 1 AS a") + assert(ctas \ "analyze_success" === JBool(true)) + assert(ctas \ "target_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_tv")))) + val view = evalStatement("CREATE VIEW parse_sql_tv AS SELECT 1 AS a") + assert(view \ "analyze_success" === JBool(true)) + assert(view \ "target_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_tv")))) + } + } + } + + test("ANALYZE expands star select lists") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_star_t") { + sql("CREATE TABLE parse_sql_star_t (a INT) USING parquet") + val j = evalStatement("SELECT * FROM parse_sql_star_t") + assert(j \ "analyze_success" === JBool(true)) + assert(j \ "select_list" === typedColumns("a" -> jsonInt)) + } + } + } + + test("ANALYZE keeps session-catalog write and DDL references") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTable("parse_sql_ref_s", "parse_sql_ref_t", "parse_sql_ref_u") { + sql("CREATE TABLE parse_sql_ref_s (a INT) USING parquet") + sql("CREATE TABLE parse_sql_ref_t (a INT) USING parquet") + val insert = evalStatement( + "INSERT INTO parse_sql_ref_t SELECT a FROM parse_sql_ref_s") + assert(insert \ "analyze_success" === JBool(true)) + assert(insert \ "target_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ref_t")))) + assert(insert \ "source_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ref_s")))) + val ctas = evalStatement( + "CREATE TABLE parse_sql_ref_u USING parquet AS SELECT a FROM parse_sql_ref_s") + assert(ctas \ "analyze_success" === JBool(true)) + assert(ctas \ "target_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ref_u")))) + assert(ctas \ "source_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ref_s")))) + val view = evalStatement( + "CREATE VIEW parse_sql_ref_v AS SELECT a FROM parse_sql_ref_s") + assert(view \ "analyze_success" === JBool(true)) + assert(view \ "target_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ref_v")))) + assert(view \ "source_table_references" === JArray(List( + ref("spark_catalog", "default", "parse_sql_ref_s")))) + assert(!spark.catalog.tableExists("parse_sql_ref_u")) + assert(!spark.catalog.tableExists("parse_sql_ref_v")) + } + } + } + + test("ANALYZE qualifies session-qualified temp view names") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + withTempView("parse_sql_sess_v") { + sql("CREATE TEMP VIEW parse_sql_sess_v AS SELECT 1 AS a") + val drop = evalStatement("DROP VIEW session.parse_sql_sess_v") + assert(drop \ "analyze_success" === JBool(true)) + assert(drop \ "target_table_references" === JArray(List( + ref("system", "session", "parse_sql_sess_v")))) + val qualified = evalStatement("DROP VIEW system.session.parse_sql_sess_v") + assert(qualified \ "analyze_success" === JBool(true)) + assert(qualified \ "target_table_references" === JArray(List( + ref("system", "session", "parse_sql_sess_v")))) + assert(spark.catalog.tableExists("parse_sql_sess_v")) + } + } + } + + test("ANALYZE does not load a misconfigured catalog while qualifying") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + spark.conf.set("spark.sql.catalog.badcat", "com.example.Missing") + try { + val results = evalBatch("SELECT 1 AS a; SELECT * FROM badcat.ns.t") + assert(results.map(_ \ "parse_success") === List(JBool(true), JBool(true))) + assert(results.head \ "analyze_success" === JBool(true)) + assert(results(1) \ "analyze_success" === JBool(false)) + } finally { + spark.conf.unset("spark.sql.catalog.badcat") + } + } + } + + test("ANALYZE statements do not see earlier batch side effects") { + withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") { + val schema = spark.sql("SELECT current_schema()").head().getString(0) + sql("CREATE DATABASE IF NOT EXISTS parse_sql_iso_db") + try { + val results = evalBatch( + """USE parse_sql_iso_db; Review Comment: Not addressed in this push. The USE leakage check and the routine-case behavior tests still have no failure signal. Leaving this open. -- 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]
