spark-test commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4405430806
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -93,22 +190,109 @@ case class ParseSql(child: Expression)
"configKey" -> SQLConf.PARSE_SQL_ENABLED.key,
"configValue" -> "true"))
}
- super.checkInputDataTypes()
+ val typeCheck = super.checkInputDataTypes()
+ if (!typeCheck.isSuccess) {
+ typeCheck
+ } else {
+ val parsedMode = if (mode.foldable) {
Review Comment:
**Non-blocking (P2):** The documented rule that ANALYZE rejects non-constant
arguments with `DATATYPE_MISMATCH.NON_FOLDABLE_INPUT` is enforced only when
`mode` itself is foldable. With a column-valued mode, `parsedMode` is `None`
and the check succeeds. Also, `constant()` accepts any deterministic non-leaf
with constant children, which admits non-foldable expressions such as a
deterministic ScalaUDF over a literal. Such calls are not folded, so `SELECT
parse_sql(s, m) FROM t` with a row `('SELECT 1', 'ANALYZE')` runs in a task.
There `getActiveSession` is `None`, so it fails with
`PARSE_SQL_ANALYZE_REQUIRES_SESSION`, which says an active session is missing
although the caller has one. The same rows from `VALUES` succeed, because
`ConvertToLocalRelation` evaluates them on the driver. Suggested fix: reject a
non-foldable `mode` at analysis, and accept ANALYZE arguments only if they
become foldable once the `current_*()` functions are replaced by literals.
**Verification:**
- **Regression:** A mode read from a column, a deterministic UDF over a
literal, and a scalar subquery over a table each fail analysis with
NON_FOLDABLE_INPUT instead of failing at run time.
- **Behavior:** PARSE over a column of statements with a literal or omitted
mode, and ANALYZE with literal and current_*() arguments, still return results.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -407,12 +721,109 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(plan: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(plan)
+ 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(plan)
+ }
+ }
+
+ private def qualifyRelation(
+ parts: Seq[String],
+ session: ClassicSparkSession,
+ resolve: Boolean = true): Seq[String] = {
+ val catalog = session.sessionState.catalog
+ val sessionIdent = Seq(CatalogManager.SYSTEM_CATALOG_NAME,
+ CatalogManager.SESSION_NAMESPACE, parts.last)
+ parts match {
+ case Seq(name) if catalog.getRawTempView(name).isDefined =>
+ sessionIdent
+ case _ if CatalogManager.isSessionQualifiedViewName(parts) &&
catalog.isTempView(parts) =>
+ sessionIdent
+ case _ if catalog.isTempView(parts) =>
+ parts match {
+ case Seq(db, name) => Seq(db, name)
+ case other => other
+ }
+ case _ =>
+ Option.when(resolve)(resolvedRelationIdentity(parts, session)).flatten
+ .getOrElse(qualifyWithCurrentNamespace(parts, session))
+ }
+ }
+
+ /**
+ * Identity of the relation `parts` resolves to, so a name found through the
+ * path is reported where it was found. None when it does not resolve.
+ */
+ private def resolvedRelationIdentity(
+ parts: Seq[String],
+ session: ClassicSparkSession): Option[Seq[String]] = {
+ try {
+ session.sessionState.analyzer.execute(UnresolvedRelation(parts)) match {
+ case s: SubqueryAlias if s.identifier.qualifier.nonEmpty =>
+ Some(s.identifier.qualifier :+ s.identifier.name)
+ case _ => None
+ }
+ } catch {
+ case NonFatal(_) => None
+ }
+ }
+
+ private def qualifyRoutine(
Review Comment:
**Blocking (P1):** `qualifyRoutine` rebuilds identities from the parse-time
name instead of using what analysis bound. `SELECT * FROM range(3)` reports
`["spark_catalog","default","range"]`. `range` is registered only in
`TableFunctionRegistry` (FunctionRegistry.scala:1490-1491), not in
`FunctionRegistry.builtin`, so it falls through to the current-namespace branch
even though analysis binds the builtin table function. The check order also
disagrees with resolution. The default
`spark.sql.functionResolution.sessionOrder` is `second`, builtin before
session, so after `CREATE TEMPORARY FUNCTION upper(...)` a call to `upper('a')`
binds the builtin but is reported as `system.session.upper`. The SQL path is
never consulted, so a function found through a `current_path` entry is reported
in the current schema. On the relation side, `qualifyRelation` returns
`system.session` for any same-named temp view before `RelationResolution`'s
path-ordered steps. Lineage consumers get confidently wr
ong dependencies under `analyze_success: true`.
See **Shared repair plan 2** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ if (conf.getConf(SQLConf.PATH_ENABLED) != prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, prevPathEnabled)
+ }
+ }
+ }
+ }
+
+ private def applyPathOverride(
+ mgr: CatalogManager,
+ conf: SQLConf,
+ path: String): Unit = {
+ if (path.trim.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ val elements = try {
+ parser.get().parsePathElements(path)
+ } catch {
+ case _: ParseException =>
+ throw QueryExecutionErrors.invalidParseSqlContext("current_path",
path)
+ }
+ val expanded = PathElement.validateNoStaticDuplicates(
+ PathElement.expand(elements, conf, mgr),
+ conf.caseSensitiveAnalysis)
+ if (expanded.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ mgr.setSessionPath(expanded)
+ }
}
}
/** Build success JSON from an already-parsed unresolved plan. */
- private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement):
JObject = {
+ private def successJson(
+ plan: LogicalPlan,
+ 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(segment.start + 1)
- fields += "length" -> JInt(segment.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) {
+ // Qualify from the ANALYZE session even when analysis failed: names and
+ // builtins are known from parse plus current catalog / schema / path.
+ // A DML target is the existing object its name resolves to; a DDL target
+ // is named in the current schema.
+ val dmlTarget = plan match {
+ case _: InsertIntoStatement | _: UnresolvedInsert | _: UpdateTable | _:
DeleteFromTable |
+ _: MergeIntoTable => true
+ case _ => false
+ }
+ val targetTables = session.map(s =>
refs.targetTables.map(qualifyRelation(_, s, dmlTarget)))
Review Comment:
**Blocking (P1):** Non-DML targets are qualified with `qualifyRelation(_, s,
resolve = false)`, which decides temp versus persistent only from whether a
temp view with that name already exists. The parser hands `CreateViewCommand`
and `CreateTempViewUsing` targets over as bare names, and
`CommandExecutionMode.SKIP` never creates the view. So `parse_sql('CREATE
TEMPORARY VIEW v AS SELECT 1 AS a', 'ANALYZE')` reports
`[["spark_catalog","default","v"]]`, a persistent object the statement never
writes. The same statement is reported as `system.session.v` once `v` exists.
Conversely, `CREATE TABLE t (a INT)` while a temp view `t` exists is reported
as `system.session.t`, although it creates a persistent table. The reported
write target of common DDL depends on unrelated session state instead of the
statement's object kind.
See **Shared repair plan 2** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -407,12 +721,109 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(plan: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(plan)
+ 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(plan)
Review Comment:
**Blocking (P1):** The fallback on this line re-reads the analyzed plan
instead of the parsed one, and `primaryQueryPlan` has no case for analyzed
commands that hold their query as a field. On the session catalog, `CREATE
TABLE t2 USING parquet AS SELECT 1 AS a` is analyzed into
`CreateDataSourceTableAsSelectCommand`, a `LeafRunnableCommand` with an empty
output. The statement is then reported with `analyze_success: true` and no
`select_list`, while PARSE reports `[{"name":["a"]}]` and a v2 CTAS gets typed
columns. In the other direction, the new `V2WriteCommand` case unwraps the
analyzer's row-level rewrites (`ReplaceData` and `WriteDelta`) for DELETE,
UPDATE and MERGE. Their internal rewrite columns, such as `__row_operation`,
become the public select list where PARSE reports none. Suggested fix: type the
statement's own query. Unwrap commands that hold it as a field (data-source and
Hive CTAS, data-source inserts), and report an INSERT's query columns rather
than the target-ali
gned projection. Leave row-level rewrites unwrapped, and fall back to the
parsed plan's select list.
**Verification:**
- **Regression:** ANALYZE of a default-catalog CTAS and of a Hive CTAS
reports the query's column names with types, and an INSERT with a column list
reports the query's columns in select order.
- **Regression:** ANALYZE of DELETE, UPDATE and MERGE on a v2 table with
row-level operations reports no select list and no internal columns.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
Review Comment:
**Non-blocking (P2):** `setCurrentNamespace` re-validates the namespace
through `assertNamespaceExist` or `setCurrentDatabaseWithNameCheck`. But
`prevNamespace` is the derived `currentNamespace`, which falls back to the
catalog's `defaultNamespace()`. After `USE CATALOG jdbc_cat` that is the empty
array, and `JDBCTableCatalog.namespaceExists(Array())` is false. So
`parse_sql('SELECT 1', 'ANALYZE', current_path => 'system.builtin')` analyzes
fine and then throws `SCHEMA_NOT_FOUND` here. The path restore and the
`spark.sql.path.enabled` restore after it never run, so the override path stays
on the session. `analyzeBatch`'s catch then replaces the correct result with
that error. The same happens after `USE ns; DROP NAMESPACE ns`, which
`DropNamespaceSuiteBase` shows is allowed. A `current_schema => 'prod'` call
then leaves the session in `prod`, so a later unqualified `DROP TABLE x`
targets `prod.x`.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
Review Comment:
**Non-blocking (P2):** The restore writes back the snapshot taken before the
override without checking whether anything else changed the session meanwhile,
and it reads and restores the four fields one at a time. Every operation of the
session shares this `CatalogManager` and `SQLConf`. Concurrent Spark Connect
operations, or threads on one `SparkSession`, can interleave as follows. T1
runs `parse_sql(sql, 'ANALYZE', current_schema => 'staging')` and snapshots
`prod`. T2 runs `USE dev` to completion. T1's restore then puts the session
back into `prod`, so T2's next unqualified INSERT or DROP resolves in `prod`.
Two overlapping overriding calls nest their snapshots, so the session can end
up on the first call's override. The usage text accepts that concurrent queries
may observe the override during the call, but not that committed changes are
silently lost afterwards.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ if (conf.getConf(SQLConf.PATH_ENABLED) != prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, prevPathEnabled)
Review Comment:
**Non-blocking (P2):** `prevPathEnabled` is the effective value, so when the
key was never set this line writes an explicit `spark.sql.path.enabled=false`
into the session. `Analyzer.retainResolutionConfigsForAnalysis` overlays
explicitly set SESSION-bound configs onto view and SQL-function bodies. So
after a single `current_path` call, a view or SQL function saved with
`spark.sql.path.enabled=true` stops using its pinned path, and views created
afterwards capture `false`. The same snapshot-and-write-back pins the catalog:
`setCurrentCatalog(prevCatalog)` publishes `_currentCatalogName = Some(...)`.
After one `current_catalog` override, a later `SET spark.sql.defaultCatalog =
x` therefore no longer changes the session's current catalog. The documented
restore should leave unset state unset.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,343 @@
+/*
+ * 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 invoker schema and can override current_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"))))))
+
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT a FROM t',
+ | 'ANALYZE',
+ | current_schema =>
'parse_sql_ctx_a')""".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 \ "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_b")
+ } 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 and current_catalog inherit from invoker") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | mode => 'ANALYZE',
+ | current_catalog =>
current_catalog())""".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 current_path override is restored on the invoker") {
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ SQLConf.PATH_ENABLED.key -> "true") {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ try {
+ val before = spark.sql("SELECT current_path()").head().getString(0)
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | 'ANALYZE',
+ | current_path => 'system.builtin,
spark_catalog.default')""".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(spark.sql("SELECT current_path()").head().getString(0) ===
before)
+ } finally {
+ sql("SET PATH = DEFAULT_PATH")
+ }
+ }
+ }
+
+ test("ANALYZE reports invalid current_schema as JSON instead of failing the
query") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ "SELECT parse_sql('SELECT 1 AS a', 'ANALYZE', current_schema =>
'!!!')")
+ .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))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a")))))))
+ assert((analyzed \ "error" \ "errorClass").asInstanceOf[JString].s ===
+ "INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT")
+ }
+ }
+
+ 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"
+
+ private def queryContext(statement: JObject): JObject =
+ statement \ "error" \ "queryContext" match {
+ case JArray(List(ctx: JObject)) => ctx
+ case other => fail(s"expected one query context, got: $other")
+ }
+
+ private val emoji = new String(Character.toChars(0x1F600))
+
+ 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)))
Review Comment:
**Nit (P3):** This test asserts `parse_success` and the call count but not
`analyze_success`. If the suppressed CALL failed analysis before
`InvokeProcedures` ran, `calls` would stay 0 and the test would still pass.
Other tests have similar gaps. The `current_path` test analyzes `SELECT 1` with
`spark.sql.path.enabled` already set to true, so it can neither see the
override take effect nor catch the flag being left on. The `current_catalog`
test overrides with `current_catalog()`, a no-op. The script-locals test only
checks `analyze_success: false`, which any unrelated failure also satisfies.
Nothing covers script text, DDL or DML targets, CTAS or INSERT, builtin table
functions, temp-function shadowing, a real catalog switch, a NULL or
non-constant mode, multi-statement batches, CHAR/VARCHAR or `*` columns, or
concurrent overrides. Suggested fix: make each of these assertions sensitive to
the behavior it names, and add the missing cases.
**Verification:**
- **Inspection:** Each documented ANALYZE behavior is covered by an
assertion that a deliberate regression of it would fail. This includes CALL
analyzed but not run, the current_path override taking effect and the SQL path
flag restored, a real catalog switch restored, NULL and invalid modes,
multi-statement isolation, typed CHAR/VARCHAR and expanded columns, and
commands whose effects stay unapplied.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -43,18 +92,24 @@ import
org.apache.spark.sql.execution.datasources.CreateTempViewUsing
* executors without a session, so only the stock parser is available under
* distributed eval.
*
- * Every statement object includes its 1-based UTF-16 code-unit `start` in the
- * original batch and its UTF-16 code-unit `length`, excluding surrounding
- * whitespace and the terminating semicolon. On success it also includes
+ * Every statement object includes its 1-based character `start` in the
+ * original batch and its character `length`, excluding surrounding
+ * whitespace and the terminating semicolon. Characters are Unicode code
+ * points (Spark SQL `length()` / UTF-8 strings): a supplementary-plane
+ * emoji is one character. Java `String` offsets are UTF-16 (emoji = 2);
+ * ANTLR `CodePointCharStream` offsets are already code points (emoji = 1).
+ * The splitter maps ANTLR indexes to UTF-16 for `substring`; JSON converts
+ * those spans back to code points. On success it also includes
* `parse_success`, the statement
* identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional
* fields (`target_table_references`, `source_table_references`,
* `function_references`, `select_list`, `parameter_markers`) when empty. On
* parse failure the statement object contains `parse_success: false` with
* source location and a nested STANDARD-format error object, and parsing
- * continues with later statements. Nested error locations are relative to the
- * individual statement, while `start` is relative to the original batch. An
- * empty or closed-comment-only batch produces an empty array. Only
+ * continues with later statements. Nested `queryContext` indexes are 1-based
+ * Unicode code points relative to the individual statement, while `start` is
+ * relative to the original batch. An empty or closed-comment-only batch
+ * produces an empty array. Only
Review Comment:
**Nit (P3):** This sentence predates ANALYZE. In ANALYZE mode, `analyze()`
and `analyzeBatch`'s catch also convert every non-`INTERNAL_ERROR`
`SparkThrowable` into JSON, for example `TABLE_OR_VIEW_NOT_FOUND` or
`INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT`. Maintainers reading the object doc
get the wrong error-handling contract. Suggested fix: say that ANALYZE also
reports non-internal analysis and context errors as JSON.
See **Shared repair plan 3** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,71 +18,168 @@
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; an emoji is
one
+ character), 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, which is 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 error locations are
statement-relative
+ Unicode code points. 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, local temp views and
temp
Review Comment:
**Nit (P3):** Several statements in the usage and note promise more than the
code does. (1) "an emoji is one character" (line 60) holds only for
single-code-point emoji. Positions count code points, so a flag or skin-tone
sequence counts as two or more. (2) "omitted or NULL keeps the invoker value"
(lines 72-73) does not hold for `current_schema` when `current_catalog`
switches catalogs, because `setCurrentCatalog` resets the namespace to the new
catalog's default. (3) "Nested error locations are statement-relative" (lines
81-82 and 130-131) excludes view and EXECUTE IMMEDIATE contexts, which
`rewriteErrorCharIndexes` deliberately leaves as is. (4) On this line, "ANALYZE
qualifies persistent names, local temp views and temp functions as
`system.session`" reads as if persistent names become `system.session`; they
get their catalog and schema. Suggested fix: reword each statement to match the
implemented behavior.
See **Shared repair plan 3** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ if (conf.getConf(SQLConf.PATH_ENABLED) != prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, prevPathEnabled)
+ }
+ }
+ }
+ }
+
+ private def applyPathOverride(
+ mgr: CatalogManager,
+ conf: SQLConf,
+ path: String): Unit = {
+ if (path.trim.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ val elements = try {
+ parser.get().parsePathElements(path)
+ } catch {
+ case _: ParseException =>
+ throw QueryExecutionErrors.invalidParseSqlContext("current_path",
path)
+ }
+ val expanded = PathElement.validateNoStaticDuplicates(
+ PathElement.expand(elements, conf, mgr),
+ conf.caseSensitiveAnalysis)
+ if (expanded.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ mgr.setSessionPath(expanded)
+ }
}
}
/** Build success JSON from an already-parsed unresolved plan. */
- private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement):
JObject = {
+ private def successJson(
+ plan: LogicalPlan,
+ 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(segment.start + 1)
- fields += "length" -> JInt(segment.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) {
+ // Qualify from the ANALYZE session even when analysis failed: names and
+ // builtins are known from parse plus current catalog / schema / path.
+ // A DML target is the existing object its name resolves to; a DDL target
+ // is named in the current schema.
+ val dmlTarget = plan match {
+ case _: InsertIntoStatement | _: UnresolvedInsert | _: UpdateTable | _:
DeleteFromTable |
+ _: MergeIntoTable => true
+ case _ => false
+ }
+ val targetTables = session.map(s =>
refs.targetTables.map(qualifyRelation(_, s, dmlTarget)))
+ .getOrElse(refs.targetTables)
+ val sourceTables = session.map(s =>
refs.sourceTables.map(qualifyRelation(_, s)))
+ .getOrElse(refs.sourceTables)
+ val functions = session.map(s => refs.functions.map(qualifyRoutine(_, s)))
+ .getOrElse(refs.functions)
+ if (targetTables.nonEmpty) {
fields += "target_table_references" ->
- JArray(refs.targetTables.map(partsToJArray).toList)
+ JArray(targetTables.map(partsToJArray).toList)
}
- if (refs.sourceTables.nonEmpty) {
+ if (sourceTables.nonEmpty) {
fields += "source_table_references" ->
- JArray(refs.sourceTables.map(partsToJArray).toList)
+ JArray(sourceTables.map(partsToJArray).toList)
}
- if (refs.functions.nonEmpty) {
- fields += "function_references" ->
JArray(refs.functions.map(partsToJArray).toList)
+ if (functions.nonEmpty) {
+ fields += "function_references" ->
JArray(functions.map(partsToJArray).toList)
}
- val selectList = collectSelectList(plan)
+ val selectList = analyzed.map(collectTypedSelectList)
+ .getOrElse(collectSelectList(plan))
if (selectList.nonEmpty) {
fields += "select_list" -> JArray(selectList.toList)
}
refs.parameterMarkers.foreach(markers => fields += "parameter_markers" ->
markers)
+ analyzeError.foreach { e =>
+ fields += "error" -> errorObject(e, span.text)
+ }
JObject(fields.toList)
}
- private def errorJson(
- e: SparkThrowable with Throwable,
- segment: PositionedSqlStatement): JObject = {
+ /** 1-based Unicode code-point offset of `segment` in `batch`. */
+ private def charStart(batch: String, segment: PositionedSqlStatement): Int =
{
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ batch.codePointCount(0, utf16) + 1
+ }
+
+ /** Unicode code-point length of the trimmed statement text. */
+ private def charLength(segment: PositionedSqlStatement): Int = {
+ val text = segment.statement
+ if (text.isEmpty) 0 else text.codePointCount(0, text.length)
+ }
+
+ /** STANDARD JSON of `e` with `queryContext` in code points of `statement`.
*/
+ private def errorObject(e: SparkThrowable with Throwable, statement:
String): JObject = {
+ val errorObj = parseJson(
+ SparkThrowableHelper.getMessage(e,
ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
+ rewriteErrorCharIndexes(errorObj, statement)
+ }
+
+ /**
+ * Recompute the fragment of each statement `queryContext` by code points.
+ * The indexes are parser (code-point) offsets, but [[SQLQueryContext]]
+ * slices its fragment by UTF-16 units, which is off after a supplementary
+ * character. A context that falls outside the statement covers all of it.
+ * Contexts of another object (a view body, an EXECUTE IMMEDIATE string) are
+ * left as is.
+ */
+ private def rewriteErrorCharIndexes(errorObj: JObject, statement: String):
JObject = {
+ JObject(errorObj.obj.map {
+ case ("queryContext", JArray(contexts)) =>
+ "queryContext" -> JArray(contexts.map {
+ case ctx: JObject => rewriteQueryContextIndexes(ctx, statement)
+ case other => other
+ }.toList)
+ case other => other
+ })
+ }
+
+ private def rewriteQueryContextIndexes(ctx: JObject, statement: String):
JObject = {
+ val objectType = ctx.obj.collectFirst { case ("objectType", JString(t)) =>
t }
+ val start = ctx.obj.collectFirst { case ("startIndex", JInt(n)) => n.toInt
}
+ val stop = ctx.obj.collectFirst { case ("stopIndex", JInt(n)) => n.toInt }
+ (objectType, start, stop) match {
+ case (Some(""), Some(from), Some(to)) =>
Review Comment:
**Nit (P3):** An empty `objectType` does not mean the context belongs to
this statement. A SQL table function body is parsed from its own text, and its
nodes inherit the call site's empty `objectType`. `ResolveSQLTableFunctions`
then analyzes the body without re-positioning errors to the call site. Suppose
the table that a SQL table function `f` reads is dropped: `parse_sql('SELECT *
FROM f()', 'ANALYZE')` reports `TABLE_OR_VIEW_NOT_FOUND` with body-relative
indexes. This branch then slices the statement text at those offsets, producing
a wrong fragment, or widens the context to the whole statement when the offsets
exceed its length. PARSE is unaffected, because parse errors never come from a
function body. Suggested fix: rewrite a context only when its source text is
the analyzed statement, for example by comparing the exception's
`SQLQueryContext` text before rewriting.
**Verification:**
- **Regression:** ANALYZE of a statement calling a SQL table function whose
body reads a dropped table reports the body's fragment and indexes, not a slice
of the statement text or the whole statement.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
Review Comment:
**Blocking (P1):** This call is what makes ANALYZE run scripts. For BEGIN
... END text the parser returns a `CompoundBody`, and `QueryExecution.analyzed`
calls `assertSqlScriptExecuted()` first. That runs
`SqlScriptingExecution.executeSqlScript` and collects every statement before
any analysis happens. `CommandExecutionMode.SKIP` only governs eager command
execution, and the CALL suppression covers CALL only. So `SELECT
parse_sql('BEGIN INSERT INTO t VALUES (1); DROP TABLE u; END', 'ANALYZE')`
inserts the row and drops `u` while the optimizer folds the call, and an
EXPLAIN of that query does the same. The statement is then reported with
`analyze_success: true`, and a USE, SET or DECLARE in the script stays in
effect on the caller's session. Tools that pass SQL text to ANALYZE precisely
so that it is not run lose data without any error. PARSE mode is unaffected.
**Recommended change:** Check each parsed statement before analysis and do
not build a QueryExecution for a top-level CompoundBody (BEGIN ... END). Report
that statement with analyze_success false and an explicit unsupported-statement
error, as EXECUTE IMMEDIATE rejects scripts, and keep analyzing every other
statement through executePlan with CommandExecutionMode.SKIP as now.
**Why this works:** A statement-kind check in the analysis entry keeps every
script away from QueryExecution, so its script-execution step is never reached.
Other statements keep the existing CommandExecutionMode.SKIP analysis, which
runs no commands and begins no transaction, and the existing CALL suppression
and hidden scripting context keep covering CALL and outer-script locals.
**Scope:** Keep SQL scripts out of ANALYZE's analysis entry and report them
as not analyzable.
**Compatibility:** CALL statements stay analyzed without invoking the
procedure, and SQL scripts that call parse_sql still do not lend it their local
variables.
**Risks:** Until the follow-up that analyzes nested script statements lands,
a script in the batch is reported as not analyzable, so tools get no typed
output or qualified references for it.
**Constraints:** The script check must run before any QueryExecution or
script execution is created for a script statement.
**Exclusions:**
- Analyzing the nested statements of a SQL script is left to the planned
follow-up for script handling; this repair only guarantees that scripts are
never executed.
**Success:** ANALYZE of BEGIN ... END text leaves table data, catalog
objects and session state unchanged and reports that statement with
analyze_success false and an unsupported-statement error. Non-script statements
report the same analysis results as before the change.
**Verification:**
- **Regression:** Analyzing a script that inserts into a table, drops
another table and runs USE leaves the inserted table's contents, the dropped
table and the current schema unchanged, and reports the script statement as not
analyzable.
- **Behavior:** Queries, DML, DDL and CALL statements analyzed in the same
batch keep their typed select lists, qualified references and analyze_success
values.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ if (conf.getConf(SQLConf.PATH_ENABLED) != prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, prevPathEnabled)
+ }
+ }
+ }
+ }
+
+ private def applyPathOverride(
+ mgr: CatalogManager,
+ conf: SQLConf,
+ path: String): Unit = {
+ if (path.trim.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ val elements = try {
+ parser.get().parsePathElements(path)
+ } catch {
+ case _: ParseException =>
+ throw QueryExecutionErrors.invalidParseSqlContext("current_path",
path)
+ }
+ val expanded = PathElement.validateNoStaticDuplicates(
+ PathElement.expand(elements, conf, mgr),
+ conf.caseSensitiveAnalysis)
+ if (expanded.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ mgr.setSessionPath(expanded)
+ }
}
}
/** Build success JSON from an already-parsed unresolved plan. */
- private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement):
JObject = {
+ private def successJson(
+ plan: LogicalPlan,
+ 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(segment.start + 1)
- fields += "length" -> JInt(segment.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) {
+ // Qualify from the ANALYZE session even when analysis failed: names and
+ // builtins are known from parse plus current catalog / schema / path.
+ // A DML target is the existing object its name resolves to; a DDL target
+ // is named in the current schema.
+ val dmlTarget = plan match {
+ case _: InsertIntoStatement | _: UnresolvedInsert | _: UpdateTable | _:
DeleteFromTable |
+ _: MergeIntoTable => true
+ case _ => false
+ }
+ val targetTables = session.map(s =>
refs.targetTables.map(qualifyRelation(_, s, dmlTarget)))
+ .getOrElse(refs.targetTables)
+ val sourceTables = session.map(s =>
refs.sourceTables.map(qualifyRelation(_, s)))
Review Comment:
**Nit (P3):** De-duplication happens on the raw name parts in
`collectPlanReferences`, before this qualification step, and nothing
de-duplicates the qualified identities. `builtinFunctionIdentifier` lowercases
names, so `SELECT count(a), COUNT(b) FROM t` reports
`[["system","builtin","count"],["system","builtin","count"]]`. `SELECT * FROM t
UNION ALL SELECT * FROM default.t` lists the same table identity twice. That
contradicts the documented first-seen de-duplication, and consumers that count
or join references over-count.
See **Shared repair plan 2** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,71 +18,168 @@
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; an emoji is
one
+ character), 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, which is 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 error locations are
statement-relative
+ Unicode code points. 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)
Review Comment:
**Nit (P3):** `PARSE_SQL_CONTEXT` is raised only for a blank catalog, a
`current_schema` that fails `parseMultipartIdentifier`, and a `current_path`
that fails `parsePathElements`. A well-formed but invalid value fails later and
surfaces its own class. `current_catalog => 'no_such_cat'` gives
`CATALOG_NOT_FOUND` from `setCurrentCatalog`, `current_schema => 'missing_db'`
gives `SCHEMA_NOT_FOUND`, and a repeated static path entry gives the
`PathElement.validateNoStaticDuplicates` error. The outer query still does not
fail, but a consumer keyed on the documented class misses most invalid
contexts. Suggested fix: map every failure raised while applying an override to
`INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT`, naming the parameter and value and
keeping the original error as the cause, or document the classes that can occur.
**Verification:**
- **Regression:** Unknown catalogs, missing schemas, unparsable schemas,
unparsable paths and duplicate path entries all report
INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT naming the offending parameter,
without failing the outer query.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,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 json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.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) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ 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.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): 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; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): 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, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ if (conf.getConf(SQLConf.PATH_ENABLED) != prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, prevPathEnabled)
+ }
+ }
+ }
+ }
+
+ private def applyPathOverride(
+ mgr: CatalogManager,
+ conf: SQLConf,
+ path: String): Unit = {
+ if (path.trim.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ val elements = try {
+ parser.get().parsePathElements(path)
+ } catch {
+ case _: ParseException =>
+ throw QueryExecutionErrors.invalidParseSqlContext("current_path",
path)
+ }
+ val expanded = PathElement.validateNoStaticDuplicates(
+ PathElement.expand(elements, conf, mgr),
+ conf.caseSensitiveAnalysis)
+ if (expanded.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ mgr.setSessionPath(expanded)
+ }
}
}
/** Build success JSON from an already-parsed unresolved plan. */
- private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement):
JObject = {
+ private def successJson(
+ plan: LogicalPlan,
+ 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(segment.start + 1)
- fields += "length" -> JInt(segment.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) {
+ // Qualify from the ANALYZE session even when analysis failed: names and
+ // builtins are known from parse plus current catalog / schema / path.
+ // A DML target is the existing object its name resolves to; a DDL target
+ // is named in the current schema.
+ val dmlTarget = plan match {
+ case _: InsertIntoStatement | _: UnresolvedInsert | _: UpdateTable | _:
DeleteFromTable |
+ _: MergeIntoTable => true
+ case _ => false
+ }
+ val targetTables = session.map(s =>
refs.targetTables.map(qualifyRelation(_, s, dmlTarget)))
+ .getOrElse(refs.targetTables)
+ val sourceTables = session.map(s =>
refs.sourceTables.map(qualifyRelation(_, s)))
+ .getOrElse(refs.sourceTables)
+ val functions = session.map(s => refs.functions.map(qualifyRoutine(_, s)))
+ .getOrElse(refs.functions)
+ if (targetTables.nonEmpty) {
fields += "target_table_references" ->
- JArray(refs.targetTables.map(partsToJArray).toList)
+ JArray(targetTables.map(partsToJArray).toList)
}
- if (refs.sourceTables.nonEmpty) {
+ if (sourceTables.nonEmpty) {
fields += "source_table_references" ->
- JArray(refs.sourceTables.map(partsToJArray).toList)
+ JArray(sourceTables.map(partsToJArray).toList)
}
- if (refs.functions.nonEmpty) {
- fields += "function_references" ->
JArray(refs.functions.map(partsToJArray).toList)
+ if (functions.nonEmpty) {
+ fields += "function_references" ->
JArray(functions.map(partsToJArray).toList)
}
- val selectList = collectSelectList(plan)
+ val selectList = analyzed.map(collectTypedSelectList)
+ .getOrElse(collectSelectList(plan))
if (selectList.nonEmpty) {
fields += "select_list" -> JArray(selectList.toList)
}
refs.parameterMarkers.foreach(markers => fields += "parameter_markers" ->
markers)
+ analyzeError.foreach { e =>
+ fields += "error" -> errorObject(e, span.text)
+ }
JObject(fields.toList)
}
- private def errorJson(
- e: SparkThrowable with Throwable,
- segment: PositionedSqlStatement): JObject = {
+ /** 1-based Unicode code-point offset of `segment` in `batch`. */
+ private def charStart(batch: String, segment: PositionedSqlStatement): Int =
{
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ batch.codePointCount(0, utf16) + 1
Review Comment:
**Non-blocking (P2):** `batch.codePointCount(0, utf16)` scans the whole
prefix whenever the batch String is UTF-16-coded, which a single CJK character,
emoji or typographic quote anywhere in the batch causes. `segmentSpan` calls it
for every statement in both PARSE and ANALYZE, so a row with S statements over
L UTF-16 units costs O(S x L). The splitter that produced `segment.start`
already builds a one-time code-point map precisely so that token-boundary
lookup is O(1) (`SqlStatementSplitter.scala:161`). A large multi-statement row,
such as a migration dump read with `wholetext`, becomes a CPU straggler that
partitioning cannot split. The crossover is estimated at thousands of
statements per row; it was not measured. Pure Latin-1 batches are unaffected.
This comes from the stacked SPARK-60620 commit (74184cf9f46), so the fix can
land there. Suggested fix: compute the starts in one forward pass, either by
accumulating `codePointCount` between consecutive segment starts or by carryi
ng the ANTLR code-point start out of the splitter.
**Verification:**
- **Behavior:** Statement starts and lengths for batches with supplementary
characters in earlier and later statements match the code-point offsets
reported today.
- **Regression:** Processing time for a batch of many thousands of short
statements that contains one non-Latin-1 character grows linearly with the
number of statements.
--
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]
