dtenedor commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4486023409
##########
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:
**Medium: References written with a non-literal `IDENTIFIER(...)` are
dropped, and their identity can be credited to another name.**
`withIdentClause` folds only a bare string literal at parse time, so
`IDENTIFIER(src)` (a session variable), `IDENTIFIER('db2' || '.t')`, and every
`CREATE TEMP VIEW IDENTIFIER(...)` (that overload never folds) stay a
`PlanWithUnresolvedIdentifier` with no parse name. Since parse names now decide
which references exist, the resolved relation is not reported at all. Its
analyzed identity also stays in the pool, where this first-suffix match hands
it to any one-part name with the same last part: under `USE db1` with `db1.t`,
`db2.t` and `DECLARE VARIABLE src STRING DEFAULT 'db2.t'`, `parse_sql('SELECT *
FROM t WHERE a IN (SELECT a FROM IDENTIFIER(src))', 'ANALYZE')` returns
`analyze_success: true` with only `[["spark_catalog","db2","t"]]` (subqueries
are walked before children), where round 2 reported both tables. Evaluating a
foldable `identifierExpr`
in the parse walk and visiting the plan it builds, appending analyzed
identities that no parse name claimed when the statement has a dynamic name,
and preferring, among suffix matches, the one equal to the name-based
qualification would fix this; a test with `IDENTIFIER(...)` next to a
same-named table would pin it.
##########
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:
**Medium: Views from a v2 catalog with nested namespaces get a truncated
identity.** `viewParts` reads `v.desc.identifier`, which `RelationResolution`
builds with `asLegacyTableIdentifier` and which keeps only the last namespace
part. With `InMemoryRelationCatalog` as `relcat`, `CREATE VIEW relcat.a.b.v AS
SELECT 1 AS x` and `USE relcat.a.b`, `SELECT * FROM v` reports
`[["relcat","b","v"]]`, and the one-part name `v` claims that identity instead
of falling back to the correct `relcat.a.b.v`. The `CatalogTable` carries the
full name (`fullIdent`); one helper used by `viewParts`, the
`HiveTableRelation` arm and `logicalRelationParts` that takes the full name and
prefixes `spark_catalog` when the catalog is missing would fix it, and would
also cover `spark.sql.legacy.v1IdentifierNoCatalog=true`, where only
`logicalRelationParts` adds the catalog today and views come out as two parts.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -452,11 +616,41 @@ object ParseSqlResult {
case u: UnresolvedView => add(u.multipartIdentifier)
case u: UnresolvedTableOrView => add(u.multipartIdentifier)
case u: UnresolvedIdentifier if role == TableRefRole.Target =>
add(u.nameParts)
- case c: CreateViewCommand => addTarget(tableIdentifierParts(c.name))
- case c: CreateTempViewUsing =>
addTarget(tableIdentifierParts(c.tableIdent))
+ case c: CreateViewCommand => addTarget(Seq(c.name.table))
+ case c: CreateTempViewUsing => addTarget(Seq(c.tableIdent.table))
+ case c: CreateDataSourceTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateDataSourceTableAsSelectCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: DropTableCommand =>
+ addTarget(tableIdentifierParts(c.tableName))
+ case c: DropTempViewCommand =>
+ addTarget(tempViewParts(c.ident))
case c: CacheTable if c.multipartIdentifier.nonEmpty =>
Review Comment:
**Medium: `CACHE TABLE db.t` reports `["db","t"]`.** Analysis keeps
`CacheTable.multipartIdentifier` as written, this arm puts that name into the
analyzed pool too, and the two-part parse name matches it at equal length, so
the target is never qualified: `CACHE TABLE db.t` gives `[["db","t"]]`, `CACHE
TABLE global_temp.gv` gives `[["global_temp","gv"]]`, and with a v2 current
catalog `CACHE TABLE ns.t` gives `[["ns","t"]]`, while `CACHE TABLE t`,
`UNCACHE TABLE db.t` and a failed `CACHE TABLE db.missing` all come out fully
qualified. Collecting the written name only for an unanalyzed `CacheTable`
(`!c.isAnalyzed`; the arm runs before the slots are walked, so the guard is
needed) and otherwise walking the resolved `c.table` with the target role would
fix it. There is no ANALYZE test for CACHE or UNCACHE yet.
##########
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(
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ 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
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ 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 =>
Review Comment:
**Medium: These target arms alone decide the ANALYZE output for global and
USING temp views, and nothing tests them.** For `CREATE [GLOBAL] TEMP VIEW` (AS
and USING), both the parse target and the analyzed target are the one-part
name, which never matches, so deleting L857-862 would report
`spark_catalog.default.v` or `system.session.v` with every test passing; the
DROP VIEW tests mentioned in the earlier thread never build a global
`CreateViewCommand` or a `CreateTempViewUsing`. ANALYZE tests for `CREATE
GLOBAL TEMP VIEW ... AS`, `CREATE TEMPORARY VIEW ... USING parquet OPTIONS
(path ...)` and `CREATE GLOBAL TEMPORARY VIEW ... USING ...` would pin them.
Relatedly, `qualifyTarget` only looks at the plan root, so `EXPLAIN CREATE
TEMPORARY VIEW v AS SELECT 1` reports `spark_catalog.default.v`; a first arm
`case ExplainCommand(inner, _) => qualifyTarget(parts, inner, session)` would
fix that.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala:
##########
@@ -64,6 +64,7 @@ class SparkOptimizer(
override def defaultBatches: Seq[Batch] = flattenBatches(Seq(
preOptimizationBatches,
+ Batch("Evaluate parse_sql", Once, EvaluateParseSql),
Review Comment:
**Low: This batch runs before Finish Analysis, so the SQL text can be built
before `SpecialDatetimeValues` and `ComputeCurrentTime` have run.** With ANSI
off, `parse_sql('SELECT DATE''' || CAST(CAST('epoch' AS DATE) AS STRING) || '''
AS d', 'ANALYZE')` is NULL, because the cast is NULL until
`SpecialDatetimeValues` rewrites it, while the same call with `'PARSE'` (folded
later) returns JSON. `try_cast('today' AS DATE)` does the same with ANSI on,
and with ConstantFolding excluded the ANSI error is swallowed and the call ends
up failing in a task. `current_date()` and `current_timestamp()` in the text
are also read before the query's own snapshot. Running the batch right after
Finish Analysis (still before Inline CTE) would avoid this.
##########
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:
**Medium: Cached and streamed plans run ANALYZE in a cloned session that has
lost the current v2 catalog and namespace.** ANALYZE uses whichever session is
active while the optimizer runs. `CacheManager` optimizes the plan it caches
inside `getOrCloneSessionWithConfigsOff`, which clones under default settings,
and streaming micro-batches are optimized in `sparkSessionForStream`; a clone
copies the v1 current database, temp views and the SQL path, but not the v2
current catalog or namespace. So after `USE testcat.ns`, `SELECT
parse_sql('SELECT a FROM t', 'ANALYZE')` reports `testcat.ns.t`, but `CACHE
TABLE c AS SELECT parse_sql('SELECT a FROM t', 'ANALYZE') AS j` caches
`TABLE_OR_VIEW_NOT_FOUND` (or a different `spark_catalog.default.t`), and later
identical calls are served that cached JSON. This is not new in these commits
and comes from an existing session-cloning gap (`current_catalog()` in a cached
plan behaves the same way), but ANALYZE now depends on it for every unqualified
name. Documenting it next to "inherits the invoker's current catalog, schema,
and SQL path" with a test, or capturing the invoker's catalog and namespace
when the call is analyzed, would address it.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -452,11 +616,41 @@ object ParseSqlResult {
case u: UnresolvedView => add(u.multipartIdentifier)
case u: UnresolvedTableOrView => add(u.multipartIdentifier)
case u: UnresolvedIdentifier if role == TableRefRole.Target =>
add(u.nameParts)
- case c: CreateViewCommand => addTarget(tableIdentifierParts(c.name))
- case c: CreateTempViewUsing =>
addTarget(tableIdentifierParts(c.tableIdent))
+ case c: CreateViewCommand => addTarget(Seq(c.name.table))
+ case c: CreateTempViewUsing => addTarget(Seq(c.tableIdent.table))
+ case c: CreateDataSourceTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateDataSourceTableAsSelectCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: DropTableCommand =>
+ addTarget(tableIdentifierParts(c.tableName))
+ case c: DropTempViewCommand =>
+ addTarget(tempViewParts(c.ident))
case c: CacheTable if c.multipartIdentifier.nonEmpty =>
addTarget(c.multipartIdentifier)
case u: UnresolvedTableValuedFunction => addFunction(u.name)
+ case w: ReplaceData => rewriteTargets ++= rewriteTargetParts(w.table)
Review Comment:
**Low: `rewriteTargets` misses insert-only MERGE and matches by name.** An
insert-only MERGE becomes `InsertOnlyMerge`, which re-reads the target in a
LeftAnti join and isn't handled here or in `primaryQueryPlan`, so `MERGE ...
WHEN NOT MATCHED THEN INSERT *` gets a typed `select_list` of the target
columns (references are fine today only because the source is walked first).
Filtering by name also hides a source the statement really names when it is the
same table, as in `DELETE FROM t WHERE id IN (SELECT id FROM t)`, which then
falls back to name-based qualification. Recognizing the rewrite's reads
structurally (a `DataSourceV2Relation` over a `RowLevelOperationTable`, plus
the `InsertOnlyMerge` re-read) would avoid both.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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 scala.util.control.NonFatal
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal,
ParseSql}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.PARSE_SQL
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Evaluates constant ANALYZE `parse_sql` calls on the driver, where the
session catalog is
+ * available. ANALYZE must run here even when `ConstantFolding` is excluded.
PARSE calls are
+ * left to `ConstantFolding`, which keeps its tolerance for failing constants
in untaken
+ * branches.
+ *
+ * A constant ANALYZE call whose SQL argument fails to evaluate is left in
place. The error
+ * then surfaces when the expression runs, so a failure in an untaken branch
stays unobserved.
+ */
+object EvaluateParseSql extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan =
+ plan.transformDownWithSubqueriesAndPruning(_.containsPattern(PARSE_SQL)) {
+ case subQuery =>
+
subQuery.transformAllExpressionsWithPruning(_.containsPattern(PARSE_SQL)) {
+ case p: ParseSql if p.isConstantAnalyze => evaluate(p)
+ }
+ }
+
+ private def evaluate(p: ParseSql): Expression =
Review Comment:
**Medium: This catch also swallows failures of ANALYZE itself, not just a
failing SQL argument as the scaladoc says.** `p.eval()` evaluates `sqlStmt` and
then runs `ParseSqlResult.fromSql`, which deliberately lets internal errors and
non-Spark exceptions escape; those are swallowed here too, and the call is left
in place. At top level ConstantFolding then re-runs the whole ANALYZE before
the error surfaces, but in a taken CASE / IF branch, or with ConstantFolding
excluded, the call reaches a task and fails with
`PARSE_SQL_ANALYZE_REQUIRES_SESSION` instead of the real error. For example,
`SELECT CASE WHEN id = 0 THEN parse_sql('SELECT 1 LIMIT
length(format_string(''%d'', ''a''))', 'ANALYZE') END FROM range(1)` loses the
`IllegalFormatConversionException` that the same call raises at top level. A
call left in place is also re-evaluated at every enclosing operator, because
`transformAllExpressionsWithPruning` here rewrites each node's whole subtree.
Evaluating only `p.sqlStmt` inside
the try and running ANALYZE outside it, and transforming only each node's own
expressions (`transformExpressionsWithPruning`), would fix both.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -64,26 +94,66 @@ import org.apache.spark.unsafe.types.UTF8String
12
> SELECT get_json_object(_FUNC_('SELEC'), '$[0].error.errorClass');
PARSE_SYNTAX_ERROR
+ > SELECT get_json_object(_FUNC_('SELECT 1 AS a', 'ANALYZE'),
'$[0].select_list[0].type.name');
+ int
+ > SELECT get_json_object(_FUNC_('SELECT a FROM missing_t', 'ANALYZE'),
'$[0].analyze_success');
+ false
+ """,
+ note = """
+ Requires spark.sql.function.parseSql.enabled=true. The result type is
STRING JSON.
+ ANALYZE needs an active session, and `sqlStmt` and `mode` must be constant
for the
+ query (DATATYPE_MISMATCH.NON_FOLDABLE_INPUT otherwise); a DataFrame column
of
+ statements can use PARSE only. Each JSON object has `start` (1-based
character
+ offset in the batch), `length`, and `parse_success`. Parse success also has
+ `statement_identifier` / `statement_code` (ISO/IEC 9075-2:2023 Table 39).
ANALYZE
+ adds `analyze_success`. Optional fields are omitted when empty:
+ `target_table_references`, `source_table_references`,
`function_references` (arrays
+ of name parts; ANALYZE qualifies persistent names to
catalog.schema.object, local
+ temp views and temp functions as `system.session`, builtins as
`system.builtin`;
+ CTE aliases in scope are not listed as tables), `select_list` (`{name}`
parts;
+ ANALYZE lists the output columns, with `*` expanded, and adds `type` as a
+ DESCRIBE TABLE EXTENDED AS JSON type object), `parameter_markers` (`named`
and/or
+ `unnamed_count`), and `error` (STANDARD format). Nested error locations are
+ statement-relative Unicode code points. Later statements are still
processed
+ after a per-statement parse or analysis failure. ANALYZE of a SQL script
reports
+ `analyze_success` false without executing the script. Path-based relations
keep
+ `[format, path]` as written. Analysis may still run time-travel subqueries,
+ infer file schemas, and call Python UDTF `analyze`.
""",
group = "misc_funcs",
since = "4.4.0")
// scalastyle:on nonascii
// scalastyle:on line.size.limit
-case class ParseSql(child: Expression)
- extends UnaryExpression
+case class ParseSql(
+ sqlStmt: Expression,
+ mode: Expression)
+ extends Expression
with ImplicitCastInputTypes
- with CodegenFallback {
+ with CodegenFallback
+ with QueryErrorsBase {
+
+ def this(sqlStmt: Expression) = this(sqlStmt, Literal("PARSE"))
override def prettyName: String = "parse_sql"
- override def nullable: Boolean = true
+ override def children: Seq[Expression] = Seq(sqlStmt, mode)
+
+ override def nullable: Boolean = sqlStmt.nullable || mode.nullable
+
+ // Foldable so constant SQL is evaluated on the driver, where ANALYZE can
see the session.
Review Comment:
**Low: Some doc text no longer matches the code.** This comment predates
`EvaluateParseSql`: constant ANALYZE now reaches the driver through that rule,
and `foldable` mainly lets ConstantFolding fold constant PARSE calls
(`isConstantAnalyze` could also reuse `foldable`). In the usage and note,
"Nested error locations are statement-relative" sits next to "view and EXECUTE
IMMEDIATE bodies keep their own text"; "fully qualifies resolved relations and
routines" holds while path-based relations stay as written and routine names
don't follow `SET PATH` yet; ANALYZE's `select_list` for INSERT is the
target-aligned projection (`INSERT INTO t SELECT 1 AS x` reports `t`'s column),
which isn't mentioned; and the "Analysis may still run time-travel subqueries"
sentence appears in both the usage and the note.
##########
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(
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ 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
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ 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 =>
Review Comment:
**Low: A few new arms and imports are unreachable.** `qualifyTarget` only
receives the parsed plan, which never contains the v1
`execution.datasources.CreateTable` (written here as an inline fully qualified
name) or a persisted-view `CreateViewCommand` (L868), because the parser builds
`CreateView` for those. The `SQLFunctionNode` import at L31 duplicates the
`analysis._` wildcard, and `jsonBoolean` in the suite is unused.
##########
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(
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ 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
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ 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
+ // Same order as FunctionResolution: `first` is session, builtin,
persistent;
+ // `second` (default) is builtin, session, persistent; `last` is builtin,
persistent, session.
+ val sessionOrder =
+
session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER)
+ def isTemp(name: String): Boolean =
+ catalog.isTemporaryFunction(FunctionIdentifier(name))
+ def isPersistent(name: String): Boolean =
+ catalog.isPersistentFunction(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 sessionOrder == "first" && isTemp(name) =>
+ sessionIdent(name)
+ case Seq(name) if isBuiltin(name) || isSessionBuiltin(name) =>
+ builtinIdent(name)
+ case Seq(name) if sessionOrder == "last" && isPersistent(name) =>
+ qualifyWithCurrentNamespace(parts, session)
+ 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(
Review Comment:
**Low: `isKnownCatalogName` matches more loosely than catalog lookup.**
`Catalogs.load` reads `spark.sql.catalog.<name>` with the exact name, so with
only `spark.sql.catalog.prod` set the analyzer treats `Prod.ns.t` as namespace
`Prod` in the current catalog, while this reports `Prod` as a catalog. It also
copies every setting through `getAllConfs` for each name;
`session.sessionState.conf.contains(s"spark.sql.catalog.$name")` is exact and
cheap. On the same fallback path, a missing `system.session.x` becomes the
four-part `spark_catalog.system.session.x`, because only existing temp views
are mapped to `system.session`; mapping session-qualified names first would fix
that.
##########
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:
**Low: Several negative checks can't fail for the wrong error.**
`TABLE_OR_VIEW_NOT_FOUND` and `UNRESOLVED_ROUTINE` have no subclasses, so the
`== X || contains(X)` checks here and further down reduce to `contains`. The
`startsWith` arm in the script-locals check makes its exact alternatives dead,
the `NON_FOLDABLE_INPUT` checks don't assert `inputName` (so they can't tell
whether `sqlStmt` or `mode` was rejected), and the new `functionName` parameter
of `PARSE_SQL_ANALYZE_REQUIRES_SESSION` and
`PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT` is never asserted. `checkError` with the
condition and parameters, or exact comparisons of the error JSON, would make
these meaningful.
##########
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(
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ 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
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ 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
+ // Same order as FunctionResolution: `first` is session, builtin,
persistent;
+ // `second` (default) is builtin, session, persistent; `last` is builtin,
persistent, session.
+ val sessionOrder =
+
session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER)
+ def isTemp(name: String): Boolean =
+ catalog.isTemporaryFunction(FunctionIdentifier(name))
+ def isPersistent(name: String): Boolean =
+ catalog.isPersistentFunction(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 sessionOrder == "first" && isTemp(name) =>
+ sessionIdent(name)
+ case Seq(name) if isBuiltin(name) || isSessionBuiltin(name) =>
+ builtinIdent(name)
+ case Seq(name) if sessionOrder == "last" && isPersistent(name) =>
Review Comment:
**Low: With `sessionOrder=last`, this checks for a persistent function in
the session catalog, not in the catalog it reports.** `isPersistentFunction`
looks in the v1 session catalog's current database, which is `default` while a
v2 catalog is current, but `FunctionResolution` and
`qualifyWithCurrentNamespace` use the current catalog and namespace. With
`default.f` and a temp `f`, after `USE testcat.ns`, `SELECT f()` binds
`system.session.f` but is reported as `testcat.ns.f` (round 2 got this case
right); with `testcat.ns.f` plus a temp `f` it reports `system.session.f`.
Probing the current catalog's `FunctionCatalog`
(`functionExists(Identifier.of(currentNamespace, name))`, which is the same
check as today for `spark_catalog`) would fix it, and the comment above could
say that the SQL path isn't followed yet.
##########
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(
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ 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
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ 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(
Review Comment:
**Medium: For several common commands the analyzed walk sees nothing, so
this guess decides the reference, and it can disagree with the analyzer.**
After analysis, v1 CTAS keeps its query in `innerChildren`, `CreateViewCommand`
and `AlterViewAsCommand` have no children, and
`InsertIntoHadoopFsRelationCommand` / `InsertIntoDataSourceCommand` keep the
target in a field. So CTAS and CREATE VIEW sources and v1 INSERT targets always
fall back to "temp view if one exists, else the current namespace", while
relation resolution follows the SQL path. With `spark.sql.path.enabled` and
`SET PATH = spark_catalog.db2, system.builtin`, `INSERT INTO s SELECT 1` or
`CREATE TABLE c USING parquet AS SELECT a FROM s` (with `s` only in `db2`)
reports `spark_catalog.default.s`. With
`spark.sql.functionResolution.sessionOrder=last` a persistent `t` wins over a
temp view `t`, but this reports `system.session.t`. The written case is also
kept: `CREATE VIEW v AS SELECT * FROM PARSE_SQL_CASE_T` reports tha
t spelling, while SELECT reports the stored name. Separately, `ALTER VIEW v AS
SELECT * FROM t` still reports `v` as a source and has no target, because
`AlterViewAs` is not in `isTargetTableChild` and has no arm like `CreateView`.
Walking those analyzed slots (`CreateDataSourceTableAsSelectCommand.query`,
`CreateViewCommand.plan`, `AlterViewAsCommand.query` and
`InsertIntoDataSourceCommand.query` as sources;
`InsertIntoHadoopFsRelationCommand.catalogTable` as the target), adding an
`AlterViewAs` arm, and having the fallback follow
`resolutionPathEntriesForAnalysis` would fix this.
##########
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:
**Medium: Three of the new guards have no test that fails without them.**
With suffix matching, any relation collected from a body the statement doesn't
name can claim a one-part name. (a) No test has a scalar SQL function whose
body reads a same-named table: with `CREATE FUNCTION f() RETURNS INT RETURN
(SELECT max(x) FROM db2.t)`, `SELECT f() FROM t` should report only
`spark_catalog.default.t`, but without the `SQLScalarFunction` skip at L503 the
body is walked first and `t` claims `db2.t`. (b) Every view in the suite has a
literal body, so this `View` / `ResolvedMetricView` stop is untested (for
example a view over `db2.t` in `SELECT * FROM v JOIN t`). Metric views have no
`View` wrapper, so this stop is their only guard, and since no arm records
`ResolvedMetricView`'s identifier, a metric view's name always comes from the
fallback. (c) The row-level tests use group-based tables only, so dropping
`WriteDelta` from L635 or L975 keeps the suite green, while on a table with
`'supp
orts-deltas' = 'true'` MERGE would then report its target as a source and
`select_list` would include `__row_operation` and the row id.
##########
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:
**Low: This catch reverts all three lists to raw names when any single
lookup fails, and no test reaches it.** One failing probe (for example a
metastore error from `databaseExists` or `isPersistentFunction`) drops
identities that were already matched from the analyzed plan, while
`analyze_success` stays true. With `isKnownCatalogName` in place I couldn't
find an input that reaches it, and the misconfigured-catalog test passes with
either mechanism alone and asserts only `analyze_success`. Catching per name
(keeping the parse name for that name only) or removing the catch, and
asserting the references and error class in that test, would make this
observable.
##########
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(
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ 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
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
Review Comment:
**Low: `isFileReference` can report a v2-catalog table as a file path, and
doesn't look at the analyzed plan as its doc says.** It only checks that the
first part names a data source and is not a session-catalog database. So with a
v2 current catalog `testcat` that has a namespace `json`, `CREATE TEMPORARY
VIEW tv AS SELECT a FROM json.s` (whose query is not walked after analysis)
reports `[["json","s"]]` instead of `[["testcat","json","s"]]`. Requiring the
session catalog to be current (or checking the current catalog's namespaces),
catching only the class-not-found case as `ResolveSQLOnFile` does instead of a
broad `Try`, and rewording the doc would tighten it. The one-part arm of
`isPathReference` above now only sees parse names, so it can only misfire on a
quoted table name that contains `/`.
##########
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:
**Low: A few behaviors still have no failure signal.** No statement after
`USE parse_sql_iso_db` depends on the current schema, so a USE that leaked into
later statements would still pass; a later statement that reads a table
existing only in `default`, with its qualified source asserted, would catch it.
NULL `mode` is untested since the switch to `constantMode`
(`checkEvaluation(ParseSql(Literal("SELECT 1"), Literal.create(null,
StringType)), null)` would cover it, including serialization). The routine
cases other than `second` and `last` (temp-first `first`, explicit
`system.session.f` and `system.builtin.f`, a persistent function under the
default order, an injected function) are still untested too.
--
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]
