dtenedor commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4460514110
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +660,183 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(
+ analyzed: LogicalPlan,
+ parsed: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(analyzed)
+ if (query.resolved && query.output.nonEmpty) {
+ query.output.map { attr =>
+ JObject(
+ "name" -> partsToJArray(Seq(attr.name)),
+ "type" -> DescribeRelationJsonCommand.jsonType(
+
CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType)))
+ }
+ } else {
+ collectSelectList(parsed)
+ }
+ }
+
+ private def uniqueIdentities(
+ ids: Seq[Seq[String]],
+ session: ClassicSparkSession): Seq[Seq[String]] = {
+ val sensitive = session.sessionState.conf.caseSensitiveAnalysis
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ ids.filter { parts =>
+ val key = if (sensitive) parts else parts.map(_.toLowerCase(Locale.ROOT))
+ seen.add(key)
+ }
+ }
+
+ private def sessionIdent(name: String): Seq[String] =
+ Seq(CatalogManager.SYSTEM_CATALOG_NAME, CatalogManager.SESSION_NAMESPACE,
name)
+
+ private def catalogIdentParts(catalog: String, ident: Identifier):
Seq[String] =
+ Seq(catalog) ++ ident.namespace() :+ ident.name()
+
+ private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] =
+ (r.catalog, r.identifier) match {
+ case (Some(catalog), Some(ident)) =>
Some(catalogIdentParts(catalog.name(), ident))
+ case _ => None
+ }
+
+ private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] =
+ r.catalogTable.map { t =>
+ val id = t.identifier
+ if (id.catalog.isEmpty && id.database.isDefined) {
+ CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id)
+ } else {
+ tableIdentifierParts(id)
+ }
+ }.orElse {
+ r.relation match {
+ case fs: HadoopFsRelation =>
+ fs.location.rootPaths.headOption.map(p => Seq(p.toString))
+ case _ => None
+ }
+ }
+
+ 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 ns = ident.namespace()
+ val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
+ if (ns.length == 1 && ns.head.equalsIgnoreCase(globalTemp)) {
+ Seq(CatalogManager.SESSION_CATALOG_NAME, ns.head, ident.name())
+ } else if (ns.isEmpty) {
+ sessionIdent(ident.name())
+ } else {
+ catalogIdentParts(CatalogManager.SESSION_CATALOG_NAME, ident)
+ }
+ }
+
+ private def isTempViewCreate(plan: LogicalPlan): Boolean = plan match {
+ case c: CreateViewCommand =>
+ c.viewType == LocalTempView || c.viewType == GlobalTempView
+ case _: CreateTempViewUsing => true
+ case _ => false
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = {
+ if (isTempViewCreate(plan)) {
+ sessionIdent(parts.last)
+ } else {
+ qualifyWithCurrentNamespace(parts, session)
+ }
+ }
+
+ private def qualifyRoutine(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ val catalog = session.sessionState.catalog
+ val tempFirst =
+
session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER) ==
"first"
+ def isTemp(name: String): Boolean =
+ catalog.isTemporaryFunction(FunctionIdentifier(name))
+ def isBuiltin(name: String): Boolean = {
+ val id = FunctionIdentifier(name)
+ FunctionRegistry.builtin.functionExists(id) ||
+ TableFunctionRegistry.builtin.functionExists(id)
+ }
+ def isSessionBuiltin(name: String): Boolean = {
Review Comment:
**High: `isSessionBuiltin` fails the query for any unqualified function that
is neither builtin nor temp.** In ANALYZE, every statement's function names go
through `qualifyRoutine` (L286-288), and this probes the session registries
with an unqualified `FunctionIdentifier(name)`. Those registries are
`FunctionRegistry.builtin.clone()`, and `SimpleFunctionRegistry.clone()` drops
the `BuiltinRegistryMixin` that qualifies the name, so the 3-part assert in
`normalizeFuncName` fires: `SELECT parse_sql('SELECT no_such_fn(1)',
'ANALYZE')`, or an unqualified call to a persistent or extension-injected
function (the case this check was added for), fails with `INTERNAL_ERROR`
instead of returning JSON. Probing with
`FunctionRegistry.builtinFunctionIdentifier(name)`, as `SessionCatalog` does
for extension functions, fixes the crash; taking the owner from
`FunctionResolution`'s candidate walk would also fix the remaining PATH and
`sessionOrder=last` mislabels here.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -342,6 +455,14 @@ object ParseSqlResult {
visitPlan(cts.plan, nextScope, TableRefRole.Source)(f)
case _: CacheTable =>
// Name is taken from multipartIdentifier or the non-child `table`
slot.
+ case _: View =>
Review Comment:
**Medium: The walk stops at `View` but descends into other bodies the
statement doesn't name.** SQL table-function bodies (`SQLFunctionNode`), scalar
SQL UDF bodies (`SQLScalarFunction`) and metric views (`ResolvedMetricView`)
are still in the plan after analysis, so `SELECT * FROM tf()` reports the
tables inside `tf`, and a metric-view query reports the view's source table
instead of the metric view. Row-level DML rewritten to `ReplaceData` /
`WriteDelta` also lists its own target as a source. Stopping at those nodes the
way `View` does (while still collecting table-function arguments), and skipping
the read that the row-level rewrite adds, would keep `source_table_references`
to what the statement names.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -452,11 +582,52 @@ object ParseSqlResult {
case u: UnresolvedView => add(u.multipartIdentifier)
case u: UnresolvedTableOrView => add(u.multipartIdentifier)
case u: UnresolvedIdentifier if role == TableRefRole.Target =>
add(u.nameParts)
- case c: CreateViewCommand => addTarget(tableIdentifierParts(c.name))
- case c: CreateTempViewUsing =>
addTarget(tableIdentifierParts(c.tableIdent))
+ case c: CreateViewCommand =>
Review Comment:
**Medium: These arms also run in PARSE, so PARSE output changed for
temp-view DDL.** The parser emits `CreateViewCommand` and `CreateTempViewUsing`
directly, so `parse_sql('CREATE TEMP VIEW v AS SELECT 1')` now reports the
target as `["system","session","v"]` (master: `["v"]`) and a global temp view
as `["spark_catalog","global_temp","v"]`, while `SELECT * FROM v` in the same
batch still reports `["v"]`. The description says PARSE stays as today, and no
PARSE test covers temp-view DDL. I'd keep the PARSE arms syntax-only, apply
session and global-temp naming only to analyzed plans, and add PARSE goldens
for `CREATE [GLOBAL] TEMP VIEW` with `AS` and `USING`.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -65,148 +114,215 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse): String = {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- val starts = charStarts(sql, statements)
- compact(render(JArray(statements.zip(starts).map {
- case (segment, start) => parseStatement(segment, start)
- }.toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(statements, spans)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each against the invoker session.
+ * Statements are not executed, so one does not see the effect of an earlier
+ * one. Analysis failures stay JSON (`analyze_success` false) per statement.
+ */
+ private def analyzeBatch(
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan]): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement, charStart: Int):
JObject = {
+ /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the
call. */
+ private def isInternalError(e: SparkThrowable): Boolean =
+ Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR"))
+
+ /** Character span of a statement object in the original batch. */
+ private case class StatementSpan(start: Int, length: Int)
+
+ /**
+ * Code-point `start` for each statement in one forward pass over `batch`.
+ * Statements are in source order, so prefix counts are not repeated.
+ */
+ private def statementSpans(
+ batch: String,
+ statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = {
+ var lastUtf16 = 0
+ var lastCp = 0
+ statements.map { segment =>
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ if (utf16 >= lastUtf16) {
+ lastCp += batch.codePointCount(lastUtf16, utf16)
+ lastUtf16 = utf16
+ } else {
+ lastCp = batch.codePointCount(0, utf16)
+ lastUtf16 = utf16
+ }
+ StatementSpan(lastCp + 1, charLength(segment))
+ }
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ segment: PositionedSqlStatement,
+ span: StatementSpan): ParsedStatement = {
val sql = segment.statement
- val length = sql.codePointCount(0, sql.length)
try {
// Do not inherit the outer query's origin from the parse_sql expression.
- // Errors and parsed nodes refer to the individual statement. Parser
- // origins are code-point offsets, so this one is too.
+ // Errors and parsed nodes refer to the individual statement.
val origin = if (sql.nonEmpty) {
Origin(
startIndex = Some(0),
- stopIndex = Some(length - 1),
+ stopIndex = Some(sql.codePointCount(0, sql.length) - 1),
sqlText = Some(sql))
} else {
Origin(sqlText = Some(sql))
}
CurrentOrigin.withOrigin(origin) {
- val plan = parser.get().parsePlan(sql)
- successJson(plan, charStart, length)
+ Parsed(segment, parser.get().parsePlan(sql))
}
} catch {
- // User-facing parse / scripting failures become JSON; everything else
fails.
case e: ParseException =>
- errorJson(e, segment, charStart, length)
+ ParseFailed(errorJson(e, span))
case e: SqlScriptingException =>
- errorJson(e, segment, charStart, length)
+ ParseFailed(errorJson(e, span))
+ }
+ }
+
+ /** Analyze `plan` without executing it. Scripts never reach QueryExecution.
*/
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ plan match {
+ case _: CompoundBody =>
+ (None, Some(QueryExecutionErrors.parseSqlAnalyzeUnsupportedScript()))
+ case _ =>
+ try {
+ val toAnalyze = plan match {
+ case s: SupervisingCommand =>
+ s.withTransformedSupervisedPlan { inner =>
+ session.sessionState.executePlan(inner,
CommandExecutionMode.SKIP).analyzed
+ }
+ case other => other
+ }
+ val analyzed =
+ session.sessionState.executePlan(toAnalyze,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), 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) {
+ // Functions and parameter markers always come from the parse tree.
Relation
+ // identities come from the analyzed plan when analysis succeeded.
+ val parseRefs = collectPlanReferences(plan)
+ val tableRefs = analyzed.map(collectPlanReferences).getOrElse(parseRefs)
Review Comment:
**High: ANALYZE now reports fewer references than PARSE for common
session-catalog statements.** When analysis succeeds, the parse references are
replaced by a walk of the analyzed plan, which only follows `children` and
subqueries, while session-catalog commands keep their tables in fields or
`innerChildren`. For parquet tables, `INSERT INTO t SELECT a FROM s` loses its
target (it is in `InsertIntoHadoopFsRelationCommand.catalogTable`), a CTAS
loses both target and source, `CREATE [TEMP] VIEW v AS SELECT ... FROM s` loses
`s`, `ALTER VIEW ... AS` reports nothing, `TRUNCATE`, `REFRESH`, `ALTER TABLE
... ADD COLUMNS`, `RENAME`, `ANALYZE TABLE` and `UNCACHE` lose their target,
and `CACHE TABLE t` reports the raw `["t"]`; v2 tables and DESCRIBE are fine.
The CTAS and mixed-batch tests run some of these but don't assert references.
I'd keep the parse references as the authority for which references exist and
in which role, use the analyzed plan only to qualify them, and add a table-dr
iven test that asserts the exact references for each statement shape.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +660,183 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(
+ analyzed: LogicalPlan,
+ parsed: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(analyzed)
+ if (query.resolved && query.output.nonEmpty) {
+ query.output.map { attr =>
+ JObject(
+ "name" -> partsToJArray(Seq(attr.name)),
+ "type" -> DescribeRelationJsonCommand.jsonType(
+
CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType)))
+ }
+ } else {
+ collectSelectList(parsed)
+ }
+ }
+
+ private def uniqueIdentities(
+ ids: Seq[Seq[String]],
+ session: ClassicSparkSession): Seq[Seq[String]] = {
+ val sensitive = session.sessionState.conf.caseSensitiveAnalysis
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ ids.filter { parts =>
+ val key = if (sensitive) parts else parts.map(_.toLowerCase(Locale.ROOT))
+ seen.add(key)
+ }
+ }
+
+ private def sessionIdent(name: String): Seq[String] =
+ Seq(CatalogManager.SYSTEM_CATALOG_NAME, CatalogManager.SESSION_NAMESPACE,
name)
+
+ private def catalogIdentParts(catalog: String, ident: Identifier):
Seq[String] =
+ Seq(catalog) ++ ident.namespace() :+ ident.name()
+
+ private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] =
+ (r.catalog, r.identifier) match {
+ case (Some(catalog), Some(ident)) =>
Some(catalogIdentParts(catalog.name(), ident))
+ case _ => None
+ }
+
+ private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] =
+ r.catalogTable.map { t =>
+ val id = t.identifier
+ if (id.catalog.isEmpty && id.database.isDefined) {
+ CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id)
+ } else {
+ tableIdentifierParts(id)
+ }
+ }.orElse {
+ r.relation match {
+ case fs: HadoopFsRelation =>
+ fs.location.rootPaths.headOption.map(p => Seq(p.toString))
Review Comment:
**Low: Path-based relations lose their shape.** This reports only the first
root path as a one-part name (for example `["file:/tmp/p"]`), dropping the
format and any further root paths after glob expansion, while PARSE reports
`["parquet","/tmp/p"]`; a one-part path also looks like an unqualified table
name. Reporting `[format, path]` as written, with the analyzed plan only
confirming it resolved, and documenting that shape would keep the two modes
consistent.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.procedures.{
+ BoundProcedure,
+ ProcedureParameter,
+ UnboundProcedure
+}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataTypes, StructType}
+
+class ParseSqlAnalyzeSuite extends QueryTest with SharedSparkSession {
+
+ private def evalJson(sql: String, mode: String = "ANALYZE"): JValue = {
+ val result = spark.sql(s"SELECT parse_sql($sql,
'$mode')").head().getString(0)
+ parse(result)
+ }
+
+ private def evalStatement(sql: String, mode: String = "ANALYZE"): JObject = {
+ evalJson("'" + sql.replace("'", "''") + "'", mode) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ }
+
+ private def jsonType(name: String): JObject = JObject("name" ->
JString(name))
+ private val jsonInt: JObject = jsonType("int")
+ private val jsonBigint: JObject = jsonType("bigint")
+ private val jsonBoolean: JObject = jsonType("boolean")
+ private val jsonString: JObject = JObject(
+ "name" -> JString("string"),
+ "collation" -> JString("UTF8_BINARY"))
+
+ test("ANALYZE types selected columns and PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val analyzed = evalStatement("SELECT 1 AS a", "ANALYZE")
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+
+ val parsed = evalStatement("SELECT 1 AS a", "PARSE")
+ assert(parsed \ "analyze_success" === JNothing)
+ assert(parsed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a")))))))
+ }
+ }
+
+ test("ANALYZE fully qualifies tables and builtin routines") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_analyze_t") {
+ sql("CREATE TABLE parse_sql_analyze_t (a INT, b STRING) USING parquet")
+ val j = evalStatement(
+ "SELECT count(a) AS c FROM parse_sql_analyze_t", "analyze")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("default"),
JString("parse_sql_analyze_t"))))))
+ assert(j \ "function_references" === JArray(List(JArray(List(
+ JString("system"), JString("builtin"), JString("count"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("c"))), "type" ->
jsonBigint))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies local temp views as system.session") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTempView("parse_sql_analyze_v") {
+ sql("CREATE OR REPLACE TEMP VIEW parse_sql_analyze_v AS SELECT 1 AS a,
'x' AS b")
+ val j = evalStatement("SELECT a, b FROM parse_sql_analyze_v")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("system"), JString("session"),
JString("parse_sql_analyze_v"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt),
+ JObject("name" -> JArray(List(JString("b"))), "type" ->
jsonString))))
+ }
+ }
+ }
+
+ test("ANALYZE keeps parse metadata when analysis fails") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val j = evalStatement("SELECT a FROM missing_parse_sql_analyze_t")
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(j \ "statement_identifier" === JString("SELECT"))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"),
+ JString("default"),
+ JString("missing_parse_sql_analyze_t"))))))
+ val errorClass = (j \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(errorClass == "TABLE_OR_VIEW_NOT_FOUND" ||
+ errorClass.contains("TABLE_OR_VIEW_NOT_FOUND"))
+ }
+ }
+
+ test("ANALYZE inherits the invoker schema") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE parse_sql_ctx_a")
+ sql("CREATE DATABASE parse_sql_ctx_b")
+ try {
+ sql("CREATE TABLE parse_sql_ctx_a.t (a INT) USING parquet")
+ sql("USE parse_sql_ctx_b")
+ try {
+ val missing = evalStatement("SELECT a FROM t")
+ assert(missing \ "analyze_success" === JBool(false))
+ assert(missing \ "source_table_references" ===
JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_b"),
JString("t"))))))
+ } finally {
+ sql("USE parse_sql_ctx_a")
+ }
+ try {
+ val found = evalStatement("SELECT a FROM t")
+ assert(found \ "analyze_success" === JBool(true))
+ assert(found \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_a"),
JString("t"))))))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
+ "parse_sql_ctx_a")
+ } finally {
+ sql("USE default")
+ }
+ } finally {
+ sql("DROP TABLE IF EXISTS parse_sql_ctx_a.t")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_a CASCADE")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_b CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE named mode argument") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | mode => 'ANALYZE')""".stripMargin).head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+ }
+ }
+
+ test("ANALYZE without parameter marker types leaves the query unresolved") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql("SELECT parse_sql('SELECT abs(?) AS x', 'ANALYZE')")
+ .head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(false))
+ }
+ }
+
+ private def evalBatch(sql: String): List[JObject] =
+ evalJson("'" + sql.replace("'", "''") + "'") match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def typedColumns(columns: (String, JObject)*): JArray =
JArray(columns.map {
+ case (name, dataType) => JObject("name" -> JArray(List(JString(name))),
"type" -> dataType)
+ }.toList)
+
+ private def ref(parts: String*): JArray = JArray(parts.map(JString).toList)
+
+ /** Run a full `SELECT parse_sql(...)` and return its statement objects. */
+ private def evalCall(query: String): List[JObject] =
+ parse(spark.sql(query).head().getString(0)) match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def errorClass(statement: JObject): JValue = statement \ "error" \
"errorClass"
+
+ test("ANALYZE does not run CALL, directly or through EXECUTE IMMEDIATE") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ spark.conf.set("spark.sql.catalog.parse_sql_cat",
classOf[InMemoryCatalog].getName)
+ try {
+ val catalog =
spark.sessionState.catalogManager.catalog("parse_sql_cat")
+ .asInstanceOf[InMemoryCatalog]
+ catalog.createProcedure(
+ Identifier.of(Array("ns"), "record"),
ParseSqlAnalyzeSuite.UnboundRecord)
+ ParseSqlAnalyzeSuite.calls.set(0)
+ val results = evalBatch(
+ """CALL parse_sql_cat.ns.record(1);
+ |EXECUTE IMMEDIATE 'CALL
parse_sql_cat.ns.record(2)'""".stripMargin)
+ assert(results.map(_ \ "parse_success") === List(JBool(true),
JBool(true)))
+ assert(results.map(_ \ "analyze_success") === List(JBool(true),
JBool(true)))
+ assert(ParseSqlAnalyzeSuite.calls.get === 0)
+ sql("CALL parse_sql_cat.ns.record(3)").collect()
+ assert(ParseSqlAnalyzeSuite.calls.get === 1)
+ } finally {
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf("spark.sql.catalog.parse_sql_cat")
+ }
+ }
+ }
+
+ test("ANALYZE inside a SQL script does not see the script's locals") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """BEGIN
+ | DECLARE parse_sql_outer INT DEFAULT 5;
+ | SELECT parse_sql('SELECT parse_sql_outer AS x', 'ANALYZE');
+ |END""".stripMargin).head().getString(0)
+ val statement = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(statement \ "analyze_success" === JBool(false))
+ val cls = (statement \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(cls === "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION" ||
+ cls === "UNRESOLVED_COLUMN.WITH_SUGGESTION" ||
+ cls.startsWith("UNRESOLVED_COLUMN"))
+ }
+ }
+
+ test("ANALYZE requires constant arguments; PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val query = "SELECT parse_sql(c, '%s') FROM VALUES ('SELECT 1') AS t(c)"
+ val e =
intercept[AnalysisException](spark.sql(query.format("ANALYZE")).collect())
+ assert(e.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ assert(spark.sql(query.format("PARSE")).collect().length === 1)
+ val modeCol = intercept[AnalysisException](
+ spark.sql("SELECT parse_sql(c, m) FROM VALUES ('SELECT 1', 'PARSE') AS
t(c, m)")
+ .collect())
+ assert(modeCol.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ }
+ }
+
+ test("ANALYZE does not execute SQL scripts") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_script_t", "parse_sql_script_u") {
+ sql("CREATE TABLE parse_sql_script_t (a INT) USING parquet")
+ sql("CREATE TABLE parse_sql_script_u (a INT) USING parquet")
+ val schema = spark.sql("SELECT current_schema()").head().getString(0)
+ val j = evalStatement(
+ """BEGIN
+ | INSERT INTO parse_sql_script_t VALUES (1);
+ | DROP TABLE parse_sql_script_u;
+ |END""".stripMargin)
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(errorClass(j) ===
JString("PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT"))
+ assert(spark.table("parse_sql_script_t").count() === 0)
+ assert(spark.catalog.tableExists("parse_sql_script_u"))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
schema)
+ }
+ }
+ }
+
+ test("ANALYZE reports range as system.builtin and de-duplicates count") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_dedup_t") {
+ sql("CREATE TABLE parse_sql_dedup_t (a INT, b INT) USING parquet")
+ val rangeStmt = evalStatement("SELECT * FROM range(3)")
+ assert(rangeStmt \ "analyze_success" === JBool(true))
+ assert(rangeStmt \ "function_references" === JArray(List(
+ ref("system", "builtin", "range"))))
+ val counts = evalStatement(
+ "SELECT count(a), COUNT(b) FROM parse_sql_dedup_t")
+ assert(counts \ "analyze_success" === JBool(true))
+ assert(counts \ "function_references" === JArray(List(
+ ref("system", "builtin", "count"))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies DDL targets by statement kind") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val tempView = evalStatement("CREATE TEMPORARY VIEW parse_sql_ddl_v AS
SELECT 1 AS a")
+ assert(tempView \ "analyze_success" === JBool(true))
+ assert(tempView \ "target_table_references" === JArray(List(
+ ref("system", "session", "parse_sql_ddl_v"))))
+ assert(!spark.catalog.tableExists("parse_sql_ddl_v"))
+ withTempView("parse_sql_ddl_t") {
+ sql("CREATE TEMP VIEW parse_sql_ddl_t AS SELECT 1 AS a")
+ val table = evalStatement("CREATE TABLE parse_sql_ddl_t (a INT) USING
parquet")
+ assert(table \ "analyze_success" === JBool(true))
+ assert(table \ "target_table_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_ddl_t"))))
+ }
+ }
+ }
+
+ test("ANALYZE types session-catalog CTAS query columns") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ 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") {
+ 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") {
+ val explain = evalStatement("EXPLAIN SELECT 1 AS a")
+ assert(explain \ "analyze_success" === JBool(true))
+ assert(explain \ "select_list" === typedColumns("a" -> jsonInt))
+ 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 from the analyzed plan") {
+ 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))
+ val sources = (j \ "source_table_references").asInstanceOf[JArray].arr
+ assert(sources.nonEmpty)
+ val parts =
sources.head.asInstanceOf[JArray].arr.map(_.asInstanceOf[JString].s)
+ assert(parts.exists(_.contains(path)) || parts == Seq("parquet", path))
Review Comment:
**Medium: This assertion accepts several different shapes.**
`parts.exists(_.contains(path)) || parts == Seq("parquet", path)` on the first
reference passes for the current `["file:..."]`, for the parse-time
`["parquet", path]`, and for the old `["spark_catalog","parquet",path]`, so it
can't catch a regression in the shape. Comparing the full
`source_table_references` array against the intended shape would.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -65,148 +114,215 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse): String = {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- val starts = charStarts(sql, statements)
- compact(render(JArray(statements.zip(starts).map {
- case (segment, start) => parseStatement(segment, start)
- }.toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(statements, spans)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each against the invoker session.
+ * Statements are not executed, so one does not see the effect of an earlier
+ * one. Analysis failures stay JSON (`analyze_success` false) per statement.
+ */
+ private def analyzeBatch(
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan]): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement, charStart: Int):
JObject = {
+ /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the
call. */
+ private def isInternalError(e: SparkThrowable): Boolean =
+ Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR"))
+
+ /** Character span of a statement object in the original batch. */
+ private case class StatementSpan(start: Int, length: Int)
+
+ /**
+ * Code-point `start` for each statement in one forward pass over `batch`.
+ * Statements are in source order, so prefix counts are not repeated.
+ */
+ private def statementSpans(
+ batch: String,
+ statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = {
+ var lastUtf16 = 0
+ var lastCp = 0
+ statements.map { segment =>
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ if (utf16 >= lastUtf16) {
+ lastCp += batch.codePointCount(lastUtf16, utf16)
+ lastUtf16 = utf16
+ } else {
+ lastCp = batch.codePointCount(0, utf16)
+ lastUtf16 = utf16
+ }
+ StatementSpan(lastCp + 1, charLength(segment))
+ }
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ segment: PositionedSqlStatement,
+ span: StatementSpan): ParsedStatement = {
val sql = segment.statement
- val length = sql.codePointCount(0, sql.length)
try {
// Do not inherit the outer query's origin from the parse_sql expression.
- // Errors and parsed nodes refer to the individual statement. Parser
- // origins are code-point offsets, so this one is too.
+ // Errors and parsed nodes refer to the individual statement.
val origin = if (sql.nonEmpty) {
Origin(
startIndex = Some(0),
- stopIndex = Some(length - 1),
+ stopIndex = Some(sql.codePointCount(0, sql.length) - 1),
sqlText = Some(sql))
} else {
Origin(sqlText = Some(sql))
}
CurrentOrigin.withOrigin(origin) {
- val plan = parser.get().parsePlan(sql)
- successJson(plan, charStart, length)
+ Parsed(segment, parser.get().parsePlan(sql))
}
} catch {
- // User-facing parse / scripting failures become JSON; everything else
fails.
case e: ParseException =>
- errorJson(e, segment, charStart, length)
+ ParseFailed(errorJson(e, span))
case e: SqlScriptingException =>
- errorJson(e, segment, charStart, length)
+ ParseFailed(errorJson(e, span))
+ }
+ }
+
+ /** Analyze `plan` without executing it. Scripts never reach QueryExecution.
*/
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ plan match {
+ case _: CompoundBody =>
+ (None, Some(QueryExecutionErrors.parseSqlAnalyzeUnsupportedScript()))
+ case _ =>
+ try {
+ val toAnalyze = plan match {
+ case s: SupervisingCommand =>
Review Comment:
**Low: Nested supervising commands aren't analyzed.** Only the outermost
supervised plan goes through `withTransformedSupervisedPlan`, so `EXPLAIN
EXPLAIN SELECT nope FROM missing_t` still reports `analyze_success: true` with
unqualified names. Applying the transformation recursively would cover it.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.procedures.{
+ BoundProcedure,
+ ProcedureParameter,
+ UnboundProcedure
+}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataTypes, StructType}
+
+class ParseSqlAnalyzeSuite extends QueryTest with SharedSparkSession {
+
+ private def evalJson(sql: String, mode: String = "ANALYZE"): JValue = {
+ val result = spark.sql(s"SELECT parse_sql($sql,
'$mode')").head().getString(0)
+ parse(result)
+ }
+
+ private def evalStatement(sql: String, mode: String = "ANALYZE"): JObject = {
+ evalJson("'" + sql.replace("'", "''") + "'", mode) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ }
+
+ private def jsonType(name: String): JObject = JObject("name" ->
JString(name))
+ private val jsonInt: JObject = jsonType("int")
+ private val jsonBigint: JObject = jsonType("bigint")
+ private val jsonBoolean: JObject = jsonType("boolean")
+ private val jsonString: JObject = JObject(
+ "name" -> JString("string"),
+ "collation" -> JString("UTF8_BINARY"))
+
+ test("ANALYZE types selected columns and PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val analyzed = evalStatement("SELECT 1 AS a", "ANALYZE")
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+
+ val parsed = evalStatement("SELECT 1 AS a", "PARSE")
+ assert(parsed \ "analyze_success" === JNothing)
+ assert(parsed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a")))))))
+ }
+ }
+
+ test("ANALYZE fully qualifies tables and builtin routines") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_analyze_t") {
+ sql("CREATE TABLE parse_sql_analyze_t (a INT, b STRING) USING parquet")
+ val j = evalStatement(
+ "SELECT count(a) AS c FROM parse_sql_analyze_t", "analyze")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("default"),
JString("parse_sql_analyze_t"))))))
+ assert(j \ "function_references" === JArray(List(JArray(List(
+ JString("system"), JString("builtin"), JString("count"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("c"))), "type" ->
jsonBigint))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies local temp views as system.session") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTempView("parse_sql_analyze_v") {
+ sql("CREATE OR REPLACE TEMP VIEW parse_sql_analyze_v AS SELECT 1 AS a,
'x' AS b")
+ val j = evalStatement("SELECT a, b FROM parse_sql_analyze_v")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("system"), JString("session"),
JString("parse_sql_analyze_v"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt),
+ JObject("name" -> JArray(List(JString("b"))), "type" ->
jsonString))))
+ }
+ }
+ }
+
+ test("ANALYZE keeps parse metadata when analysis fails") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val j = evalStatement("SELECT a FROM missing_parse_sql_analyze_t")
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(j \ "statement_identifier" === JString("SELECT"))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"),
+ JString("default"),
+ JString("missing_parse_sql_analyze_t"))))))
+ val errorClass = (j \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(errorClass == "TABLE_OR_VIEW_NOT_FOUND" ||
+ errorClass.contains("TABLE_OR_VIEW_NOT_FOUND"))
+ }
+ }
+
+ test("ANALYZE inherits the invoker schema") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE parse_sql_ctx_a")
+ sql("CREATE DATABASE parse_sql_ctx_b")
+ try {
+ sql("CREATE TABLE parse_sql_ctx_a.t (a INT) USING parquet")
+ sql("USE parse_sql_ctx_b")
+ try {
+ val missing = evalStatement("SELECT a FROM t")
+ assert(missing \ "analyze_success" === JBool(false))
+ assert(missing \ "source_table_references" ===
JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_b"),
JString("t"))))))
+ } finally {
+ sql("USE parse_sql_ctx_a")
+ }
+ try {
+ val found = evalStatement("SELECT a FROM t")
+ assert(found \ "analyze_success" === JBool(true))
+ assert(found \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_a"),
JString("t"))))))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
+ "parse_sql_ctx_a")
+ } finally {
+ sql("USE default")
+ }
+ } finally {
+ sql("DROP TABLE IF EXISTS parse_sql_ctx_a.t")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_a CASCADE")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_b CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE named mode argument") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | mode => 'ANALYZE')""".stripMargin).head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+ }
+ }
+
+ test("ANALYZE without parameter marker types leaves the query unresolved") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql("SELECT parse_sql('SELECT abs(?) AS x', 'ANALYZE')")
+ .head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(false))
+ }
+ }
+
+ private def evalBatch(sql: String): List[JObject] =
+ evalJson("'" + sql.replace("'", "''") + "'") match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def typedColumns(columns: (String, JObject)*): JArray =
JArray(columns.map {
+ case (name, dataType) => JObject("name" -> JArray(List(JString(name))),
"type" -> dataType)
+ }.toList)
+
+ private def ref(parts: String*): JArray = JArray(parts.map(JString).toList)
+
+ /** Run a full `SELECT parse_sql(...)` and return its statement objects. */
+ private def evalCall(query: String): List[JObject] =
+ parse(spark.sql(query).head().getString(0)) match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def errorClass(statement: JObject): JValue = statement \ "error" \
"errorClass"
+
+ test("ANALYZE does not run CALL, directly or through EXECUTE IMMEDIATE") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ spark.conf.set("spark.sql.catalog.parse_sql_cat",
classOf[InMemoryCatalog].getName)
+ try {
+ val catalog =
spark.sessionState.catalogManager.catalog("parse_sql_cat")
+ .asInstanceOf[InMemoryCatalog]
+ catalog.createProcedure(
+ Identifier.of(Array("ns"), "record"),
ParseSqlAnalyzeSuite.UnboundRecord)
+ ParseSqlAnalyzeSuite.calls.set(0)
+ val results = evalBatch(
+ """CALL parse_sql_cat.ns.record(1);
+ |EXECUTE IMMEDIATE 'CALL
parse_sql_cat.ns.record(2)'""".stripMargin)
+ assert(results.map(_ \ "parse_success") === List(JBool(true),
JBool(true)))
+ assert(results.map(_ \ "analyze_success") === List(JBool(true),
JBool(true)))
+ assert(ParseSqlAnalyzeSuite.calls.get === 0)
+ sql("CALL parse_sql_cat.ns.record(3)").collect()
+ assert(ParseSqlAnalyzeSuite.calls.get === 1)
+ } finally {
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf("spark.sql.catalog.parse_sql_cat")
+ }
+ }
+ }
+
+ test("ANALYZE inside a SQL script does not see the script's locals") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """BEGIN
+ | DECLARE parse_sql_outer INT DEFAULT 5;
+ | SELECT parse_sql('SELECT parse_sql_outer AS x', 'ANALYZE');
+ |END""".stripMargin).head().getString(0)
+ val statement = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(statement \ "analyze_success" === JBool(false))
+ val cls = (statement \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(cls === "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION" ||
+ cls === "UNRESOLVED_COLUMN.WITH_SUGGESTION" ||
+ cls.startsWith("UNRESOLVED_COLUMN"))
+ }
+ }
+
+ test("ANALYZE requires constant arguments; PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val query = "SELECT parse_sql(c, '%s') FROM VALUES ('SELECT 1') AS t(c)"
+ val e =
intercept[AnalysisException](spark.sql(query.format("ANALYZE")).collect())
+ assert(e.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ assert(spark.sql(query.format("PARSE")).collect().length === 1)
+ val modeCol = intercept[AnalysisException](
+ spark.sql("SELECT parse_sql(c, m) FROM VALUES ('SELECT 1', 'PARSE') AS
t(c, m)")
+ .collect())
+ assert(modeCol.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ }
+ }
+
+ test("ANALYZE does not execute SQL scripts") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_script_t", "parse_sql_script_u") {
+ sql("CREATE TABLE parse_sql_script_t (a INT) USING parquet")
+ sql("CREATE TABLE parse_sql_script_u (a INT) USING parquet")
+ val schema = spark.sql("SELECT current_schema()").head().getString(0)
+ val j = evalStatement(
+ """BEGIN
+ | INSERT INTO parse_sql_script_t VALUES (1);
+ | DROP TABLE parse_sql_script_u;
+ |END""".stripMargin)
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(errorClass(j) ===
JString("PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT"))
+ assert(spark.table("parse_sql_script_t").count() === 0)
+ assert(spark.catalog.tableExists("parse_sql_script_u"))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
schema)
+ }
+ }
+ }
+
+ test("ANALYZE reports range as system.builtin and de-duplicates count") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_dedup_t") {
+ sql("CREATE TABLE parse_sql_dedup_t (a INT, b INT) USING parquet")
+ val rangeStmt = evalStatement("SELECT * FROM range(3)")
+ assert(rangeStmt \ "analyze_success" === JBool(true))
+ assert(rangeStmt \ "function_references" === JArray(List(
+ ref("system", "builtin", "range"))))
+ val counts = evalStatement(
+ "SELECT count(a), COUNT(b) FROM parse_sql_dedup_t")
+ assert(counts \ "analyze_success" === JBool(true))
+ assert(counts \ "function_references" === JArray(List(
+ ref("system", "builtin", "count"))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies DDL targets by statement kind") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val tempView = evalStatement("CREATE TEMPORARY VIEW parse_sql_ddl_v AS
SELECT 1 AS a")
+ assert(tempView \ "analyze_success" === JBool(true))
+ assert(tempView \ "target_table_references" === JArray(List(
+ ref("system", "session", "parse_sql_ddl_v"))))
+ assert(!spark.catalog.tableExists("parse_sql_ddl_v"))
+ withTempView("parse_sql_ddl_t") {
+ sql("CREATE TEMP VIEW parse_sql_ddl_t AS SELECT 1 AS a")
+ val table = evalStatement("CREATE TABLE parse_sql_ddl_t (a INT) USING
parquet")
+ assert(table \ "analyze_success" === JBool(true))
+ assert(table \ "target_table_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_ddl_t"))))
+ }
+ }
+ }
+
+ test("ANALYZE types session-catalog CTAS query columns") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ 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") {
+ 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") {
+ val explain = evalStatement("EXPLAIN SELECT 1 AS a")
Review Comment:
**Medium: This test can't catch the bug it was added for.** `SELECT 1 AS a`
is already fully typed after parsing, so this passes with or without the new
`SupervisingCommand` branch, and the case from the earlier thread, `EXPLAIN
SELECT nope FROM missing_t` reporting `analyze_success: true`, still has no
test. Adding that case (expecting `analyze_success` false with
`TABLE_OR_VIEW_NOT_FOUND`) and `EXPLAIN SELECT a FROM <table>` (a typed column
and a qualified source) would pin the fix.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,43 +18,71 @@
package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry,
FunctionRegistryBase, TypeCheckResult}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{
+ ExpressionBuilder, FunctionRegistry, FunctionRegistryBase, TypeCheckResult
+}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.catalyst.parser.ParseSqlResult
+import org.apache.spark.sql.catalyst.parser.{ParseSqlMode, ParseSqlResult}
+import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature,
InputParameter}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase,
QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.StringTypeWithCollation
import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
import org.apache.spark.unsafe.types.UTF8String
/**
* Parses a SQL batch string and returns a compact JSON array describing its
- * unresolved statements (source position, identifier/code, lineage references,
- * select-list names, parameters). A statement that does not parse is
represented
- * by a STANDARD-format error object at its position in the array.
+ * statements (source position, identifier/code, lineage references,
+ * select-list names, parameters). `PARSE` mode is unresolved. `ANALYZE`
+ * mode also types the select list and fully qualifies resolved relations
+ * and routines. A statement that does not parse is represented by a
+ * STANDARD-format error object at its position in the array.
*
* Behind [[SQLConf.PARSE_SQL_ENABLED]] while the JSON contract is still
* evolving. Designed for batch evaluation over DataFrames of SQL text.
- * User-facing parse errors become JSON; unexpected internal failures
propagate.
+ * User-facing parse / analyze errors become JSON; unexpected internal
+ * failures propagate. `ANALYZE` needs an active Spark session (catalog)
+ * and inherits the invoker's current catalog, schema, and SQL path.
*/
// scalastyle:off line.size.limit
// scalastyle:off nonascii
@ExpressionDescription(
- usage = """_FUNC_(sqlStmt) - Splits `sqlStmt` into SQL statements, parses
each with
- the stock Spark SQL parser, and returns a JSON array describing them
(1-based
- Unicode code-point `start` and code-point `length`, parse success, Table 39
- statement identifier/code, target and source table references for lineage,
- select-list column names, and parameter markers).
- Statement length excludes surrounding whitespace and the terminating
semicolon.
- Session parser extensions are not applied.
- Requires spark.sql.function.parseSql.enabled=true. On syntax / parse error
returns JSON
- for that statement with `parse_success` false, source location, and a
nested STANDARD
- error object instead of throwing or stopping the remaining statements.
Nested error
- locations are statement-relative Unicode code points. An empty or
closed-comment-only
- batch returns `[]`.""",
+ usage = """_FUNC_(sqlStmt[, mode]) -
Review Comment:
**Low: The `usage` text still breaks the docs generator, and `mode` is
under-documented.** As in the earlier docs thread, this first line ends in `)
-` followed by a newline, so the `(.*parse_sql.*) - ` split in
`sql/gen-sql-functions-docs.py` finds no match and the script exits with -1;
starting the description on this line fixes it. Separately, the `mode` argument
is documented as "an expression that evaluates to a string", but it must be a
constant in both modes.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,71 +18,172 @@
package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry,
FunctionRegistryBase, TypeCheckResult}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{
+ ExpressionBuilder, FunctionRegistry, FunctionRegistryBase, TypeCheckResult
+}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.catalyst.parser.ParseSqlResult
+import org.apache.spark.sql.catalyst.parser.{
+ ParseSqlAnalyzeContext, ParseSqlMode, ParseSqlResult
+}
+import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature,
InputParameter}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase,
QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.StringTypeWithCollation
import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
import org.apache.spark.unsafe.types.UTF8String
/**
* Parses a SQL batch string and returns a compact JSON array describing its
- * unresolved statements (source position, identifier/code, lineage references,
- * select-list names, parameters). A statement that does not parse is
represented
- * by a STANDARD-format error object at its position in the array.
+ * statements (source position, identifier/code, lineage references,
+ * select-list names, parameters). `PARSE` mode is unresolved. `ANALYZE`
+ * mode also types the select list and fully qualifies resolved relations
+ * and routines. A statement that does not parse is represented by a
+ * STANDARD-format error object at its position in the array.
*
* Behind [[SQLConf.PARSE_SQL_ENABLED]] while the JSON contract is still
* evolving. Designed for batch evaluation over DataFrames of SQL text.
- * User-facing parse errors become JSON; unexpected internal failures
propagate.
+ * User-facing parse / analyze errors become JSON; unexpected internal
+ * failures propagate. `ANALYZE` needs an active Spark session (catalog)
+ * and inherits the invoker's current catalog, schema, and SQL path.
+ * Optional named `current_catalog`, `current_schema`, and `current_path`
+ * override that context for the duration of analysis.
*/
// scalastyle:off line.size.limit
// scalastyle:off nonascii
@ExpressionDescription(
- usage = """_FUNC_(sqlStmt) - Splits `sqlStmt` into SQL statements, parses
each with
- the stock Spark SQL parser, and returns a JSON array describing them
(1-based
- UTF-16 code-unit `start` and UTF-16 code-unit `length`, parse success,
Table 39
- statement identifier/code, target and source table references for lineage,
- select-list column names, and parameter markers).
- Statement length excludes surrounding whitespace and the terminating
semicolon.
- Session parser extensions are not applied.
- Requires spark.sql.function.parseSql.enabled=true. On syntax / parse error
returns JSON
- for that statement with `parse_success` false, source location, and a
nested STANDARD
- error object instead of throwing or stopping the remaining statements.
Nested error
- locations are statement-relative. An empty or closed-comment-only batch
returns `[]`.""",
+ usage = """_FUNC_(sqlStmt[, mode[, current_catalog[, current_schema[,
current_path]]]]) -
+ Splits `sqlStmt` into SQL statements, parses each with the stock Spark SQL
parser, and
+ returns a JSON array describing them so callers can inspect SQL text
(lineage, statement
+ kind, select-list types) without running the statements. Each object has
1-based
+ character `start` and character `length` (Unicode code points; a
single-code-point
+ character, including a BMP glyph or one supplementary emoji, counts as
one),
+ parse success, Table 39 statement identifier/code, target and source
+ table references, select-list column names, and parameter markers. Every
parameter
+ may be passed by name (`sqlStmt => ...`, `mode => ...`).
+ `mode` is `PARSE` (default) or `ANALYZE` (case-insensitive). `PARSE` is
unresolved and
+ does not need a catalog. `ANALYZE` needs an active session; it types
selected columns,
+ fully qualifies resolved relations and routines, and reports analysis
failures as JSON
+ without aborting the batch. Each statement is analyzed against the session
as it is,
+ without being executed, so a DECLARE, CREATE, USE, or SET does not affect
later
+ statements. `CALL` is analyzed but the procedure is not run. A SQL script
calling
+ `parse_sql` does not lend it its locals or cursors. `ANALYZE` inherits the
invoker's
+ current catalog, schema, and SQL path. Named `current_catalog`,
`current_schema`, and
+ `current_path` override that context for analysis only. Omitted or NULL
keeps the
+ invoker value except that `current_catalog` alone uses the new catalog's
default
+ namespace. The invoker catalog, namespace, path, and PATH flag are
restored after
+ the call. An override changes the session for the duration of the call, so
other
+ queries running concurrently on the same session can observe it;
`current_path`
+ also enables the SQL path for the call.
+ `PARSE` ignores the catalog arguments. Statement length excludes
surrounding
+ whitespace and the terminating semicolon. Session parser extensions are
not applied.
+ The JSON contract may still change while
spark.sql.function.parseSql.enabled remains
+ off by default. On syntax / parse error returns JSON for that statement
with
+ `parse_success` false, source location, and a nested STANDARD error object
instead of
+ throwing or stopping later statements. Nested `queryContext` is the
STANDARD
+ renderer output (view and EXECUTE IMMEDIATE bodies keep their own text).
+ Unexpected internal failures still fail the outer query. An
+ empty or closed-comment-only batch returns `[]`. NULL `sqlStmt` or `mode`
returns
+ NULL.""",
arguments = """
Arguments:
* sqlStmt - A SQL batch string to split and parse.
- An expression that evaluates to a string.
+ An expression that evaluates to a string. NULL returns NULL.
+ * mode - Optional `PARSE` or `ANALYZE` (case-insensitive). Defaults to
`PARSE`.
+ An expression that evaluates to a string. NULL returns NULL. Any other
value
+ fails the outer query with INVALID_PARAMETER_VALUE.PARSE_SQL_MODE.
+ * current_catalog - Optional catalog name used as current catalog during
`ANALYZE`.
+ NULL or omitted inherits the invoker session. Ignored in `PARSE`. The
invoker
+ catalog, schema, and path are restored after analysis.
+ * current_schema - Optional schema / namespace used as current schema
during `ANALYZE`.
+ NULL or omitted inherits the invoker session. Ignored in `PARSE`.
+ * current_path - Optional SQL PATH used during `ANALYZE` (SET PATH
grammar, or the
+ string returned by `current_path()`). NULL or omitted inherits the
invoker session.
+ Ignored in `PARSE`.
""",
examples = """
Examples:
> SELECT _FUNC_('SELECT 1;SELECT 2');
[{"start":1,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]},{"start":10,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}]
> SELECT get_json_object(_FUNC_('SELECT ''😀'';SELECT 2'), '$[1].start');
- 13
+ 12
> SELECT get_json_object(_FUNC_('SELEC'), '$[0].error.errorClass');
PARSE_SYNTAX_ERROR
+ > SELECT get_json_object(_FUNC_('SELECT 1 AS a', 'ANALYZE'),
'$[0].select_list[0].type.name');
+ int
+ > SELECT get_json_object(_FUNC_('SELECT a FROM missing_t', 'ANALYZE'),
'$[0].analyze_success');
+ false
+ """,
+ note = """
+ Requires spark.sql.function.parseSql.enabled=true. The result type is
STRING JSON.
+ ANALYZE needs an active session, and every argument must be constant for
the query
+ (DATATYPE_MISMATCH.NON_FOLDABLE_INPUT otherwise); a DataFrame column of
statements
+ can use PARSE only. An invalid `current_catalog`, `current_schema`, or
`current_path`
+ is reported as JSON (`analyze_success` false,
INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT)
+ and does not fail the outer query. Each JSON object has `start` (1-based
character
+ offset in the batch), `length`, and `parse_success`. Parse success also has
+ `statement_identifier` / `statement_code` (ISO/IEC 9075-2:2023 Table 39).
ANALYZE
+ adds `analyze_success`. Optional fields are omitted when empty:
+ `target_table_references`, `source_table_references`,
`function_references` (arrays
+ of name parts; ANALYZE qualifies persistent names to
catalog.schema.object, local
+ temp views and temp functions as `system.session`, builtins as
`system.builtin`;
+ CTE aliases in scope are not listed as tables), `select_list` (`{name}`
parts;
+ ANALYZE lists the output columns, with `*` expanded, and adds `type` as a
+ DESCRIBE TABLE EXTENDED AS JSON type object), `parameter_markers` (`named`
and/or
+ `unnamed_count`), and `error` (STANDARD format). Later statements are still
+ processed after a per-statement parse or analysis failure. ANALYZE of a SQL
+ script reports `analyze_success` false without executing the script.
""",
group = "misc_funcs",
since = "4.4.0")
// scalastyle:on nonascii
// scalastyle:on line.size.limit
-case class ParseSql(child: Expression)
- extends UnaryExpression
+case class ParseSql(
+ sqlStmt: Expression,
+ mode: Expression,
+ currentCatalog: Expression,
+ currentSchema: Expression,
+ currentPath: Expression)
+ extends Expression
with ImplicitCastInputTypes
- with CodegenFallback {
+ with CodegenFallback
+ with QueryErrorsBase {
+
+ def this(
+ sqlStmt: Expression,
+ mode: Expression,
+ currentCatalog: Expression,
+ currentSchema: Expression) =
+ this(sqlStmt, mode, currentCatalog, currentSchema, Literal.create(null,
StringType))
+
+ def this(sqlStmt: Expression, mode: Expression, currentCatalog: Expression) =
+ this(sqlStmt, mode, currentCatalog, Literal.create(null, StringType))
+
+ def this(sqlStmt: Expression, mode: Expression) =
+ this(sqlStmt, mode, Literal.create(null, StringType))
+
+ def this(sqlStmt: Expression) = this(sqlStmt, Literal("PARSE"))
override def prettyName: String = "parse_sql"
- override def nullable: Boolean = true
+ override def children: Seq[Expression] =
+ Seq(sqlStmt, mode, currentCatalog, currentSchema, currentPath)
+
+ override def nullable: Boolean = sqlStmt.nullable || mode.nullable
+
+ // Foldable so constant SQL is evaluated on the driver, where ANALYZE can
see the session.
+ override def foldable: Boolean = children.forall(_.foldable)
Review Comment:
Thanks. The concern is the dependence on an optimizer rule rather than the
arguments: ANALYZE reaches the driver only because `ConstantFolding` folds it,
and `ConstantFolding` is not in `nonExcludableRules`. With
`spark.sql.optimizer.excludedRules` set to
`org.apache.spark.sql.catalyst.optimizer.ConstantFolding`, `SELECT
parse_sql('SELECT 1', 'ANALYZE')` is evaluated in a task and fails with
`PARSE_SQL_ANALYZE_REQUIRES_SESSION`. Without any conf, the same happens when
the call is a CASE or IF branch over table rows and folding throws (for example
on a misconfigured catalog), because `ConstantFolding` leaves a failed
conditional branch for runtime. Evaluating ANALYZE once in a rule listed in
`SparkOptimizer.nonExcludableRules` would remove that dependence.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +660,183 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(
+ analyzed: LogicalPlan,
+ parsed: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(analyzed)
+ if (query.resolved && query.output.nonEmpty) {
+ query.output.map { attr =>
+ JObject(
+ "name" -> partsToJArray(Seq(attr.name)),
+ "type" -> DescribeRelationJsonCommand.jsonType(
+
CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType)))
+ }
+ } else {
+ collectSelectList(parsed)
+ }
+ }
+
+ private def uniqueIdentities(
+ ids: Seq[Seq[String]],
+ session: ClassicSparkSession): Seq[Seq[String]] = {
+ val sensitive = session.sessionState.conf.caseSensitiveAnalysis
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ ids.filter { parts =>
+ val key = if (sensitive) parts else parts.map(_.toLowerCase(Locale.ROOT))
+ seen.add(key)
+ }
+ }
+
+ private def sessionIdent(name: String): Seq[String] =
+ Seq(CatalogManager.SYSTEM_CATALOG_NAME, CatalogManager.SESSION_NAMESPACE,
name)
+
+ private def catalogIdentParts(catalog: String, ident: Identifier):
Seq[String] =
+ Seq(catalog) ++ ident.namespace() :+ ident.name()
+
+ private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] =
+ (r.catalog, r.identifier) match {
+ case (Some(catalog), Some(ident)) =>
Some(catalogIdentParts(catalog.name(), ident))
+ case _ => None
+ }
+
+ private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] =
+ r.catalogTable.map { t =>
+ val id = t.identifier
+ if (id.catalog.isEmpty && id.database.isDefined) {
+ CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id)
+ } else {
+ tableIdentifierParts(id)
+ }
+ }.orElse {
+ r.relation match {
+ case fs: HadoopFsRelation =>
+ fs.location.rootPaths.headOption.map(p => Seq(p.toString))
+ case _ => None
+ }
+ }
+
+ 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] = {
Review Comment:
**Medium: Several temp-view identities come out wrong.** `DROP VIEW
session.v` and `DROP VIEW system.session.v` reach `tempViewParts` with the
namespace as written and become `spark_catalog.session.v` and the 4-part
`spark_catalog.system.session.v`. The `CreateTempViewUsing` arm (L595) ignores
`global`, so `CREATE GLOBAL TEMPORARY VIEW ... USING` reports `system.session`.
When analysis fails, `qualifyTarget` reports a global temp create as
`system.session.<v>`, and sources that are temp views become `<current
catalog>.<current schema>.<v>`. Mapping session-qualified namespaces to
`system.session`, honoring `global`, and checking temp views first on the
failure path would fix these.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,70 +122,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, spans, context)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false).
+ * After a context override failure, references stay unqualified parse names
+ * rather than the invoker or the attempted context.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ successJson(plan, None, Some(e), Some(session), span, qualify
= false)
+ }.toList)
+ }
+ }
+ }
+ }
+
+ /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the
call. */
+ private def isInternalError(e: SparkThrowable): Boolean =
+ Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR"))
+
+ /** Character span of a statement object in the original batch. */
+ private case class StatementSpan(start: Int, length: Int)
+
+ /**
+ * Code-point `start` for each statement in one forward pass over `batch`.
+ * Statements are in source order, so prefix counts are not repeated.
+ */
+ private def statementSpans(
+ batch: String,
+ statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = {
+ var lastUtf16 = 0
+ var lastCp = 0
+ statements.map { segment =>
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ if (utf16 >= lastUtf16) {
+ lastCp += batch.codePointCount(lastUtf16, utf16)
+ lastUtf16 = utf16
+ } else {
+ lastCp = batch.codePointCount(0, utf16)
+ lastUtf16 = utf16
+ }
+ StatementSpan(lastCp + 1, charLength(segment))
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): JObject = {
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ segment: PositionedSqlStatement,
+ span: StatementSpan): ParsedStatement = {
val sql = segment.statement
try {
// Do not inherit the outer query's origin from the parse_sql expression.
// Errors and parsed nodes refer to the individual statement.
val origin = if (sql.nonEmpty) {
- Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText
= Some(sql))
+ Origin(
+ startIndex = Some(0),
+ stopIndex = Some(sql.codePointCount(0, sql.length) - 1),
+ sqlText = Some(sql))
} else {
Origin(sqlText = Some(sql))
}
CurrentOrigin.withOrigin(origin) {
- val plan = parser.get().parsePlan(sql)
- successJson(plan, segment)
+ Parsed(segment, parser.get().parsePlan(sql))
}
} catch {
- // User-facing parse / scripting failures become JSON; everything else
fails.
case e: ParseException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, span))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, span))
+ }
+ }
+
+ /** Analyze `plan` without executing it. Scripts never reach QueryExecution.
*/
Review Comment:
To clarify, this one is about execution rather than naming.
`EvalSubqueriesForTimeTravel` runs the subquery in `TIMESTAMP AS OF (SELECT
...)` as a Spark job during analysis, before the table is looked up, so ANALYZE
reads data (and runs any UDFs in that subquery) although the docs say
statements are not executed, and the subquery's value can come back in the
error JSON. I'd either refuse these in the analysis-only scope next to
`withoutInvocation`, or document what analysis still runs (this, file schema
inference and Python UDTF `analyze`).
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala:
##########
@@ -253,50 +253,4 @@ class ParseSqlResultSuite extends SparkFunSuite {
assert(j \ "parse_success" === JBool(false))
assert(j \ "error" \ "errorClass" === JString("PARSE_SYNTAX_ERROR"))
}
-
- test("error queryContext indexes are Unicode code points") {
Review Comment:
**Medium: This drops the parse_sql test that SPARK-60672 added.** Leaving
nested error fragments to the STANDARD renderer makes sense now that
SPARK-60672 has landed, but this also deletes that change's parse_sql test
(code-point `startIndex`, `stopIndex` and `position` for syntax, typed-literal,
later-statement and script errors) and the doc sentence that nested error
locations are statement-relative Unicode code points, and the description
doesn't mention either. If the renderer covers this, the test should pass
unchanged at this head, and as far as I can tell it is the only parse_sql-level
coverage of these indexes, so I'd keep it and the doc sentence.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -452,11 +582,52 @@ object ParseSqlResult {
case u: UnresolvedView => add(u.multipartIdentifier)
case u: UnresolvedTableOrView => add(u.multipartIdentifier)
case u: UnresolvedIdentifier if role == TableRefRole.Target =>
add(u.nameParts)
- case c: CreateViewCommand => addTarget(tableIdentifierParts(c.name))
- case c: CreateTempViewUsing =>
addTarget(tableIdentifierParts(c.tableIdent))
+ case c: CreateViewCommand =>
+ addTarget(c.viewType match {
+ case LocalTempView => sessionIdent(c.name.table)
+ case GlobalTempView =>
+ Seq(
+ CatalogManager.SESSION_CATALOG_NAME,
+ SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE),
+ c.name.table)
+ case _ => tableIdentifierParts(c.name)
+ })
+ case c: CreateTempViewUsing =>
addTarget(sessionIdent(c.tableIdent.table))
+ case c: CreateDataSourceTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: DropTableCommand =>
+ addTarget(
+ if (c.isView && c.tableName.database.isEmpty) {
+ sessionIdent(c.tableName.table)
+ } else {
+ persistedTableParts(c.tableName)
+ })
+ case c: DropTempViewCommand =>
+ addTarget(tempViewParts(c.ident))
case c: CacheTable if c.multipartIdentifier.nonEmpty =>
addTarget(c.multipartIdentifier)
case u: UnresolvedTableValuedFunction => addFunction(u.name)
+ case _: CTERelationRef =>
+ case r: DataSourceV2Relation =>
+ v2RelationParts(r).foreach(add)
+ case r: LogicalRelation =>
+ logicalRelationParts(r).foreach(add)
+ case h: HiveTableRelation =>
+ add(tableIdentifierParts(h.tableMeta.identifier))
+ case v: View =>
+ add(viewParts(v))
+ case r: ResolvedTable =>
+ add(catalogIdentParts(r.catalog.name(), r.identifier))
+ case r: ResolvedPersistentView =>
+ add(catalogIdentParts(r.catalog.name(), r.identifier))
+ case r: ResolvedTempView =>
+ add(tempViewParts(r.identifier))
+ case r: ResolvedIdentifier if role == TableRefRole.Target =>
+ add(catalogIdentParts(r.catalog.name(), r.identifier))
+ case _: Range =>
Review Comment:
**Low: Some code is now dead or redundant.** This `Range` arm can't affect
the output: parsed plans don't contain `Range`, and function references from
the analyzed walk are discarded (L286-288). Likewise, the `CTERelationRef` arm
is a no-op, `ParseSql.parameterNames` has no callers, the runtime
`invalidParseSqlMode` in `ParseSql.eval` is unreachable because `mode` is
validated during analysis (and it quotes the value differently from the
analysis-time error), and the `CreateDataSourceTableAsSelectCommand` /
`InsertIntoDataSourceCommand` arms in `primaryQueryPlan` duplicate its
`LeafRunnableCommand` arm. Also, `eval` returns NULL whenever either input is
NULL, so `nullIntolerant` could be `true`.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,70 +122,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, spans, context)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false).
+ * After a context override failure, references stay unqualified parse names
+ * rather than the invoker or the attempted context.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
Review Comment:
With the batch-level catch gone, this now fails the whole outer query rather
than one statement: identity qualification runs outside `analyze()`'s try, and
`isCatalogRegistered` in `qualifyWithCurrentNamespace` loads the plugin and
only catches `CatalogNotFoundException`. After `SET
spark.sql.catalog.badcat=com.example.Missing`, `parse_sql('SELECT 1 AS a;
SELECT * FROM badcat.ns.t', 'ANALYZE')` fails with `CANNOT_LOAD_CATALOG`
instead of returning JSON for both statements. The same happens for a script,
which is never analyzed (`BEGIN SELECT * FROM badcat.ns.t; END`), and for a
function name like `badcat.ns.f(1)`. A registration check that never loads a
plugin (the session catalog name, catalogs already loaded, or whether
`spark.sql.catalog.<name>` is set), or a per-statement catch that keeps the
parsed name, would keep statements independent.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.procedures.{
+ BoundProcedure,
+ ProcedureParameter,
+ UnboundProcedure
+}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataTypes, StructType}
+
+class ParseSqlAnalyzeSuite extends QueryTest with SharedSparkSession {
+
+ private def evalJson(sql: String, mode: String = "ANALYZE"): JValue = {
+ val result = spark.sql(s"SELECT parse_sql($sql,
'$mode')").head().getString(0)
+ parse(result)
+ }
+
+ private def evalStatement(sql: String, mode: String = "ANALYZE"): JObject = {
+ evalJson("'" + sql.replace("'", "''") + "'", mode) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ }
+
+ private def jsonType(name: String): JObject = JObject("name" ->
JString(name))
+ private val jsonInt: JObject = jsonType("int")
+ private val jsonBigint: JObject = jsonType("bigint")
+ private val jsonBoolean: JObject = jsonType("boolean")
+ private val jsonString: JObject = JObject(
+ "name" -> JString("string"),
+ "collation" -> JString("UTF8_BINARY"))
+
+ test("ANALYZE types selected columns and PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val analyzed = evalStatement("SELECT 1 AS a", "ANALYZE")
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+
+ val parsed = evalStatement("SELECT 1 AS a", "PARSE")
+ assert(parsed \ "analyze_success" === JNothing)
+ assert(parsed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a")))))))
+ }
+ }
+
+ test("ANALYZE fully qualifies tables and builtin routines") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_analyze_t") {
+ sql("CREATE TABLE parse_sql_analyze_t (a INT, b STRING) USING parquet")
+ val j = evalStatement(
+ "SELECT count(a) AS c FROM parse_sql_analyze_t", "analyze")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("default"),
JString("parse_sql_analyze_t"))))))
+ assert(j \ "function_references" === JArray(List(JArray(List(
+ JString("system"), JString("builtin"), JString("count"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("c"))), "type" ->
jsonBigint))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies local temp views as system.session") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTempView("parse_sql_analyze_v") {
+ sql("CREATE OR REPLACE TEMP VIEW parse_sql_analyze_v AS SELECT 1 AS a,
'x' AS b")
+ val j = evalStatement("SELECT a, b FROM parse_sql_analyze_v")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("system"), JString("session"),
JString("parse_sql_analyze_v"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt),
+ JObject("name" -> JArray(List(JString("b"))), "type" ->
jsonString))))
+ }
+ }
+ }
+
+ test("ANALYZE keeps parse metadata when analysis fails") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val j = evalStatement("SELECT a FROM missing_parse_sql_analyze_t")
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(j \ "statement_identifier" === JString("SELECT"))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"),
+ JString("default"),
+ JString("missing_parse_sql_analyze_t"))))))
+ val errorClass = (j \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(errorClass == "TABLE_OR_VIEW_NOT_FOUND" ||
+ errorClass.contains("TABLE_OR_VIEW_NOT_FOUND"))
+ }
+ }
+
+ test("ANALYZE inherits the invoker schema") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE parse_sql_ctx_a")
+ sql("CREATE DATABASE parse_sql_ctx_b")
+ try {
+ sql("CREATE TABLE parse_sql_ctx_a.t (a INT) USING parquet")
+ sql("USE parse_sql_ctx_b")
+ try {
+ val missing = evalStatement("SELECT a FROM t")
+ assert(missing \ "analyze_success" === JBool(false))
+ assert(missing \ "source_table_references" ===
JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_b"),
JString("t"))))))
+ } finally {
+ sql("USE parse_sql_ctx_a")
+ }
+ try {
+ val found = evalStatement("SELECT a FROM t")
+ assert(found \ "analyze_success" === JBool(true))
+ assert(found \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_a"),
JString("t"))))))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
+ "parse_sql_ctx_a")
+ } finally {
+ sql("USE default")
+ }
+ } finally {
+ sql("DROP TABLE IF EXISTS parse_sql_ctx_a.t")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_a CASCADE")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_b CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE named mode argument") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | mode => 'ANALYZE')""".stripMargin).head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+ }
+ }
+
+ test("ANALYZE without parameter marker types leaves the query unresolved") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql("SELECT parse_sql('SELECT abs(?) AS x', 'ANALYZE')")
+ .head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(false))
+ }
+ }
+
+ private def evalBatch(sql: String): List[JObject] =
+ evalJson("'" + sql.replace("'", "''") + "'") match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def typedColumns(columns: (String, JObject)*): JArray =
JArray(columns.map {
+ case (name, dataType) => JObject("name" -> JArray(List(JString(name))),
"type" -> dataType)
+ }.toList)
+
+ private def ref(parts: String*): JArray = JArray(parts.map(JString).toList)
+
+ /** Run a full `SELECT parse_sql(...)` and return its statement objects. */
+ private def evalCall(query: String): List[JObject] =
+ parse(spark.sql(query).head().getString(0)) match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def errorClass(statement: JObject): JValue = statement \ "error" \
"errorClass"
+
+ test("ANALYZE does not run CALL, directly or through EXECUTE IMMEDIATE") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ spark.conf.set("spark.sql.catalog.parse_sql_cat",
classOf[InMemoryCatalog].getName)
+ try {
+ val catalog =
spark.sessionState.catalogManager.catalog("parse_sql_cat")
+ .asInstanceOf[InMemoryCatalog]
+ catalog.createProcedure(
+ Identifier.of(Array("ns"), "record"),
ParseSqlAnalyzeSuite.UnboundRecord)
+ ParseSqlAnalyzeSuite.calls.set(0)
+ val results = evalBatch(
+ """CALL parse_sql_cat.ns.record(1);
+ |EXECUTE IMMEDIATE 'CALL
parse_sql_cat.ns.record(2)'""".stripMargin)
+ assert(results.map(_ \ "parse_success") === List(JBool(true),
JBool(true)))
+ assert(results.map(_ \ "analyze_success") === List(JBool(true),
JBool(true)))
+ assert(ParseSqlAnalyzeSuite.calls.get === 0)
+ sql("CALL parse_sql_cat.ns.record(3)").collect()
+ assert(ParseSqlAnalyzeSuite.calls.get === 1)
+ } finally {
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf("spark.sql.catalog.parse_sql_cat")
+ }
+ }
+ }
+
+ test("ANALYZE inside a SQL script does not see the script's locals") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """BEGIN
+ | DECLARE parse_sql_outer INT DEFAULT 5;
+ | SELECT parse_sql('SELECT parse_sql_outer AS x', 'ANALYZE');
+ |END""".stripMargin).head().getString(0)
+ val statement = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(statement \ "analyze_success" === JBool(false))
+ val cls = (statement \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(cls === "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION" ||
+ cls === "UNRESOLVED_COLUMN.WITH_SUGGESTION" ||
+ cls.startsWith("UNRESOLVED_COLUMN"))
+ }
+ }
+
+ test("ANALYZE requires constant arguments; PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val query = "SELECT parse_sql(c, '%s') FROM VALUES ('SELECT 1') AS t(c)"
+ val e =
intercept[AnalysisException](spark.sql(query.format("ANALYZE")).collect())
+ assert(e.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ assert(spark.sql(query.format("PARSE")).collect().length === 1)
+ val modeCol = intercept[AnalysisException](
+ spark.sql("SELECT parse_sql(c, m) FROM VALUES ('SELECT 1', 'PARSE') AS
t(c, m)")
+ .collect())
+ assert(modeCol.getCondition === "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ }
+ }
+
+ test("ANALYZE does not execute SQL scripts") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_script_t", "parse_sql_script_u") {
+ sql("CREATE TABLE parse_sql_script_t (a INT) USING parquet")
+ sql("CREATE TABLE parse_sql_script_u (a INT) USING parquet")
+ val schema = spark.sql("SELECT current_schema()").head().getString(0)
+ val j = evalStatement(
+ """BEGIN
+ | INSERT INTO parse_sql_script_t VALUES (1);
+ | DROP TABLE parse_sql_script_u;
+ |END""".stripMargin)
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(errorClass(j) ===
JString("PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT"))
+ assert(spark.table("parse_sql_script_t").count() === 0)
+ assert(spark.catalog.tableExists("parse_sql_script_u"))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
schema)
+ }
+ }
+ }
+
+ test("ANALYZE reports range as system.builtin and de-duplicates count") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_dedup_t") {
+ sql("CREATE TABLE parse_sql_dedup_t (a INT, b INT) USING parquet")
+ val rangeStmt = evalStatement("SELECT * FROM range(3)")
+ assert(rangeStmt \ "analyze_success" === JBool(true))
+ assert(rangeStmt \ "function_references" === JArray(List(
+ ref("system", "builtin", "range"))))
+ val counts = evalStatement(
+ "SELECT count(a), COUNT(b) FROM parse_sql_dedup_t")
+ assert(counts \ "analyze_success" === JBool(true))
+ assert(counts \ "function_references" === JArray(List(
+ ref("system", "builtin", "count"))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies DDL targets by statement kind") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val tempView = evalStatement("CREATE TEMPORARY VIEW parse_sql_ddl_v AS
SELECT 1 AS a")
+ assert(tempView \ "analyze_success" === JBool(true))
+ assert(tempView \ "target_table_references" === JArray(List(
+ ref("system", "session", "parse_sql_ddl_v"))))
+ assert(!spark.catalog.tableExists("parse_sql_ddl_v"))
+ withTempView("parse_sql_ddl_t") {
+ sql("CREATE TEMP VIEW parse_sql_ddl_t AS SELECT 1 AS a")
+ val table = evalStatement("CREATE TABLE parse_sql_ddl_t (a INT) USING
parquet")
+ assert(table \ "analyze_success" === JBool(true))
+ assert(table \ "target_table_references" === JArray(List(
+ ref("spark_catalog", "default", "parse_sql_ddl_t"))))
+ }
+ }
+ }
+
+ test("ANALYZE types session-catalog CTAS query columns") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ 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") {
+ 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;
Review Comment:
**Low: A few gaps in the batch tests.** Nothing checks per-statement
isolation in ANALYZE (an earlier `USE`, `SET`, `DECLARE` or `CREATE TEMP VIEW`
that later statements must not see, or a failing statement followed by a
passing one), there's still no `EXECUTE IMMEDIATE` DDL or DML payload case, and
`PARSE_SQL_ANALYZE_REQUIRES_SESSION` is never exercised. Also,
`parse_sql_batch_u` here and `parse_sql_ctas_t` above aren't wrapped in
`withTable`, so a regression that runs them would leak tables into later tests.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -6922,6 +6927,18 @@
],
"sqlState" : "42601"
},
+ "PARSE_SQL_ANALYZE_REQUIRES_SESSION" : {
+ "message" : [
+ "parse_sql ANALYZE mode requires an active Spark session with catalog
access."
Review Comment:
**Low: The two new error messages could say what to do.**
`PARSE_SQL_ANALYZE_REQUIRES_SESSION` only fires when the call wasn't folded on
the driver (for example with `ConstantFolding` excluded), where the caller does
have a session, so the message points at the wrong cause.
`PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT` gives no remedy, such as using PARSE for
`BEGIN ... END`. Both also hard-code `parse_sql` instead of using
`<functionName>` like the `INVALID_PARAMETER_VALUE` subclass.
--
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]
