dtenedor commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4499313072
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +711,328 @@ 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) and the session
catalog current, a
+ * two-part name whose first part is a data source and not a session
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 = {
+ val mgr = session.sessionState.catalogManager
+ parts match {
+ case Seq(format, _)
+ if
mgr.currentCatalog.name().equalsIgnoreCase(CatalogManager.SESSION_CATALOG_NAME)
=>
+ val isDataSource =
Review Comment:
**Medium: `isFileReference` now lets data source lookup errors escape, so
every reference in the statement reverts to raw names.** My round-3 suggestion
to catch only `ClassNotFoundException` here was wrong. `lookupDataSource`
throws `AnalysisException` for `kafka` and `avro` when their connectors aren't
loaded (`KAFKA_DATA_SOURCE_NOT_ENABLED` / `AVRO_DATA_SOURCE_NOT_ENABLED`), and
it runs before `databaseExists`, so the exception reaches the catch in
`successJson`, which reverts targets, sources and functions to raw parse names
while `analyze_success` stays true. With `CREATE DATABASE kafka` and a table
`kafka.t`, `parse_sql('CREATE VIEW kafka.v AS SELECT upper(s) AS u FROM
kafka.t', 'ANALYZE')` reports `[["kafka","v"]]`, `[["kafka","t"]]` and
`[["upper"]]`, where the previous revision reported `spark_catalog.kafka.v`,
`spark_catalog.kafka.t` and `system.builtin.upper`; `TRUNCATE TABLE kafka.t`
and `UNCACHE TABLE kafka.t` take the same path. Unlike in `ResolveSQLOnFile`,
analysis
has already succeeded here, so any non-internal lookup failure just means
"not a file": checking `databaseExists` first and treating every such failure
as false would fix it. The session-catalog requirement I suggested also loses a
real case: `ResolveSQLOnFile` reads `` parquet.`rel_dir` `` as a file whichever
catalog is current, so under `USE testcat.ns` that reference now comes out as
`["testcat","parquet","rel_dir"]` instead of `["parquet","rel_dir"]`. Checking
the current catalog's namespaces, instead of requiring the session catalog,
keeps both cases right.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.{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.
+ *
+ * Only the SQL argument is evaluated under the guard. If it fails, the call
is left in place
+ * and the error surfaces when the expression runs. Failures of ANALYZE itself
are not guarded,
+ * so they are not hidden from the query.
+ */
+object EvaluateParseSql extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan =
+ plan.transformDownWithSubqueriesAndPruning(_.containsPattern(PARSE_SQL)) {
+ case subQuery =>
+ // Per node: only this node's own expressions. A subtree rewrite would
re-run
+ // the call once per enclosing operator.
+ subQuery.transformExpressionsWithPruning(_.containsPattern(PARSE_SQL))
{
+ case p: ParseSql if p.isConstantAnalyze && sqlArgumentEvaluates(p) =>
Review Comment:
**Medium: Nothing tests the narrowed guard, so the round-3 behavior could
come back unnoticed.** Every existing test passes with the old catch around all
of `p.eval()`: the untaken-branch tests fail in the SQL argument, which both
versions leave in place, and no test makes ANALYZE itself throw. With `call =
"parse_sql('SELECT 1 LIMIT length(format_string(''%d'', ''a''))', 'ANALYZE')"`,
running `SELECT CASE WHEN id = 0 THEN $call END FROM range(1)` and, with
ConstantFolding excluded, `SELECT $call`, and intercepting
`java.util.IllegalFormatConversionException` in both, would pin it:
CheckAnalysis evaluates the LIMIT, the exception escapes `fromSql`, and with
the old guard both queries instead fail in a task with
`PARSE_SQL_ANALYZE_REQUIRES_SESSION`.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -375,8 +538,19 @@ object ParseSqlResult {
// An INSERT target is never resolved against a CTE, so no alias shadows
it.
case i: InsertIntoStatement =>
visitPlan(i.table, Set.empty, TableRefRole.Target)(f)
- case c: CacheTable if c.multipartIdentifier.isEmpty =>
+ case c: CacheTable if c.multipartIdentifier.isEmpty || c.isAnalyzed =>
visitPlan(c.table, scope, TableRefRole.Target)(f)
+ // The query of a v1 CTAS, CREATE VIEW, ALTER VIEW or INSERT is not a
child after analysis.
Review Comment:
**Low: This comment isn't true for every command listed, and the per-command
list still misses some.** `InsertIntoHadoopFsRelationCommand` is a
`DataWritingCommand` whose child is `query`, so that arm walks the query a
second time, and a parsed (unanalyzed) temp-view `CreateViewCommand` still has
`plan` as a child, so it is walked twice as well; that is harmless today only
because results are de-duplicated and the parser rejects parameter markers in
CREATE VIEW. Meanwhile `UncacheTable` has the same shape as `CacheTable` but
isn't walked, so `UNCACHE TABLE PARSE_SQL_CACHE_T` reports the written case
while CACHE reports the stored name, and `InsertIntoDataSourceDirCommand` and
`CreateHiveTableAsSelectCommand` keep their query only in `innerChildren`.
Walking `AnalysisOnlyCommand.childrenToAnalyze` and
`LeafRunnableCommand.innerChildren` generically, with `CacheTable` added to
`isTargetTableChild`, would replace these arms and cover the rest.
##########
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:
On the design question, a version that doesn't need session variables in the
parse walk: (1) when `identifierExpr` is foldable (a literal, or `'db2' ||
'.t'`), evaluate it in the parse walk and visit the plan it builds; (2) in
ANALYZE, when the parsed plan still has a `PlanWithUnresolvedIdentifier`,
append the analyzed identities that no parse name claimed, in their analyzed
role; (3) when several analyzed identities match a parse name, prefer the one
equal to the name-based qualification. Step (3) alone stops `t` from being
credited with `db2.t`, so if you'd rather defer the reporting part, I'd still
do (3) in this PR and list IDENTIFIER references as a known limitation. Since
the CTAS, CREATE VIEW, ALTER VIEW and INSERT queries now feed the analyzed pool
too, the misattribution now also reaches statements like `CREATE TABLE c USING
parquet AS SELECT * FROM IDENTIFIER(src) JOIN t`.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,946 @@
+/*
+ * 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 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")
+ }
+ }
+
+ 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"))
Review Comment:
**Low: Three of the earlier assertion items are still open.** This
`startsWith("UNRESOLVED_COLUMN")` arm still makes the two exact alternatives
dead; the `NON_FOLDABLE_INPUT` checks below don't assert `inputName`, so they
can't tell whether `sqlStmt` or `mode` was rejected; and the `functionName`
parameter of `PARSE_SQL_ANALYZE_REQUIRES_SESSION` and
`PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT` is never asserted. `checkError` with the
condition and parameters would cover all three.
##########
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:
Deferring this is fine with me. Could you file a JIRA for it and list it
with the known limitations in the PR description?
##########
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:
Thanks. To be specific about what's left: (a) and the view half of (b) are
covered now. Still missing are a metric-view case for the `ResolvedMetricView`
stop (no arm records the metric view's own identifier yet, so its name always
comes from the fallback), and (c) a table with `'supports-deltas' = 'true'`,
without which removing `WriteDelta` from the role match, `rewriteTargets` or
`primaryQueryPlan` keeps the suite green.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,946 @@
+/*
+ * 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 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")
+ }
+ }
+
+ 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")
+ 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")
+ }
+ }
+
+ 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;
+ |CREATE TEMP VIEW parse_sql_iso_v AS SELECT 1 AS a;
+ |SELECT * FROM parse_sql_iso_v;
+ |SELECT 1 AS a FROM missing_parse_sql_iso;
+ |SELECT 1 AS ok""".stripMargin)
+ assert(results.map(_ \ "analyze_success") ===
+ List(JBool(true), JBool(true), JBool(false), JBool(false),
JBool(true)))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
schema)
+ assert(!spark.catalog.tableExists("parse_sql_iso_v"))
+ } finally {
+ sql("DROP DATABASE IF EXISTS parse_sql_iso_db")
+ }
+ }
+ }
+
+ test("ANALYZE does not run EXECUTE IMMEDIATE DML") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_exec_t") {
+ sql("CREATE TABLE parse_sql_exec_t (a INT) USING parquet")
+ val j = evalStatement("EXECUTE IMMEDIATE 'INSERT INTO parse_sql_exec_t
VALUES (1)'")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(spark.table("parse_sql_exec_t").count() === 0)
+ }
+ }
+ }
+
+ test("ANALYZE still works when ConstantFolding is excluded") {
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") {
+ val j = evalStatement("SELECT 1 AS a")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "select_list" === typedColumns("a" -> jsonInt))
+ val scalar = evalCall(
+ "SELECT (SELECT parse_sql('SELECT 1 AS a', 'ANALYZE'))")
+ assert(scalar.head \ "analyze_success" === JBool(true))
+ assert(scalar.head \ "select_list" === typedColumns("a" -> jsonInt))
+ }
+ }
+
+ test("ANALYZE still works when ConstantFolding and EvaluateParseSql are
excluded") {
+ // EvaluateParseSql is not excludable, so this must match the test above.
The rule is
+ // what runs a constant ANALYZE on the driver; if it were dropped, a task
would fail.
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ ("org.apache.spark.sql.catalyst.optimizer.ConstantFolding," +
+ "org.apache.spark.sql.catalyst.optimizer.EvaluateParseSql")) {
+ val j = evalStatement("SELECT 1 AS a")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "select_list" === typedColumns("a" -> jsonInt))
+ }
+ }
+
+ test("PARSE keeps a failing constant in an untaken CASE branch under ANSI") {
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "true") {
+ val rows = sql(
+ "SELECT CASE WHEN id > 0 THEN parse_sql(CAST(1/0 AS STRING)) END FROM
range(1)")
+ .collect()
+ assert(rows.length === 1)
+ assert(rows.head.isNullAt(0))
+ }
+ }
+
+ test("ANALYZE of a failing constant in an untaken CASE branch does not fail
the query") {
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ SQLConf.ANSI_ENABLED.key -> "true") {
+ val rows = sql(
+ "SELECT CASE WHEN id > 0 THEN parse_sql(CAST(1/0 AS STRING),
'ANALYZE') END " +
+ "FROM range(1)").collect()
+ assert(rows.length === 1)
+ assert(rows.head.isNullAt(0))
+ }
+ }
+
+ test("ANALYZE attributes a MERGE source to the table it names, not the
row-level target") {
+ val catalog = "parse_sql_rl"
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ s"spark.sql.catalog.$catalog" ->
classOf[InMemoryRowLevelOperationTableCatalog].getName) {
+ try {
+ sql(s"CREATE NAMESPACE $catalog.prod")
+ sql(s"CREATE NAMESPACE $catalog.staging")
+ sql(s"CREATE TABLE $catalog.prod.t (id INT, v INT)")
+ sql(s"CREATE TABLE $catalog.staging.t (id INT, v INT)")
+ sql(s"USE $catalog.staging")
+ val j = evalStatement(
+ s"""MERGE INTO $catalog.prod.t AS tgt USING t AS src ON tgt.id =
src.id
+ |WHEN MATCHED THEN DELETE""".stripMargin)
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "target_table_references" === JArray(List(ref(catalog,
"prod", "t"))))
+ assert(j \ "source_table_references" === JArray(List(ref(catalog,
"staging", "t"))))
+ } finally {
+ sql("USE spark_catalog.default")
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$catalog")
+ }
+ }
+ }
+
+ test("ANALYZE does not report a SQL table function body as the statement's
source") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE IF NOT EXISTS parse_sql_tvf_db")
+ try {
+ withTable("parse_sql_tvf_db.t", "t") {
+ sql("CREATE TABLE parse_sql_tvf_db.t (x INT) USING parquet")
+ sql("CREATE TABLE t (x INT) USING parquet")
+ withUserDefinedFunction("parse_sql_tvf" -> false) {
+ sql("CREATE FUNCTION parse_sql_tvf() RETURNS TABLE (x INT) " +
+ "RETURN SELECT x FROM parse_sql_tvf_db.t")
+ val j = evalStatement("SELECT * FROM parse_sql_tvf() JOIN t")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(
+ ref("spark_catalog", "default", "t"))))
+ }
+ }
+ } finally {
+ sql("DROP DATABASE IF EXISTS parse_sql_tvf_db CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE reports the stored identity of a table named in another case")
{
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_case_t") {
+ sql("CREATE TABLE parse_sql_case_t (a INT) USING parquet")
+ val j = evalStatement("SELECT a FROM PARSE_SQL_CASE_T")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_case_t"))))
+ }
+ }
+ }
+
+ test("ANALYZE reports relative path-based datasource tables as format plus
path") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val dir = s"parse_sql_rel_${java.util.UUID.randomUUID()}"
+ try {
+ spark.range(1).write.mode("overwrite").parquet(dir)
+ val j = evalStatement(s"SELECT id FROM parquet.`$dir`")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(ref("parquet",
dir))))
+ } finally {
+ Utils.deleteRecursively(new java.io.File(dir))
+ }
+ }
+ }
+
+ test("ANALYZE reports the owner of a routine that a temp function shadows") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ // The temp and persistent functions share a name, so clean up both
explicitly.
+ try {
+ sql("CREATE FUNCTION parse_sql_ord_f() RETURNS INT RETURN 1")
+ sql("CREATE TEMPORARY FUNCTION parse_sql_ord_f() RETURNS INT RETURN 2")
+ val second = evalStatement("SELECT parse_sql_ord_f()")
+ assert(second \ "function_references" === JArray(List(
+ ref("system", "session", "parse_sql_ord_f"))))
+ withSQLConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key -> "last") {
+ // With `last`, the persistent function in the current schema is
found first.
+ val last = evalStatement("SELECT parse_sql_ord_f()")
+ assert(last \ "function_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_ord_f"))))
+ }
+ } finally {
+ sql("DROP TEMPORARY FUNCTION IF EXISTS parse_sql_ord_f")
+ sql("DROP FUNCTION IF EXISTS parse_sql_ord_f")
+ }
+ }
+ }
+
+ test("ANALYZE reports statement-relative code points for an error after an
emoji") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ // The emoji is one code point, so `nope` starts at code point 13
(1-based).
+ val emoji = new String(Character.toChars(0x1F600))
+ val j = evalStatement(s"SELECT '$emoji', nope")
+ assert(j \ "analyze_success" === JBool(false))
+ val context = (j \ "error" \ "queryContext")(0)
+ assert(context \ "startIndex" === JInt(13))
+ assert(context \ "stopIndex" === JInt(16))
+ assert(context \ "fragment" === JString("nope"))
+ }
+ }
+
+ test("ANALYZE reports CHAR and VARCHAR lengths in select_list") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_cv") {
+ sql("CREATE TABLE parse_sql_cv (c CHAR(3), v VARCHAR(10)) USING
parquet")
+ val j = evalStatement("SELECT c, v FROM parse_sql_cv")
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> ref("c"),
+ "type" -> JObject("name" -> JString("char"), "length" -> JInt(3))),
+ JObject("name" -> ref("v"),
+ "type" -> JObject("name" -> JString("varchar"), "length" ->
JInt(10))))))
+ }
+ }
+ }
+
+ test("ANALYZE of row-level DELETE, UPDATE and MERGE reports no rewrite
columns") {
+ val catalog = "parse_sql_rl2"
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ s"spark.sql.catalog.$catalog" ->
classOf[InMemoryRowLevelOperationTableCatalog].getName) {
+ try {
+ sql(s"CREATE NAMESPACE $catalog.ns")
+ sql(s"CREATE TABLE $catalog.ns.rl (id INT, v INT)")
+ sql(s"CREATE TABLE $catalog.ns.src (id INT, v INT)")
+ val statements = Seq(
+ s"DELETE FROM $catalog.ns.rl WHERE id = 1",
+ s"UPDATE $catalog.ns.rl SET v = 2 WHERE id = 1",
+ s"MERGE INTO $catalog.ns.rl AS t USING $catalog.ns.src AS s ON t.id
= s.id " +
+ "WHEN MATCHED THEN DELETE")
+ statements.foreach { statement =>
+ val j = evalStatement(statement)
+ assert(j \ "analyze_success" === JBool(true), statement)
+ assert(j \ "target_table_references" === JArray(List(ref(catalog,
"ns", "rl"))))
+ assert(j \ "select_list" === JNothing, statement)
+ }
+ } finally {
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$catalog")
+ }
+ }
+ }
+
+
+ test("ANALYZE reports temp view targets for global, USING and EXPLAIN
forms") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val globalDb = spark.sharedState.globalTempDB
+ val asGlobal = evalStatement("CREATE GLOBAL TEMP VIEW parse_sql_gtv AS
SELECT 1 AS a")
+ assert(asGlobal \ "target_table_references" === JArray(List(
+ ref("spark_catalog", globalDb, "parse_sql_gtv"))))
+ val usingLocal = evalStatement(
+ "CREATE TEMPORARY VIEW parse_sql_tu USING parquet OPTIONS (path
'/tmp/parse_sql_tu')")
+ assert(usingLocal \ "analyze_success" === JBool(true))
+ assert(usingLocal \ "target_table_references" === JArray(List(
+ ref("system", "session", "parse_sql_tu"))))
+ val usingGlobal = evalStatement(
+ "CREATE GLOBAL TEMPORARY VIEW parse_sql_tug USING parquet " +
+ "OPTIONS (path '/tmp/parse_sql_tug')")
+ assert(usingGlobal \ "target_table_references" === JArray(List(
+ ref("spark_catalog", globalDb, "parse_sql_tug"))))
+ val explain = evalStatement("EXPLAIN CREATE TEMPORARY VIEW parse_sql_etv
AS SELECT 1 AS a")
+ assert(explain \ "analyze_success" === JBool(true))
+ assert(explain \ "target_table_references" === JArray(List(
+ ref("system", "session", "parse_sql_etv"))))
+ assert(!spark.catalog.tableExists("parse_sql_gtv"))
+ assert(!spark.catalog.tableExists("parse_sql_etv"))
+ }
+ }
+
+ test("ANALYZE does not report a SQL scalar function body as the statement's
source") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE IF NOT EXISTS parse_sql_fn_db")
+ try {
+ withTable("parse_sql_fn_db.t", "t") {
+ sql("CREATE TABLE parse_sql_fn_db.t (x INT) USING parquet")
+ sql("CREATE TABLE t (x INT) USING parquet")
+ withUserDefinedFunction("parse_sql_fn_f" -> false) {
+ sql("CREATE FUNCTION parse_sql_fn_f() RETURNS INT " +
+ "RETURN (SELECT max(x) FROM parse_sql_fn_db.t)")
+ val j = evalStatement("SELECT parse_sql_fn_f() FROM t")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(
+ ref("spark_catalog", "default", "t"))))
+ assert(j \ "function_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_fn_f"))))
+ }
+ }
+ } finally {
+ sql("DROP DATABASE IF EXISTS parse_sql_fn_db CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE does not report a view body as the statement's source") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE IF NOT EXISTS parse_sql_vw_db")
+ try {
+ withTable("parse_sql_vw_db.t", "t") {
+ sql("CREATE TABLE parse_sql_vw_db.t (x INT) USING parquet")
+ sql("CREATE TABLE t (x INT) USING parquet")
+ withView("parse_sql_vw_v") {
+ sql("CREATE VIEW parse_sql_vw_v AS SELECT x FROM
parse_sql_vw_db.t")
+ val j = evalStatement("SELECT * FROM parse_sql_vw_v JOIN t")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_vw_v"),
+ ref("spark_catalog", "default", "t"))))
+ }
+ }
+ } finally {
+ sql("DROP DATABASE IF EXISTS parse_sql_vw_db CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE qualifies CACHE and UNCACHE TABLE targets") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_cache_t") {
+ sql("CREATE TABLE parse_sql_cache_t (a INT) USING parquet")
+ val cache = evalStatement("CACHE TABLE parse_sql_cache_t")
Review Comment:
**Medium: Several tests added for this round's fixes also pass on the
previous revision.** The name fallback already gave the right answer for
lowercase names in the current schema, so a fix in the analyzed walk only shows
with a qualified or differently cased name, or where the fallback resolves
differently. This one-part `CACHE TABLE parse_sql_cache_t` reported the same
target before; the bug needed something like `CACHE TABLE
default.PARSE_SQL_CACHE_T`, which previously came out as
`["default","PARSE_SQL_CACHE_T"]`. The ALTER VIEW source and the CTAS / v1
INSERT tests use lowercase names, so the analyzed `AlterViewAsCommand`,
CTAS-query and INSERT-target arms are unpinned; `... FROM PARSE_SQL_VS_T` and
`INSERT INTO PARSE_SQL_REF_T ...` would pin them. Every name in the row-level
DELETE / insert-only MERGE test is three-part, and a three-part name that
matches nothing falls back to the same parts, so that test passes with
`RewriteRead` reverted to the old filter or with `InsertO
nlyMerge` removed. Two inputs that fail without them: under `USE cat.ns` with
a temp view `rl` and `sessionOrder=last`, a self-DELETE whose subquery reads
`rl`; and under `USE cat.staging`, an insert-only MERGE into `cat.prod.t` whose
USING subquery reads both `t` and `cat.staging.t`. In the temp-view test,
`tableExists("parse_sql_gtv")` can't see a global temp view
(`tableExists(globalDb, "parse_sql_gtv")` can), and the two USING views have no
such check.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -342,10 +472,27 @@ 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 =>
+ // The named object is the reference; do not collect tables from its
body.
+ case w: V2WriteCommand =>
+ visitPlan(w.table, scope, TableRefRole.Target)(f)
+ val queryRole = w match {
+ case _: ReplaceData | _: WriteDelta | _: InsertOnlyMerge =>
TableRefRole.RewriteRead
+ case _ => TableRefRole.Source
+ }
+ visitPlan(w.query, nextScope, queryRole)(f)
+ case a: AlterViewAs =>
+ visitPlan(a.child, scope, TableRefRole.Target)(f)
+ visitPlan(a.query, nextScope, TableRefRole.Source)(f)
+ case w: WithCTE =>
+ w.cteDefs.foreach(d => visitPlan(d, nextScope, TableRefRole.Source)(f))
+ visitPlan(w.plan, nextScope, role)(f)
case SubqueryAlias(_, child) =>
visitPlan(child, nextScope, role)(f)
+ case _: org.apache.spark.sql.catalyst.analysis.SQLFunctionNode =>
Review Comment:
**Low: Please restore the explicit `SQLFunctionNode` import instead of this
inline name.** My round-3 note that the import duplicated the `analysis._`
wildcard was wrong: sql/core also defines
`org.apache.spark.sql.catalyst.plans.logical.SQLFunctionNode`, and
`plans.logical._` is wildcard-imported too, so the explicit import was what
disambiguated. The class matched here is the right one (it is what
`SessionCatalog` builds); `import
org.apache.spark.sql.catalyst.analysis.SQLFunctionNode` with `case _:
SQLFunctionNode =>` reads better.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -282,6 +409,9 @@ object ParseSqlResult {
private object TableRefRole {
case object Target extends TableRefRole
case object Source extends TableRefRole
+ // The query of a row-level rewrite or insert-only MERGE. It re-reads the
write target, so
+ // the role marks the re-read for the walk. Subqueries in the query are
Source again.
+ case object RewriteRead extends TableRefRole
Review Comment:
**Low: `RewriteRead` also drops a read of the target that the user wrote in
the MERGE source.** The role covers the whole rewrite query, and the MERGE
source is a child of that query rather than a subquery, so in `MERGE INTO
cat.ns.rl USING rl ...` (or a USING subquery that reads the target) the user's
read is filtered and the name falls back: with `USE cat.ns`, a temp view `rl`
and `sessionOrder=last`, the source is reported as `system.session.rl` although
analysis read `cat.ns.rl`. Recognizing the rewrite's own reads structurally
would keep the user's: for `ReplaceData` / `WriteDelta` the re-read is a
`DataSourceV2Relation` over a `RowLevelOperationTable`, and for
`InsertOnlyMerge` it is the right side of its LeftAnti join. The comment above
could also say that the MERGE source is not one of the subqueries that go back
to `Source`.
##########
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:
The `isFileReference` change now reaches this catch from an ordinary
statement (see the new thread there: a schema named `kafka` makes
`lookupDataSource` throw), so that would make a good input for the per-name
fallback test.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +711,328 @@ 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) and the session
catalog current, a
+ * two-part name whose first part is a data source and not a session
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 = {
+ val mgr = session.sessionState.catalogManager
+ parts match {
+ case Seq(format, _)
+ if
mgr.currentCatalog.name().equalsIgnoreCase(CatalogManager.SESSION_CATALOG_NAME)
=>
+ val isDataSource =
+ try {
+ DataSource.lookupDataSource(format, session.sessionState.conf)
+ true
+ } catch {
+ case _: ClassNotFoundException => false
+ }
+ isDataSource && !session.sessionState.catalog.databaseExists(format)
+ case _ => false
+ }
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = plan match {
+ case ExplainCommand(inner, _) =>
+ qualifyTarget(parts, inner, session)
+ 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 =>
+ // Persistent DDL never names a temp view, even if one shares the name.
+ qualifyWithCurrentNamespace(parts, session)
+ case _ =>
+ qualifySource(parts, session)
+ }
+
+ private def qualifySource(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ if (CatalogManager.isSessionQualifiedViewName(parts)) {
Review Comment:
**Low: Every two-part `session.x` is now reported as a temp view.** My
round-3 comment should have been limited to the three-part form:
`system.session.x` can only name a temp view, but `session.x` also names a
table `x` in a schema called `session`, which is how the analyzer resolves it
when no temp view `x` exists. With `CREATE DATABASE session` and a table
`session.events`, `TRUNCATE TABLE session.events` now reports
`["system","session","events"]` instead of `spark_catalog.session.events`.
Using `CatalogManager.isFullyQualifiedSystemSessionViewName(parts)` here, and
leaving two-part names to the `isTempView` check below, would fix it.
##########
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:
For this PR I'd document the gap with a test that pins today's behavior
(cached and streamed plans run ANALYZE in a cloned session without the v2
current catalog or namespace), and file a follow-up to record the invoker's
catalog and namespace when the call is analyzed, which overlaps with the
overrides you mentioned for SPARK-60622.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +711,328 @@ 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) and the session
catalog current, a
+ * two-part name whose first part is a data source and not a session
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 = {
+ val mgr = session.sessionState.catalogManager
+ parts match {
+ case Seq(format, _)
+ if
mgr.currentCatalog.name().equalsIgnoreCase(CatalogManager.SESSION_CATALOG_NAME)
=>
+ val isDataSource =
+ try {
+ DataSource.lookupDataSource(format, session.sessionState.conf)
+ true
+ } catch {
+ case _: ClassNotFoundException => false
+ }
+ isDataSource && !session.sessionState.catalog.databaseExists(format)
+ case _ => false
+ }
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = plan match {
+ case ExplainCommand(inner, _) =>
+ qualifyTarget(parts, inner, session)
+ 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 =>
+ // Persistent DDL never names a temp view, even if one shares the name.
+ qualifyWithCurrentNamespace(parts, session)
+ case _ =>
+ qualifySource(parts, session)
+ }
+
+ private def qualifySource(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ if (CatalogManager.isSessionQualifiedViewName(parts)) {
+ // `session.v` and `system.session.v` name a temp view even when it does
not exist.
+ sessionIdent(parts.last)
+ } else 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))
+ // Persistent in the catalog the statement resolves names in, as
FunctionResolution does.
Review Comment:
**Low: This comment claims parity with `FunctionResolution` that the code
doesn't have.** `FunctionResolution` walks the SQL path candidates, while this
checks only the current namespace (round 3 suggested the opposite note, that
the path isn't followed yet), and `Try(...).getOrElse(false)` treats any
failure as "no function", where `FunctionResolution` treats only the not-found
and forbidden cases as absent and rethrows other errors. The new v2
`FunctionCatalog` branch also has no test: with `default.f` and a temp `f`,
`SELECT f()` under `USE testcat.ns` and `sessionOrder=last` should report
`system.session.f`.
--
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]
