srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4408726181
##########
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:
Fixed in 2c8af4ccad9. statementSpans walks the batch once in source order
and accumulates codePointCount between consecutive starts, so cost is O(L)
rather than O(S x L).
##########
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:
Fixed in 2c8af4ccad9. Usage/note now say a single code point counts as one
character, current_catalog alone uses that catalog default namespace,
queryContext is the STANDARD renderer (view/EXECUTE IMMEDIATE bodies keep their
own text), and ANALYZE qualifies persistent names to catalog.schema.object.
##########
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:
The parse_sql-local queryContext rewrite is already gone (598d1344e1e).
Nested queryContext is the STANDARD renderer output, so a SQL table function
body keeps its own fragment and indexes.
##########
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:
Fixed in 2c8af4ccad9. The object doc now says ANALYZE also reports
non-internal analysis and context errors as JSON.
--
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]
