cloud-fan commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4474401323
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +652,274 @@ 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 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]],
+ 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)) parts else fallback(parts)
+ assigned(i) = Some(ident)
+ }
+ }
+ assigned.map(_.get).toSeq
+ }
+
+ private def matchingIdentity(
+ parseParts: Seq[String],
+ analyzedIds: Seq[Seq[String]]): Option[Seq[String]] = {
+ val parsed = parseParts.map(_.toLowerCase(Locale.ROOT))
+ analyzedIds.find { analyzed =>
Review Comment:
**Blocking (P1):** This takes the first analyzed identity whose parts end
with the written name, but the analyzed pool also holds relations the statement
never names. For a MERGE into a table with row-level operations,
RewriteMergeIntoTable re-reads the target as the left side of the join inside
ReplaceData or WriteDelta, and the V2WriteCommand arm walks that query as a
source. In `MERGE INTO cat.prod.t AS tgt USING t AS src ON tgt.id = src.id WHEN
MATCHED THEN DELETE` under `USE cat.staging`, the code therefore matches the
source `t` to `cat.prod.t`, and the MERGE is reported as reading its own target
instead of `cat.staging.t`. Relations inside SQL table function and SQL scalar
function bodies are walked the same way, so in `SELECT * FROM my_tvf() JOIN t`
the user's `t` can be reported as the table the function body reads.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59873#discussion_r4460514129).
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.optimizer
+
+import org.apache.spark.sql.catalyst.expressions.{Literal, ParseSql}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Evaluates constant `parse_sql` calls on the driver. ANALYZE needs the
+ * session catalog, so this must run even when `ConstantFolding` is excluded.
+ */
+object EvaluateParseSql extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan =
+ plan.transformAllExpressionsWithSubqueries {
Review Comment:
**Non-blocking (P2):** transformAllExpressionsWithSubqueries does no
tree-pattern pruning, and this batch is unconditional and cannot be excluded,
so every optimized query pays one more walk over all its nodes, expressions and
subquery plans. That includes the default
spark.sql.function.parseSql.enabled=false, where no ParseSql can exist. Results
are unchanged, but the cost grows with wide, deep and subquery-heavy plans.
Adding a PARSE_SQL tree pattern to ParseSql's nodePatterns and switching to
transformDownWithSubqueriesAndPruning plus transformAllExpressionsWithPruning
on it, as ComputeCurrentTime and ReplaceCurrentLike do, skips plans without
parse_sql. SubqueryExpression already carries its plan's pattern bits, so calls
inside subqueries are still folded.
**Verification:**
- **Behavior:** With ConstantFolding excluded, a constant ANALYZE call at
top level and inside a scalar subquery is still evaluated on the driver.
- **Inspection:** The rule's traversal is pruned on the parse_sql tree
pattern, so a plan without parse_sql is returned without visiting its
expressions.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -43,170 +85,251 @@ 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(segment: PositionedSqlStatement, plan: LogicalPlan)
Review Comment:
**Nit (P3):** The segment is never read: both matches destructure `Parsed(_,
plan)`, and `Parsed` is private to this object. `Parsed(plan)` would drop the
unused state.
**Verification:**
- **Compile:** The parse and ANALYZE paths build Parsed with the plan only,
and every match site compiles without the segment.
- **Compatibility:** PARSE and ANALYZE JSON outputs are unchanged.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -94,22 +161,84 @@ case class ParseSql(child: Expression)
"configKey" -> SQLConf.PARSE_SQL_ENABLED.key,
"configValue" -> "true"))
}
- super.checkInputDataTypes()
+ val typeCheck = super.checkInputDataTypes()
+ if (!typeCheck.isSuccess) {
+ typeCheck
+ } else {
+ if (!mode.foldable) {
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> toSQLId("mode"),
+ "inputType" -> toSQLType(mode.dataType),
+ "inputExpr" -> toSQLExpr(mode)))
+ } else {
+ val parsedMode = Option(mode.eval()).map { value =>
+ val modeStr = value.asInstanceOf[UTF8String].toString
+ ParseSqlMode.parse(modeStr).getOrElse {
+ throw QueryCompilationErrors.invalidParseSqlMode(mode)
+ }
+ }
+ // ANALYZE needs the session, which only the driver has, so sqlStmt
+ // must fold. PARSE may take a column of statement text.
+ if (parsedMode.contains(ParseSqlMode.Analyze) && !sqlStmt.foldable) {
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> toSQLId("sqlStmt"),
+ "inputType" -> toSQLType(sqlStmt.dataType),
+ "inputExpr" -> toSQLExpr(sqlStmt)))
+ } else {
+ TypeCheckResult.TypeCheckSuccess
+ }
+ }
+ }
}
- override def nullSafeEval(input: Any): Any = {
- val sql = input.asInstanceOf[UTF8String].toString
- UTF8String.fromString(ParseSqlResult.fromSql(sql))
+ override def eval(input: InternalRow): Any = {
+ val sqlInput = sqlStmt.eval(input)
+ if (sqlInput == null) {
+ return null
+ }
+ val modeInput = mode.eval(input)
+ if (modeInput == null) {
+ return null
+ }
+ val sqlText = sqlInput.asInstanceOf[UTF8String].toString
+ val modeStr = modeInput.asInstanceOf[UTF8String].toString
+ val parsedMode = ParseSqlMode.parse(modeStr).getOrElse {
Review Comment:
**Nit (P3):** `mode` must be a constant and checkInputDataTypes already
validated it, yet eval decodes and parses it again for every row, so PARSE over
a column of statements now pays that per row. Caching the parsed mode, for
example in a `@transient lazy val` read after the NULL sqlStmt check, keeps the
results and the NULL handling.
**Verification:**
- **Behavior:** PARSE over a column of statements, NULL sqlStmt, NULL mode
and both modes return the same results as before.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -43,170 +85,251 @@ 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(segment: PositionedSqlStatement, 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)
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement, charStart: Int):
JObject = {
+ /** 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))
+ }
+ }
+
+ /** 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(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, charStart, length)
+ Parsed(segment, 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)
+ (
+ uniqueIdentities(
+ qualifyParseIdentities(
+ parseRefs.targetTables, analyzedTargets, qualifyTarget(_, plan,
s)),
+ s),
+ uniqueIdentities(
+ qualifyParseIdentities(
+ parseRefs.sourceTables, analyzedSources, 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) =>
+ (parseRefs.targetTables, parseRefs.sourceTables, parseRefs.functions)
+ }
+ if (targetTables.nonEmpty) {
fields += "target_table_references" ->
- JArray(refs.targetTables.map(partsToJArray).toList)
+ JArray(targetTables.map(partsToJArray).toList)
}
- if (refs.sourceTables.nonEmpty) {
+ if (sourceTables.nonEmpty) {
fields += "source_table_references" ->
- JArray(refs.sourceTables.map(partsToJArray).toList)
+ JArray(sourceTables.map(partsToJArray).toList)
}
- if (refs.functions.nonEmpty) {
- fields += "function_references" ->
JArray(refs.functions.map(partsToJArray).toList)
+ if (functions.nonEmpty) {
+ fields += "function_references" ->
JArray(functions.map(partsToJArray).toList)
}
- val selectList = collectSelectList(plan)
+ val selectList = analyzed.map(collectTypedSelectList(_, plan))
+ .getOrElse(collectSelectList(plan))
if (selectList.nonEmpty) {
fields += "select_list" -> JArray(selectList.toList)
}
- refs.parameterMarkers.foreach(markers => fields += "parameter_markers" ->
markers)
+ parseRefs.parameterMarkers.foreach(markers => fields +=
"parameter_markers" -> markers)
+ analyzeError.foreach { e =>
+ fields += "error" -> errorObject(e)
Review Comment:
**Nit (P3):** ANALYZE errors are rendered here, but no ANALYZE test asserts
their queryContext. The nested analysis runs under whatever CurrentOrigin the
optimizer set for the outer parse_sql expression, so an error built from that
origin instead of a parsed node's would carry the outer query's text and
offsets, and nothing would catch it. An ANALYZE case that asserts startIndex,
stopIndex and fragment for an unresolved column after a supplementary character
would pin the documented statement-relative code points.
**Verification:**
- **Regression:** ANALYZE of SELECT with a supplementary-character literal
followed by an unresolved column reports that column's statement-relative
code-point start and stop and its fragment.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +652,274 @@ 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 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]],
+ 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)) parts else fallback(parts)
+ assigned(i) = Some(ident)
+ }
+ }
+ assigned.map(_.get).toSeq
+ }
+
+ private def matchingIdentity(
+ parseParts: Seq[String],
+ analyzedIds: Seq[Seq[String]]): Option[Seq[String]] = {
+ val parsed = parseParts.map(_.toLowerCase(Locale.ROOT))
+ analyzedIds.find { analyzed =>
+ val ident = analyzed.map(_.toLowerCase(Locale.ROOT))
+ ident.endsWith(parsed) &&
+ (ident.length > parsed.length ||
+ ident.length == parsed.length && parsed.length >= 2)
+ }
+ }
+
+ private def isPathReference(parts: Seq[String]): Boolean = parts match {
+ case Seq(_, path) => path.contains("/") || path.contains(":")
+ case Seq(path) => path.contains("/") || path.startsWith("file:")
+ case _ => false
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = plan match {
+ case c: CreateViewCommand if c.viewType == LocalTempView =>
+ sessionIdent(parts.last)
+ case c: CreateViewCommand if c.viewType == GlobalTempView =>
+ globalTempIdent(parts.last)
+ case c: CreateTempViewUsing if c.global =>
+ globalTempIdent(parts.last)
+ case _: CreateTempViewUsing =>
+ sessionIdent(parts.last)
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect | _: CreateTable |
+ _: ReplaceTable | _: CreateView |
+ _: org.apache.spark.sql.execution.datasources.CreateTable =>
+ // Persistent DDL never names a temp view, even if one shares the name.
+ qualifyWithCurrentNamespace(parts, session)
+ case c: CreateViewCommand if c.viewType == PersistedView =>
+ qualifyWithCurrentNamespace(parts, session)
+ case _ =>
+ qualifySource(parts, session)
+ }
+
+ private def qualifySource(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ if (session.sessionState.catalog.isTempView(parts)) {
+ val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
+ val last = parts.last
+ if (parts.exists(_.equalsIgnoreCase(globalTemp))) {
+ globalTempIdent(last)
+ } else {
+ sessionIdent(last)
+ }
+ } else {
+ qualifyWithCurrentNamespace(parts, session)
+ }
+ }
+
+ 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 isSessionBuiltin(name: String): Boolean = {
+ val id = FunctionRegistry.builtinFunctionIdentifier(name)
+ !isTemp(name) && !isBuiltin(name) &&
+ (session.sessionState.functionRegistry.functionExists(id) ||
+ session.sessionState.tableFunctionRegistry.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) || isSessionBuiltin(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)
+ }
+ }
+
+ /**
+ * Qualify `parts` using the current catalog/namespace without loading a
+ * catalog plugin. `isCatalogRegistered` instantiates the plugin.
+ */
+ 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 isKnownCatalogName(other.head, session) => other
+ case other => catalogName +: other
+ }
+ }
+
+ private def isKnownCatalogName(
+ name: String,
+ session: ClassicSparkSession): Boolean = {
+ val mgr = session.sessionState.catalogManager
+ name.equalsIgnoreCase(CatalogManager.SESSION_CATALOG_NAME) ||
+ name.equalsIgnoreCase(mgr.currentCatalog.name()) ||
+ session.sessionState.conf.getAllConfs.exists { case (key, _) =>
+ key.equalsIgnoreCase("spark.sql.catalog." + name)
+ }
+ }
+
+ 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)
+ // Row-level rewrites add internal columns; leave them out of select_list.
+ case _: ReplaceData | _: WriteDelta => plan
Review Comment:
**Nit (P3):** This arm fixes the rewrite-column leak reported in an earlier
round, but no test reaches it, and none selects a CHAR or VARCHAR column for
the getRawType restore in collectTypedSelectList. Removing this arm would leak
columns such as `__row_operation` into select_list for row-level DELETE, UPDATE
and MERGE, and dropping getRawType would report a VARCHAR(10) column as string,
with every test still passing. CHAR/VARCHAR cases plus DELETE, UPDATE and MERGE
on a row-level InMemory table would cover both.
This extends existing discussions
[1](https://github.com/apache/spark/pull/59873#discussion_r4405430793),
[2](https://github.com/apache/spark/pull/59873#discussion_r4405430851),
[3](https://github.com/apache/spark/pull/59873#discussion_r4428489890).
**Verification:**
- **Regression:** A VARCHAR(10) and a CHAR(3) column report varchar and char
type objects with their lengths.
- **Regression:** ANALYZE of DELETE, UPDATE and MERGE on a row-level v2
table reports no select_list and no column such as __row_operation.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,603 @@
+/*
+ * 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") {
+ 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") {
Review Comment:
**Non-blocking (P2):** Every expected identity here, and in the rest of the
ANALYZE suite and its golden file, is exactly what the name-based fallback
produces on its own. The suite would still pass if matching against analyzed
identities returned nothing, and the MERGE misattribution in that code is not
caught. A case whose bound identity differs from the fallback's guess would
give it a failure signal, such as a v2 catalog table, a reference spelled in a
different case from the stored table, a row-level MERGE, or a SQL table
function.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59873#discussion_r4428489890).
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.optimizer
+
+import org.apache.spark.sql.catalyst.expressions.{Literal, ParseSql}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Evaluates constant `parse_sql` calls on the driver. ANALYZE needs the
+ * session catalog, so this must run even when `ConstantFolding` is excluded.
+ */
+object EvaluateParseSql extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan =
+ plan.transformAllExpressionsWithSubqueries {
+ case p: ParseSql if p.children.forall(_.foldable) =>
+ Literal.create(p.eval(), StringType)
Review Comment:
**Blocking (P1):** This evaluates every constant parse_sql call, PARSE
included, ahead of ConstantFolding and without its conditional-branch
tolerance. On the base branch, `SELECT CASE WHEN id > 0 THEN parse_sql(CAST(1/0
AS STRING)) END FROM range(1)` returns one NULL row: ConstantFolding catches
the DIVIDE_BY_ZERO in the untaken branch and leaves the call for runtime.
Reading this rule, the same query now fails during optimization, and so does
any constant argument that throws inside a CASE, IF, COALESCE or NVL input that
no row reaches. Only constant ANALYZE calls need driver evaluation, so folding
only those here and leaving PARSE to ConstantFolding restores the base-branch
behavior. How a failing ANALYZE call in an untaken branch should behave is up
to you.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59873#discussion_r4428489838).
**Verification:**
- **Regression:** SELECT CASE WHEN id > 0 THEN parse_sql(CAST(1/0 AS
STRING)) END FROM range(1) with ANSI on returns one NULL row.
- **Behavior:** Constant PARSE calls still fold to their JSON literal, and
ANALYZE still works with ConstantFolding excluded, at top level and in a scalar
subquery.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -43,170 +85,251 @@ 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(segment: PositionedSqlStatement, 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)
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement, charStart: Int):
JObject = {
+ /** 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))
+ }
+ }
+
+ /** 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(sql.codePointCount(0, sql.length) - 1),
Review Comment:
**Nit (P3):** statementSpans already stored this statement's code-point
count in `span.length`, so this scans the text a second time; the base branch
counted once. In this non-empty branch, `Some(span.length - 1)` gives the same
value.
**Verification:**
- **Compatibility:** Error queryContext indexes for statements with
supplementary characters, including the restored code-point test cases, are
unchanged.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +652,274 @@ 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 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]],
+ 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)) parts else fallback(parts)
+ assigned(i) = Some(ident)
+ }
+ }
+ assigned.map(_.get).toSeq
+ }
+
+ private def matchingIdentity(
+ parseParts: Seq[String],
+ analyzedIds: Seq[Seq[String]]): Option[Seq[String]] = {
+ val parsed = parseParts.map(_.toLowerCase(Locale.ROOT))
+ analyzedIds.find { analyzed =>
+ val ident = analyzed.map(_.toLowerCase(Locale.ROOT))
+ ident.endsWith(parsed) &&
+ (ident.length > parsed.length ||
+ ident.length == parsed.length && parsed.length >= 2)
+ }
+ }
+
+ private def isPathReference(parts: Seq[String]): Boolean = parts match {
+ case Seq(_, path) => path.contains("/") || path.contains(":")
Review Comment:
**Blocking (P1):** This decides whether a name is a path reference from '/'
or ':' in its text. With spark.sql.runSQLOnFiles on (the default),
ResolveSQLOnFile resolves any two-part name to a file relation without a
catalog table, so a relative path such as parquet.`data` analyzes successfully.
Because it contains neither character, it falls through to
qualifyWithCurrentNamespace and is reported as
["spark_catalog","parquet","data"], a catalog table that does not exist,
instead of ["parquet","data"]. Deciding this from the catalog-less file
relation in the analyzed plan would fix it. A missing catalog identity alone is
not enough: the query of a persistent CREATE VIEW is never walked, so its
two-part sources lack one too.
This extends existing discussions
[1](https://github.com/apache/spark/pull/59873#discussion_r4428489801),
[2](https://github.com/apache/spark/pull/59873#discussion_r4460514185).
**Verification:**
- **Regression:** ANALYZE of SELECT * FROM parquet.`data`, with data a
relative parquet directory, reports ["parquet","data"] and analyze_success true.
- **Compatibility:** Absolute paths still report [format, path], two-part
catalog tables still report their catalog identity, CREATE VIEW v AS SELECT a
FROM db.s still reports ["spark_catalog","db","s"] as its source, and failed
statements keep today's fallback.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala:
##########
@@ -125,6 +126,8 @@ class SparkOptimizer(
override def nonExcludableRules: Seq[String] = super.nonExcludableRules ++
Seq(
+ // ANALYZE needs the driver session; do not leave it to ConstantFolding.
+ EvaluateParseSql.ruleName,
Review Comment:
**Nit (P3):** Nothing tests this entry: the driver-evaluation test excludes
only ConstantFolding. If the entry were dropped, also excluding
EvaluateParseSql would send ANALYZE to a task, where it fails with
PARSE_SQL_ANALYZE_REQUIRES_SESSION. Adding
`org.apache.spark.sql.catalyst.optimizer.EvaluateParseSql` to that test's
excludedRules would pin it.
**Verification:**
- **Regression:** With both ConstantFolding and EvaluateParseSql in
spark.sql.optimizer.excludedRules, ANALYZE of a constant statement still
returns typed JSON instead of PARSE_SQL_ANALYZE_REQUIRES_SESSION.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +652,274 @@ 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 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]],
+ 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)) parts else fallback(parts)
+ assigned(i) = Some(ident)
+ }
+ }
+ assigned.map(_.get).toSeq
+ }
+
+ private def matchingIdentity(
+ parseParts: Seq[String],
+ analyzedIds: Seq[Seq[String]]): Option[Seq[String]] = {
+ val parsed = parseParts.map(_.toLowerCase(Locale.ROOT))
+ analyzedIds.find { analyzed =>
+ val ident = analyzed.map(_.toLowerCase(Locale.ROOT))
+ ident.endsWith(parsed) &&
+ (ident.length > parsed.length ||
+ ident.length == parsed.length && parsed.length >= 2)
+ }
+ }
+
+ private def isPathReference(parts: Seq[String]): Boolean = parts match {
+ case Seq(_, path) => path.contains("/") || path.contains(":")
+ case Seq(path) => path.contains("/") || path.startsWith("file:")
+ case _ => false
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = plan match {
+ case c: CreateViewCommand if c.viewType == LocalTempView =>
+ sessionIdent(parts.last)
+ case c: CreateViewCommand if c.viewType == GlobalTempView =>
+ globalTempIdent(parts.last)
+ case c: CreateTempViewUsing if c.global =>
+ globalTempIdent(parts.last)
+ case _: CreateTempViewUsing =>
+ sessionIdent(parts.last)
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect | _: CreateTable |
+ _: ReplaceTable | _: CreateView |
+ _: org.apache.spark.sql.execution.datasources.CreateTable =>
+ // Persistent DDL never names a temp view, even if one shares the name.
+ qualifyWithCurrentNamespace(parts, session)
+ case c: CreateViewCommand if c.viewType == PersistedView =>
+ qualifyWithCurrentNamespace(parts, session)
+ case _ =>
+ qualifySource(parts, session)
+ }
+
+ private def qualifySource(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ if (session.sessionState.catalog.isTempView(parts)) {
+ val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
+ val last = parts.last
+ if (parts.exists(_.equalsIgnoreCase(globalTemp))) {
+ globalTempIdent(last)
+ } else {
+ sessionIdent(last)
+ }
+ } else {
+ qualifyWithCurrentNamespace(parts, session)
+ }
+ }
+
+ 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 isSessionBuiltin(name: String): Boolean = {
+ val id = FunctionRegistry.builtinFunctionIdentifier(name)
+ !isTemp(name) && !isBuiltin(name) &&
+ (session.sessionState.functionRegistry.functionExists(id) ||
+ session.sessionState.tableFunctionRegistry.functionExists(id))
+ }
+ def builtinIdent(name: String): Seq[String] =
+ FunctionRegistry.builtinFunctionIdentifier(name).nameParts
+ parts match {
+ case Seq(name) if tempFirst && isTemp(name) =>
Review Comment:
**Nit (P3):** ANALYZE tests assert function_references only for builtins
(count, range). This temp-first branch, and the extension, temp, explicit
system.builtin/system.session and persistent branches below it, have no failure
signal, so reporting a temp function as a builtin, or an explicit
system.session.f as spark_catalog.system.session.f, would pass. The
isSessionBuiltin INTERNAL_ERROR fixed earlier in this PR came from one of these
branches.
This extends existing discussions
[1](https://github.com/apache/spark/pull/59873#discussion_r4405430851),
[2](https://github.com/apache/spark/pull/59873#discussion_r4428489890).
**Recommended change:** Report each unqualified routine's owner from the
analyzer's own resolution order: add a small helper next to FunctionResolution
that walks the same path candidates (sqlResolutionPathEntriesForAnalysis,
including builtin, table-function, extension and temp entries) and returns the
first candidate that exists. Use it in qualifyRoutine for successfully analyzed
statements, and keep today's fallback only when no candidate can be verified.
Add function-reference tests for each resolution route.
**Why this works:** The reported owner comes from the same ordered candidate
list that bound the call, so every sessionOrder value and SQL path order
selects the same routine in analysis and in the JSON. The same walk also
reports persistent functions found through a SET PATH or spark.sql.defaultPath
schema, which the author deferred to a follow-up.
**Scope:** Replace qualifyRoutine's fixed order with a
FunctionResolution-backed lookup of the bound owner, and cover each routine
route in ANALYZE tests.
**Compatibility:** Builtins and builtin table functions keep system.builtin
identities, de-duplication stays case-insensitive in first-seen order, and
PARSE output is unchanged.
**Risks:** A path walk probes catalog function lookups once per routine per
statement on the driver; it stops at the first match, so its cost is bounded by
what analysis already did. function_references change for sessions with
sessionOrder=last or a custom SQL path; they change from wrong owners to the
bound ones.
**Constraints:** Qualification must not instantiate a catalog plugin that
analysis did not load. PARSE output and failed-statement best-effort
qualification stay as they are.
**Exclusions:**
- Statements whose analysis failed keep today's best-effort routine
qualification.
- Persistent functions found only through a SET PATH or
spark.sql.defaultPath schema other than the current one may keep the
current-namespace fallback the author deferred to a follow-up; the candidate
walk reports them correctly if that follow-up is folded in.
**Success:** For a successfully analyzed statement, no unqualified routine
in function_references names a temporary or builtin function that analysis did
not bind, under any sessionOrder value or SQL path order. Temp functions,
builtins, builtin table functions and current-schema persistent functions
report the owner analysis bound.
**Verification:**
- **Regression:** With spark.sql.functionResolution.sessionOrder=last and
both a temp and a current-schema persistent function of the same name, ANALYZE
reports the persistent function as catalog.schema.name.
- **Regression:** With spark.sql.path.enabled true, a path that lists
system.session before system.builtin, and a temp function that shadows a
builtin name, ANALYZE reports system.session.<name>.
- **Behavior:** Temp functions report system.session, builtins and builtin
table functions such as range report system.builtin, injected extension
functions report system.builtin, explicit system.builtin or system.session
names keep that owner, and a persistent function in the current schema reports
catalog.schema.name.
- **Compatibility:** PARSE function_references and the misconfigured-catalog
batch behavior are unchanged.
--
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]
