Re: [PR] [SPARK-60622][SQL] Support caller-supplied context and batch carry-forward in parse_sql ANALYZE [spark]

[via GitHub]

spark-test commented on code in PR #59874:
URL: https://github.com/apache/spark/pull/59874#discussion_r4405911415


##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -407,12 +1035,116 @@ 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 resolver = session.sessionState.conf.resolver
+    val mgr = session.sessionState.catalogManager
+    val sessionIdent = SuppliedRelationRegistry.sessionIdentity(parts.last)
+    val overlaySession =
+      SuppliedRelationRegistry.lookup(sessionIdent, resolver).isDefined
+    parts match {
+      case Seq(name) if catalog.getRawTempView(name).isDefined =>
+        sessionIdent
+      case Seq(name) if overlaySession && mgr.isSystemSessionOnPath =>
+        sessionIdent
+      case _ if CatalogManager.isSessionQualifiedViewName(parts) &&
+          (catalog.isTempView(parts) || overlaySession) =>
+        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

Review Comment:
   **Blocking (P1):** `qualifyRelation` and `qualifyRoutine` re-derive each 
reported identity from the written name, `SessionCatalog` lookups and the 
current catalog and namespace, rather than using what analysis resolved, so 
several cases contradict both the documented qualification rules and the 
analyzed plan:
   
   - DDL targets use `resolve = false`, and `CREATE TEMP VIEW v AS SELECT 1 AS 
a` records its target as the one-part `TableIdentifier(v)`. With no existing 
temp view `v`, this fallthrough reports `[spark_catalog, default, v]` instead 
of `system.session.v`.
   - Any one-part name with an existing temp view is reported as 
`system.session`, regardless of statement kind or SQL path order. With a temp 
view `t`, a table `spark_catalog.default.t` and `current_path => 
'spark_catalog.default, system.builtin'`, `SELECT a FROM t` is analyzed against 
the table, but `source_table_references` reports `system.session.t`, so 
`select_list` types come from a different object than the reported name.
   - `qualifyRoutine` checks temp functions before builtins regardless of 
resolution order, recognizes builtins only through `FunctionRegistry.builtin`, 
and otherwise qualifies with the current namespace without consulting the path. 
Table-valued function names are recorded as functions, so `SELECT * FROM 
range(3)` reports `[spark_catalog, default, range]` instead of 
`system.builtin.range` (TVF-only builtins are registered in 
`TableFunctionRegistry`).
   
   Lineage consumers record these wrong identities with no error. Existing 
tests assert qualification only for a current-schema table source, a temp-view 
source, a builtin aggregate and one INSERT target. PARSE mode reports names as 
written and is unaffected.
   
   **Recommended change:** For a successfully analyzed statement, report 
relation identities from the resolved relations of the analyzed plan (resolved 
aliases and view descriptors), routine identities from resolved functions 
including table-valued functions, and a created object's identity from the 
command itself (temp views as system.session, global temp views as global_temp, 
persistent objects in their resolved catalog and schema); keep name-based 
qualification only as the fallback when analysis failed, applying the SQL path 
and the function resolution order there; add ANALYZE tests for each of these 
cases.
   
   **Why this works:** Reported names then come from the same resolution that 
typed the select list, so a temp-view shortcut, a fixed routine order or a 
missing table-valued builtin can no longer disagree with analysis, and the 
per-reference re-resolution goes away.
   
   **Scope:** Derive reported lineage from the analyzed plan, with name-based 
qualification only for failed statements.
   
   **Compatibility:** Reported names that already match analysis, such as 
current-schema tables, temp-view sources and builtin scalar functions, stay as 
they are, and the documented reference fields keep their names and shapes.
   
   **Risks:** The mapping from written names to analyzed relations must not 
drop references that analysis inlines or rewrites, such as CTE bodies and view 
expansions.
   
   **Constraints:** Failed statements keep a best-effort qualification, and the 
JSON field names and shapes stay as documented.
   
   **Success:** For successfully analyzed statements every reported relation 
and routine names the object analysis resolved, including temp views, 
path-resolved tables, table-valued builtins and temp functions that a builtin 
shadows. CREATE TEMP VIEW targets are reported as system.session and global 
temp view targets as global_temp.
   
   **Verification:**
   - **Regression:** CREATE TEMP VIEW v AS SELECT 1 reports its target as 
system.session.v.
   - **Regression:** SELECT * FROM range(3) reports range as 
system.builtin.range.
   - **Regression:** With a temp view and a table of the same name and a path 
that omits system.session, the reported source names the table analysis used.
   - **Regression:** A temporary function named like a builtin is reported as 
the builtin under the default function resolution order.
   - **Behavior:** Current qualification results for current-schema tables, 
temp-view sources and builtin scalar functions are unchanged.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -93,22 +290,180 @@ 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 `NON_FOLDABLE_INPUT` check below only fires when 
`mode` is foldable and evaluates to `ANALYZE`; for a non-foldable `mode`, 
`parsedMode` is `None` and the check succeeds. `constant()` also treats any 
deterministic non-leaf expression with constant children as constant, which 
admits an uncorrelated scalar subquery, because `SubqueryExpression.children` 
is only `outerAttrs ++ joinCond`. Neither call is foldable, so 
`ConstantFolding` skips it and `eval` decides `ANALYZE` per row:
   
   - `SELECT parse_sql(s, m) FROM VALUES ('SELECT 1 AS a', 'ANALYZE') AS v(s, 
m)` passes analysis, and `ConvertToLocalRelation` then runs ANALYZE row by row 
on the driver during optimization, including any per-row session overrides.
   - The same expression over a table scan passes analysis and fails at runtime 
with `PARSE_SQL_ANALYZE_REQUIRES_SESSION`, because the projection runs in tasks 
with no active session. `parse_sql((SELECT 'SELECT 1'), 'ANALYZE')` is accepted 
at analysis and fails at runtime the same way.
   
   So whether a non-constant ANALYZE call works depends on the optimizer's plan 
shape, and users get a runtime session error instead of the documented 
`NON_FOLDABLE_INPUT` analysis error naming the argument. `PARSE` over a column 
of statements is unaffected. No test covers a non-constant `mode` or a subquery 
argument.
   
   **Verification:**
   - **Regression:** parse_sql(s, m) over a VALUES relation and over a table, 
with m holding ANALYZE, both fail analysis with NON_FOLDABLE_INPUT.
   - **Regression:** parse_sql((SELECT 'SELECT 1'), 'ANALYZE') fails analysis 
with NON_FOLDABLE_INPUT.
   - **Behavior:** parse_sql(column, 'PARSE') and constant ANALYZE calls with 
current_catalog() and similar functions keep their current results.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +127,629 @@ 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 with optional context overrides.
+   * Statements are analyzed in order without being executed: each sees the
+   * variables, tables, and views declared or created by earlier statements
+   * (see [[AnalyzeState]]).
+   * 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) { env =>
+            val state = new AnalyzeState(env)
+            JArray(parsed.map {
+              case ParseFailed(errorObj) => errorObj
+              case Parsed(segment, plan) =>
+                state.statement(plan, 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)
+        }
+      }
+    }
+  }
+
+  /** 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))
+  }
+
+  /**
+   * Analysis state carried across the statements of one ANALYZE batch. It
+   * starts from the caller-supplied variables and relations and records the
+   * effect of each analyzed statement on later ones, without executing
+   * anything:
+   *  - DECLARE VARIABLE adds a session variable. DROP TEMPORARY VARIABLE
+   *    removes it.
+   *  - CREATE [OR REPLACE] TABLE / VIEW / TEMP VIEW (column list, AS query,
+   *    LIKE, or a temp view USING with a schema) adds a schema-only relation 
of
+   *    that identity. DROP TABLE / VIEW removes a relation known this way.
+   * Objects dropped that exist in the catalog stay visible, and statements 
that
+   * change a schema in place (for example ALTER TABLE) are not tracked.
+   */
+  private final class AnalyzeState(env: AnalyzeEnv) {
+    private val session = env.session
+    private def mgr: CatalogManager = session.sessionState.catalogManager
+    private def resolver: (String, String) => Boolean = 
session.sessionState.conf.resolver
+
+    private var variables: List[(Seq[String], DataType)] = env.variables.map {
+      case (name, dataType) => SuppliedVariableRegistry.canonicalize(name) -> 
dataType
+    }.toList
+    private var relations: List[(Seq[String], StructType)] = env.relations.map 
{
+      case (name, schema) => SuppliedRelationRegistry.canonicalize(mgr, name) 
-> schema
+    }.toList
+
+    /**
+     * Analyze one statement with the known objects, render its JSON, then
+     * record its effect.
+     */
+    def statement(plan: LogicalPlan, span: StatementSpan): JObject = {
+      SuppliedVariableRegistry.create(variables).runWith {
+        SuppliedRelationRegistry.create(relations).runWith {
+          val (analyzed, error) = analyze(plan)
+          val json = successJson(plan, analyzed, error, Some(session), span)
+          record(plan, analyzed)
+          json
+        }
+      }
+    }
+
+    private def analyze(
+        plan: LogicalPlan,
+        markers: Seq[(String, DataType)] = env.parameterMarkers)
+        : (Option[LogicalPlan], Option[SparkThrowable with Throwable]) = {
+      try {
+        val toAnalyze = bindSuppliedParameterMarkers(plan, markers)
+        val analyzed =
+          session.sessionState.executePlan(toAnalyze, 
CommandExecutionMode.SKIP).analyzed

Review Comment:
   **Blocking (P1):** `executePlan(toAnalyze, 
CommandExecutionMode.SKIP).analyzed` does not keep SQL scripts analysis-only. 
`QueryExecution.analyzed` first calls `assertSqlScriptExecuted()`, and for a 
`CompoundBody` -- or a `NameParameterizedQuery` over one, which 
`bindSuppliedParameterMarkers` builds when named markers are supplied -- that 
runs `SqlScriptingExecution.executeSqlScript`, which executes and collects 
every statement of the script. `SKIP` only skips `eagerlyExecuteCommands` after 
analysis, `InvokeProcedures.withoutInvocation` only suppresses `CALL`, and 
nothing in `analyzeBatch` or `AnalyzeState` filters compound statements.
   
   With `spark.sql.scripting.enabled` at its default `true`, `SELECT 
parse_sql('BEGIN DROP TABLE t; INSERT INTO u VALUES (1); END', 'ANALYZE')` 
drops `t` and inserts into `u` while the call is constant-folded on the driver, 
then reports the statement as if it had only been analyzed. `EXPLAIN` of the 
same query runs it too, because folding happens during optimization. That 
contradicts the documented "analyzed in order without being executed" behavior: 
a lineage or linting service that runs untrusted SQL text through `parse_sql` 
`ANALYZE` executes it with the caller's privileges. Only script statements in 
ANALYZE mode are affected, and no ANALYZE test covers one.
   
   **Verification:**
   - **Regression:** ANALYZE of a batch whose statement is a SQL script that 
drops a table and inserts rows leaves the table and the target rows unchanged 
and reports the script statement as not analyzed.
   - **Regression:** The same holds when named parameter markers are supplied, 
so the script would otherwise be wrapped in a parameterized query.
   - **Behavior:** Analysis results for non-script statements, including CALL 
suppression and EXECUTE IMMEDIATE payload analysis, match the current results.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1368,11 +1369,21 @@ class Analyzer(
     private def lookupTableOrView(
         identifier: Seq[String],
         viewOnly: Boolean = false): Option[LogicalPlan] = {
+      val sessionOverlay =
+        if (identifier.length == 1 || 
CatalogManager.isSessionQualifiedViewName(identifier)) {
+          relationResolution.lookupSuppliedTempRelation(
+            identifier, 
SuppliedRelationRegistry.sessionIdentity(identifier.last))
+        } else {
+          None
+        }
+      sessionOverlay.orElse {
       relationResolution.lookupTempView(identifier).map { tempView =>
         ResolvedTempView(identifier.asIdentifier, tempView.tableMeta)
       }.orElse {
         relationResolution.expandIdentifier(identifier) match {
           case CatalogAndIdentifier(catalog, ident) =>
+            val nameParts = ident.toQualifiedNameParts(catalog)
+            relationResolution.lookupSuppliedRelation(nameParts).orElse {

Review Comment:
   **Non-blocking (P2):** `lookupTableOrView` serves `UnresolvedTable`, 
`UnresolvedView` and `UnresolvedTableOrView` -- the `ALTER TABLE` family, 
`DESCRIBE`, `ANALYZE TABLE`, `TRUNCATE`, `SHOW PARTITIONS` and `SHOW 
TBLPROPERTIES`, `COMMENT ON`, `REFRESH`, `CREATE`/`DROP INDEX` and `ALTER VIEW` 
-- whose callers expect a `ResolvedTable` or `Resolved[Temp/Persistent]View`. 
Returning the overlay's `SubqueryAlias(LocalRelation)` here, and in the 
session-overlay branch above, hands those commands a plan shape they do not 
handle.
   
   The concrete failure: `ResolveFieldNameAndPosition`, which runs right after 
`ResolveRelations`, does `a.table.asInstanceOf[ResolvedTable]` for any resolved 
`AlterTableCommand` with an `UnresolvedFieldName`, and for `CreateIndex`. In 
`SELECT parse_sql('CREATE TABLE t (a INT); ALTER TABLE t RENAME COLUMN a TO b', 
'ANALYZE')`, carry-forward installs the overlay `spark_catalog.default.t`, and 
the `ALTER` then throws a raw `ClassCastException`. It is not a 
`SparkThrowable`, and `QueryExecution` converts only `NullPointerException`, 
`AssertionError` and `MatchError` to internal errors, so it escapes 
`AnalyzeState.analyze` and `analyzeBatch` and fails the outer query, losing 
every statement's result. A supplied relation `t` with `ALTER TABLE t ALTER 
COLUMN a TYPE BIGINT` fails the same way. For the other commands, 
`analyze_success` on an overlay name is not meaningful. Overlays used as query 
sources and INSERT targets are unaffected. No test runs a DDL command against 
an overlay.
   
   **Verification:**
   - **Regression:** CREATE TABLE t (a INT); ALTER TABLE t RENAME COLUMN a TO b 
in one ANALYZE batch returns a JSON object for both statements.
   - **Regression:** ALTER TABLE t ALTER COLUMN a TYPE BIGINT with a supplied 
relation t returns per-statement JSON.
   - **Behavior:** Overlay eclipsing for query sources and INSERT targets keeps 
its current results, and a later statement still sees the table that CREATE 
TABLE ... LIKE created from an overlay source.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +127,629 @@ 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 with optional context overrides.
+   * Statements are analyzed in order without being executed: each sees the
+   * variables, tables, and views declared or created by earlier statements
+   * (see [[AnalyzeState]]).
+   * 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) { env =>
+            val state = new AnalyzeState(env)
+            JArray(parsed.map {
+              case ParseFailed(errorObj) => errorObj
+              case Parsed(segment, plan) =>
+                state.statement(plan, 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)
+        }
+      }
+    }
+  }
+
+  /** 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))
+  }
+
+  /**
+   * Analysis state carried across the statements of one ANALYZE batch. It
+   * starts from the caller-supplied variables and relations and records the
+   * effect of each analyzed statement on later ones, without executing
+   * anything:
+   *  - DECLARE VARIABLE adds a session variable. DROP TEMPORARY VARIABLE
+   *    removes it.
+   *  - CREATE [OR REPLACE] TABLE / VIEW / TEMP VIEW (column list, AS query,
+   *    LIKE, or a temp view USING with a schema) adds a schema-only relation 
of
+   *    that identity. DROP TABLE / VIEW removes a relation known this way.
+   * Objects dropped that exist in the catalog stay visible, and statements 
that
+   * change a schema in place (for example ALTER TABLE) are not tracked.
+   */
+  private final class AnalyzeState(env: AnalyzeEnv) {
+    private val session = env.session
+    private def mgr: CatalogManager = session.sessionState.catalogManager
+    private def resolver: (String, String) => Boolean = 
session.sessionState.conf.resolver
+
+    private var variables: List[(Seq[String], DataType)] = env.variables.map {
+      case (name, dataType) => SuppliedVariableRegistry.canonicalize(name) -> 
dataType
+    }.toList
+    private var relations: List[(Seq[String], StructType)] = env.relations.map 
{
+      case (name, schema) => SuppliedRelationRegistry.canonicalize(mgr, name) 
-> schema
+    }.toList
+
+    /**
+     * Analyze one statement with the known objects, render its JSON, then
+     * record its effect.
+     */
+    def statement(plan: LogicalPlan, span: StatementSpan): JObject = {
+      SuppliedVariableRegistry.create(variables).runWith {
+        SuppliedRelationRegistry.create(relations).runWith {
+          val (analyzed, error) = analyze(plan)
+          val json = successJson(plan, analyzed, error, Some(session), span)
+          record(plan, analyzed)
+          json
+        }
+      }
+    }
+
+    private def analyze(
+        plan: LogicalPlan,
+        markers: Seq[(String, DataType)] = env.parameterMarkers)
+        : (Option[LogicalPlan], Option[SparkThrowable with Throwable]) = {
+      try {
+        val toAnalyze = bindSuppliedParameterMarkers(plan, markers)
+        val analyzed =
+          session.sessionState.executePlan(toAnalyze, 
CommandExecutionMode.SKIP).analyzed
+        (Some(analyzed), None)
+      } catch {
+        case e: SparkThrowable with Throwable if !isInternalError(e) => (None, 
Some(e))
+      }
+    }
+
+    private def record(plan: LogicalPlan, analyzed: Option[LogicalPlan]): Unit 
= plan match {
+      case c: CreateVariable =>
+        variableType(c, analyzed).foreach { dataType =>
+          c.names.foreach {
+            case UnresolvedIdentifier(nameParts, _) => 
declareVariable(nameParts, dataType)
+            case _ =>
+          }
+        }
+      case DropVariable(UnresolvedIdentifier(nameParts, _), _) =>
+        val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+        variables = variables.filterNot { case (name, _) => sameName(name, 
identity) }
+      case c: CreateTable =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach {
+          createRelation(_, StructType(c.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))
+        }
+      case r: ReplaceTable =>
+        nameOf(r.name).foreach {
+          createRelation(_, StructType(r.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))
+        }
+      case c: CreateTableAsSelect =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach { name =>
+          querySchema(c.query).foreach(createRelation(name, _))
+        }
+      case r: ReplaceTableAsSelect =>
+        nameOf(r.name).foreach(name => 
querySchema(r.query).foreach(createRelation(name, _)))
+      case c: CreateTableLike =>
+        for {
+          name <- nameOf(c.name) if !(c.ifNotExists && relationExists(name))
+          source <- nameOf(c.source)
+          schema <- querySchema(UnresolvedRelation(source))
+        } createRelation(name, schema)
+      case v: CreateView =>
+        nameOf(v.child).filterNot(n => v.allowExisting && 
relationExists(n)).foreach { name =>
+          querySchema(v.query).foreach(s =>
+            createRelation(name, withColumnNames(s, v.userSpecifiedColumns)))
+        }
+      case v: CreateViewCommand =>
+        val name = viewIdentity(v.name, v.viewType)
+        if (!(v.allowExisting && relationExists(name))) {
+          querySchema(v.plan).foreach(s =>
+            createRelation(name, withColumnNames(s, v.userSpecifiedColumns)))
+        }
+      case v: CreateTempViewUsing =>
+        val viewType = if (v.global) GlobalTempView else LocalTempView
+        
v.userSpecifiedSchema.foreach(createRelation(viewIdentity(v.tableIdent, 
viewType), _))
+      case d: DropTable => nameOf(d.child).foreach(dropRelation)
+      case d: DropView => nameOf(d.child).foreach(dropRelation)
+      case _ =>
+    }
+
+    /** The declared type, else the type of the analyzed DEFAULT expression. */
+    private def variableType(
+        c: CreateVariable,
+        analyzed: Option[LogicalPlan]): Option[DataType] = c.defaultExpr.child 
match {
+      case cast: Cast => Some(cast.dataType)
+      case Literal(null, dataType) => Some(dataType)
+      case _ =>
+        analyzed.flatMap(_.collectFirst {
+          case v: CreateVariable if v.defaultExpr.child.resolved => 
v.defaultExpr.child.dataType
+        })
+    }
+
+    private def declareVariable(nameParts: Seq[String], dataType: DataType): 
Unit = {
+      val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+      variables = (identity -> dataType) ::
+        variables.filterNot { case (name, _) => sameName(name, identity) }
+    }
+
+    private def nameOf(plan: LogicalPlan): Option[Seq[String]] = plan match {
+      case UnresolvedIdentifier(nameParts, _) => Some(nameParts)
+      case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+      case u: UnresolvedTable => Some(u.multipartIdentifier)
+      case u: UnresolvedView => Some(u.multipartIdentifier)
+      case u: UnresolvedRelation => Some(u.multipartIdentifier)
+      case _ => None
+    }
+
+    private def viewIdentity(name: TableIdentifier, viewType: ViewType): 
Seq[String] =
+      viewType match {
+        case LocalTempView => 
SuppliedRelationRegistry.sessionIdentity(name.table)
+        case GlobalTempView => 
Seq(session.sessionState.catalog.globalTempDatabase, name.table)
+        case _ => tableIdentifierParts(name)
+      }
+
+    private def withColumnNames(
+        schema: StructType,
+        columns: Seq[(String, Option[String])]): StructType = {
+      if (columns.nonEmpty && columns.length == schema.length) {
+        StructType(schema.fields.zip(columns).map { case (f, (name, _)) => 
f.copy(name = name) })
+      } else {
+        schema
+      }
+    }
+
+    private def querySchema(query: LogicalPlan): Option[StructType] =
+      analyze(query)._1.map(_.schema)
+
+    /**
+     * Whether an object of the identity `nameParts` names already exists, for
+     * IF NOT EXISTS. This checks that identity only, not what the name would
+     * resolve to through the path.
+     */
+    private def relationExists(nameParts: Seq[String]): Boolean = {
+      val identity = SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+      val catalog = session.sessionState.catalog
+      relations.exists(r => sameName(r._1, identity)) || {
+        try {
+          if (CatalogManager.isSessionQualifiedViewName(identity)) {
+            catalog.getRawTempView(identity.last).isDefined
+          } else if (identity.length == 2 && 
catalog.isGlobalTempViewDB(identity.head)) {
+            catalog.getRawGlobalTempView(identity.last).isDefined
+          } else {
+            mgr.catalog(identity.head) match {
+              case tables: TableCatalog =>
+                tables.tableExists(Identifier.of(identity.tail.init.toArray, 
identity.last))
+              case _ => false
+            }
+          }
+        } catch {
+          case NonFatal(_) => false
+        }
+      }
+    }
+
+    private def createRelation(nameParts: Seq[String], schema: StructType): 
Unit = {
+      if (schema.nonEmpty) {
+        val identity = SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+        relations = (identity -> schema) ::
+          relations.filterNot { case (name, _) => sameName(name, identity) }
+      }
+    }
+
+    /** A one-part name drops the temp view of that name first, as DROP does. 
*/
+    private def dropRelation(nameParts: Seq[String]): Unit = {
+      val sessionIdentity = 
SuppliedRelationRegistry.sessionIdentity(nameParts.last)
+      val identity =
+        if (nameParts.length == 1 && relations.exists(r => sameName(r._1, 
sessionIdentity))) {
+          sessionIdentity
+        } else {
+          SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+        }
+      relations = relations.filterNot { case (name, _) => sameName(name, 
identity) }
+    }
+
+    private def sameName(left: Seq[String], right: Seq[String]): Boolean =
+      left.length == right.length && left.zip(right).forall { case (a, b) => 
resolver(a, b) }
   }
 
-  private def parseStatement(segment: PositionedSqlStatement): JObject = {
+  /** 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)))
+    }
+  }
+
+  /** Session and caller-supplied objects for one ANALYZE batch. */
+  private case class AnalyzeEnv(
+      session: ClassicSparkSession,
+      parameterMarkers: Seq[(String, DataType)],
+      variables: Seq[(Seq[String], DataType)],
+      relations: Seq[(Seq[String], StructType)])
+
+  /**
+   * Run `f` with optional catalog / schema / SQL path overrides, then restore
+   * the invoker session. `None` on a field keeps the current value. Supplied
+   * variables and relations are validated here and overlaid per statement.
+   */
+  private def withAnalyzeContext[T](
+      session: ClassicSparkSession,
+      context: ParseSqlAnalyzeContext)(f: AnalyzeEnv => T): T = {
+    val parsedVars = context.variables.map(parseKnownVariable)
+    val parsedRels = context.relations.map(parseKnownRelation)
+    val parsedMarkers = context.parameterMarkers.map(parseKnownParameterMarker)
+    val env = AnalyzeEnv(session, parsedMarkers, parsedVars, parsedRels)
+    val needsSessionOverride =
+      context.catalog.nonEmpty || context.schema.nonEmpty || 
context.path.nonEmpty
+    if (!needsSessionOverride) {
+      f(env)
+    } 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(env)
+      } 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)
+      }
+    }
+  }
+
+  private def parseKnownVariable(
+      spec: (String, String)): (Seq[String], DataType) = {
+    val (name, typeDdl) = spec
+    val parts = parseKnownName(name, "variables")
+    val dataType = try {
+      DataType.fromDDL(typeDdl)
+    } catch {
+      case _: ParseException | _: AnalysisException =>
+        throw QueryExecutionErrors.invalidParseSqlContext("variables", typeDdl)
+    }
+    (parts, dataType)
+  }
+
+  private def parseKnownRelation(
+      spec: (String, String)): (Seq[String], StructType) = {
+    val (name, schemaDdl) = spec
+    val parts = parseKnownName(name, "relations")
+    val dataType = try {
+      DataType.fromDDL(schemaDdl)
+    } catch {
+      case _: ParseException | _: AnalysisException =>
+        throw QueryExecutionErrors.invalidParseSqlContext("relations", 
schemaDdl)
+    }
+    dataType match {
+      case schema: StructType if schema.nonEmpty =>
+        (parts, CharVarcharUtils.replaceCharVarcharWithStringInSchema(schema))
+      case _ =>
+        throw QueryExecutionErrors.invalidParseSqlContext("relations", 
schemaDdl)
+    }
+  }
+
+  private def parseKnownName(name: String, parameter: String): Seq[String] = {
+    val parts = try {
+      parser.get().parseMultipartIdentifier(name)
+    } catch {
+      case _: ParseException =>
+        throw QueryExecutionErrors.invalidParseSqlContext(parameter, name)
+    }
+    if (parts.isEmpty) {
+      throw QueryExecutionErrors.invalidParseSqlContext(parameter, name)
+    }
+    parts
+  }
+
+  private def parseKnownParameterMarker(
+      spec: (String, String)): (String, DataType) = {
+    val (name, typeDdl) = spec
+    val dataType = try {
+      DataType.fromDDL(typeDdl)
+    } catch {
+      case _: ParseException | _: AnalysisException =>
+        throw QueryExecutionErrors.invalidParseSqlContext("parameter_markers", 
typeDdl)
+    }
+    val trimmed = name.trim
+    if (trimmed == "?") {
+      ("?", dataType)
+    } else {
+      val parts = parseKnownName(trimmed, "parameter_markers")
+      if (parts.length != 1) {
+        throw QueryExecutionErrors.invalidParseSqlContext("parameter_markers", 
name)
+      }
+      (parts.head, dataType)
+    }
+  }
+
+  /**
+   * Bind caller-supplied marker types using the same [[BindParameters]] rule 
as
+   * `sql(..., args)`, then analyze the bound plan.
+   */
+  private def bindSuppliedParameterMarkers(
+      plan: LogicalPlan,
+      markers: Seq[(String, DataType)]): LogicalPlan = {
+    // Markers may sit in subqueries and CTE definitions, so look deep.
+    val used = if (markers.isEmpty) None else 
collectPlanReferences(plan).parameterMarkers
+    if (used.isEmpty) {
+      plan
+    } else {
+      val namedArgs = markers.collect {
+        case (name, dataType) if name != "?" =>
+          name -> Literal.create(null, dataType)
+      }
+      val posArgs = markers.collect {
+        case ("?", dataType) => Literal.create(null, dataType)
+      }
+      val usesNamed = used.exists(_.obj.exists(_._1 == "named"))
+      val usesPos = used.exists(_.obj.exists(_._1 == "unnamed_count"))
+      val wrapped = if (usesNamed && namedArgs.nonEmpty) {
+        NameParameterizedQuery(plan, namedArgs.toMap)
+      } else if (usesPos && posArgs.nonEmpty) {
+        PosParameterizedQuery(plan, posArgs)
+      } else {
+        plan
+      }
+      BindParameters(wrapped)
     }
   }
 
   /** 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)` walks the whole 
prefix whenever the batch string holds any character above U+00FF. 
`segmentSpan` calls this once per statement -- in PARSE, on parse failures and 
in ANALYZE -- so statement i costs O(start_i), and a batch of S statements over 
L UTF-16 units costs about S*L/2 unit scans. The merge base reported 
`segment.start + 1` in O(1).
   
   For example, in `SELECT parse_sql(script) FROM scripts`, a row holding a 
migration script of about 2M characters and 20K statements with a single CJK 
identifier, typographic quote in a comment, or emoji adds about 2e10 unit scans 
for that row; near 1e5 statements it reaches minutes per row. That turns such 
rows into deterministic stragglers that AQE cannot split, while the same batch 
in pure Latin-1 stays linear. `SqlStatementSplitter` already builds a one-pass 
code-point to UTF-16 offset table so that each boundary lookup is O(1); this 
conversion rescans the prefix instead. The impact is limited to `parse_sql` 
users (the function is opt-in) with large multi-statement texts, and no 
benchmark or test measures it.
   
   **Verification:**
   - **Behavior:** Start and length of every statement in a multi-statement 
batch containing supplementary and other non-Latin-1 characters match the 
current code-point values.
   - **Inspection:** Span computation touches each UTF-16 unit of the batch a 
bounded number of times rather than rescanning the prefix per statement.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +127,629 @@ 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 with optional context overrides.
+   * Statements are analyzed in order without being executed: each sees the
+   * variables, tables, and views declared or created by earlier statements
+   * (see [[AnalyzeState]]).
+   * 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) { env =>
+            val state = new AnalyzeState(env)
+            JArray(parsed.map {
+              case ParseFailed(errorObj) => errorObj
+              case Parsed(segment, plan) =>
+                state.statement(plan, 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)
+        }
+      }
+    }
+  }
+
+  /** 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))
+  }
+
+  /**
+   * Analysis state carried across the statements of one ANALYZE batch. It
+   * starts from the caller-supplied variables and relations and records the
+   * effect of each analyzed statement on later ones, without executing
+   * anything:
+   *  - DECLARE VARIABLE adds a session variable. DROP TEMPORARY VARIABLE
+   *    removes it.
+   *  - CREATE [OR REPLACE] TABLE / VIEW / TEMP VIEW (column list, AS query,
+   *    LIKE, or a temp view USING with a schema) adds a schema-only relation 
of
+   *    that identity. DROP TABLE / VIEW removes a relation known this way.
+   * Objects dropped that exist in the catalog stay visible, and statements 
that
+   * change a schema in place (for example ALTER TABLE) are not tracked.
+   */
+  private final class AnalyzeState(env: AnalyzeEnv) {
+    private val session = env.session
+    private def mgr: CatalogManager = session.sessionState.catalogManager
+    private def resolver: (String, String) => Boolean = 
session.sessionState.conf.resolver
+
+    private var variables: List[(Seq[String], DataType)] = env.variables.map {
+      case (name, dataType) => SuppliedVariableRegistry.canonicalize(name) -> 
dataType
+    }.toList
+    private var relations: List[(Seq[String], StructType)] = env.relations.map 
{
+      case (name, schema) => SuppliedRelationRegistry.canonicalize(mgr, name) 
-> schema
+    }.toList
+
+    /**
+     * Analyze one statement with the known objects, render its JSON, then
+     * record its effect.
+     */
+    def statement(plan: LogicalPlan, span: StatementSpan): JObject = {
+      SuppliedVariableRegistry.create(variables).runWith {
+        SuppliedRelationRegistry.create(relations).runWith {
+          val (analyzed, error) = analyze(plan)
+          val json = successJson(plan, analyzed, error, Some(session), span)
+          record(plan, analyzed)
+          json
+        }
+      }
+    }
+
+    private def analyze(
+        plan: LogicalPlan,
+        markers: Seq[(String, DataType)] = env.parameterMarkers)
+        : (Option[LogicalPlan], Option[SparkThrowable with Throwable]) = {
+      try {
+        val toAnalyze = bindSuppliedParameterMarkers(plan, markers)
+        val analyzed =
+          session.sessionState.executePlan(toAnalyze, 
CommandExecutionMode.SKIP).analyzed
+        (Some(analyzed), None)
+      } catch {
+        case e: SparkThrowable with Throwable if !isInternalError(e) => (None, 
Some(e))
+      }
+    }
+
+    private def record(plan: LogicalPlan, analyzed: Option[LogicalPlan]): Unit 
= plan match {
+      case c: CreateVariable =>
+        variableType(c, analyzed).foreach { dataType =>
+          c.names.foreach {
+            case UnresolvedIdentifier(nameParts, _) => 
declareVariable(nameParts, dataType)
+            case _ =>
+          }
+        }
+      case DropVariable(UnresolvedIdentifier(nameParts, _), _) =>
+        val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+        variables = variables.filterNot { case (name, _) => sameName(name, 
identity) }
+      case c: CreateTable =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach {
+          createRelation(_, StructType(c.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))
+        }
+      case r: ReplaceTable =>
+        nameOf(r.name).foreach {
+          createRelation(_, StructType(r.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))
+        }
+      case c: CreateTableAsSelect =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach { name =>
+          querySchema(c.query).foreach(createRelation(name, _))
+        }
+      case r: ReplaceTableAsSelect =>
+        nameOf(r.name).foreach(name => 
querySchema(r.query).foreach(createRelation(name, _)))
+      case c: CreateTableLike =>
+        for {
+          name <- nameOf(c.name) if !(c.ifNotExists && relationExists(name))
+          source <- nameOf(c.source)
+          schema <- querySchema(UnresolvedRelation(source))
+        } createRelation(name, schema)
+      case v: CreateView =>
+        nameOf(v.child).filterNot(n => v.allowExisting && 
relationExists(n)).foreach { name =>
+          querySchema(v.query).foreach(s =>
+            createRelation(name, withColumnNames(s, v.userSpecifiedColumns)))
+        }
+      case v: CreateViewCommand =>
+        val name = viewIdentity(v.name, v.viewType)
+        if (!(v.allowExisting && relationExists(name))) {
+          querySchema(v.plan).foreach(s =>
+            createRelation(name, withColumnNames(s, v.userSpecifiedColumns)))
+        }
+      case v: CreateTempViewUsing =>
+        val viewType = if (v.global) GlobalTempView else LocalTempView
+        
v.userSpecifiedSchema.foreach(createRelation(viewIdentity(v.tableIdent, 
viewType), _))
+      case d: DropTable => nameOf(d.child).foreach(dropRelation)
+      case d: DropView => nameOf(d.child).foreach(dropRelation)
+      case _ =>
+    }
+
+    /** The declared type, else the type of the analyzed DEFAULT expression. */
+    private def variableType(
+        c: CreateVariable,
+        analyzed: Option[LogicalPlan]): Option[DataType] = c.defaultExpr.child 
match {
+      case cast: Cast => Some(cast.dataType)
+      case Literal(null, dataType) => Some(dataType)
+      case _ =>
+        analyzed.flatMap(_.collectFirst {
+          case v: CreateVariable if v.defaultExpr.child.resolved => 
v.defaultExpr.child.dataType
+        })
+    }
+
+    private def declareVariable(nameParts: Seq[String], dataType: DataType): 
Unit = {
+      val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+      variables = (identity -> dataType) ::
+        variables.filterNot { case (name, _) => sameName(name, identity) }
+    }
+
+    private def nameOf(plan: LogicalPlan): Option[Seq[String]] = plan match {
+      case UnresolvedIdentifier(nameParts, _) => Some(nameParts)
+      case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+      case u: UnresolvedTable => Some(u.multipartIdentifier)
+      case u: UnresolvedView => Some(u.multipartIdentifier)
+      case u: UnresolvedRelation => Some(u.multipartIdentifier)
+      case _ => None
+    }
+
+    private def viewIdentity(name: TableIdentifier, viewType: ViewType): 
Seq[String] =
+      viewType match {
+        case LocalTempView => 
SuppliedRelationRegistry.sessionIdentity(name.table)
+        case GlobalTempView => 
Seq(session.sessionState.catalog.globalTempDatabase, name.table)
+        case _ => tableIdentifierParts(name)
+      }
+
+    private def withColumnNames(
+        schema: StructType,
+        columns: Seq[(String, Option[String])]): StructType = {
+      if (columns.nonEmpty && columns.length == schema.length) {
+        StructType(schema.fields.zip(columns).map { case (f, (name, _)) => 
f.copy(name = name) })
+      } else {
+        schema
+      }
+    }
+
+    private def querySchema(query: LogicalPlan): Option[StructType] =
+      analyze(query)._1.map(_.schema)
+
+    /**
+     * Whether an object of the identity `nameParts` names already exists, for
+     * IF NOT EXISTS. This checks that identity only, not what the name would
+     * resolve to through the path.
+     */
+    private def relationExists(nameParts: Seq[String]): Boolean = {
+      val identity = SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+      val catalog = session.sessionState.catalog
+      relations.exists(r => sameName(r._1, identity)) || {
+        try {
+          if (CatalogManager.isSessionQualifiedViewName(identity)) {
+            catalog.getRawTempView(identity.last).isDefined
+          } else if (identity.length == 2 && 
catalog.isGlobalTempViewDB(identity.head)) {
+            catalog.getRawGlobalTempView(identity.last).isDefined
+          } else {
+            mgr.catalog(identity.head) match {
+              case tables: TableCatalog =>
+                tables.tableExists(Identifier.of(identity.tail.init.toArray, 
identity.last))
+              case _ => false
+            }
+          }
+        } catch {
+          case NonFatal(_) => false
+        }
+      }
+    }
+
+    private def createRelation(nameParts: Seq[String], schema: StructType): 
Unit = {
+      if (schema.nonEmpty) {
+        val identity = SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+        relations = (identity -> schema) ::
+          relations.filterNot { case (name, _) => sameName(name, identity) }
+      }
+    }
+
+    /** A one-part name drops the temp view of that name first, as DROP does. 
*/
+    private def dropRelation(nameParts: Seq[String]): Unit = {
+      val sessionIdentity = 
SuppliedRelationRegistry.sessionIdentity(nameParts.last)
+      val identity =
+        if (nameParts.length == 1 && relations.exists(r => sameName(r._1, 
sessionIdentity))) {
+          sessionIdentity
+        } else {
+          SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+        }
+      relations = relations.filterNot { case (name, _) => sameName(name, 
identity) }
+    }
+
+    private def sameName(left: Seq[String], right: Seq[String]): Boolean =
+      left.length == right.length && left.zip(right).forall { case (a, b) => 
resolver(a, b) }
   }
 
-  private def parseStatement(segment: PositionedSqlStatement): JObject = {
+  /** 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)))
+    }
+  }
+
+  /** Session and caller-supplied objects for one ANALYZE batch. */
+  private case class AnalyzeEnv(
+      session: ClassicSparkSession,
+      parameterMarkers: Seq[(String, DataType)],
+      variables: Seq[(Seq[String], DataType)],
+      relations: Seq[(Seq[String], StructType)])
+
+  /**
+   * Run `f` with optional catalog / schema / SQL path overrides, then restore
+   * the invoker session. `None` on a field keeps the current value. Supplied
+   * variables and relations are validated here and overlaid per statement.
+   */
+  private def withAnalyzeContext[T](
+      session: ClassicSparkSession,
+      context: ParseSqlAnalyzeContext)(f: AnalyzeEnv => T): T = {
+    val parsedVars = context.variables.map(parseKnownVariable)
+    val parsedRels = context.relations.map(parseKnownRelation)
+    val parsedMarkers = context.parameterMarkers.map(parseKnownParameterMarker)
+    val env = AnalyzeEnv(session, parsedMarkers, parsedVars, parsedRels)
+    val needsSessionOverride =
+      context.catalog.nonEmpty || context.schema.nonEmpty || 
context.path.nonEmpty
+    if (!needsSessionOverride) {
+      f(env)
+    } else {
+      val mgr = session.sessionState.catalogManager
+      val conf = session.sessionState.conf
+      val prevCatalog = mgr.currentCatalog.name()

Review Comment:
   **Blocking (P1):** `withAnalyzeContext` applies `current_catalog`, 
`current_schema` and `current_path` by snapshotting the invoker session's 
`CatalogManager` and `SQLConf` state, writing the overrides into it, and 
writing the snapshot back in `finally`. That state is shared by every query of 
the session (Spark Connect runs each operation on its own thread), and nothing 
here owns or serializes it:
   
   - **Overlapping calls lose the restore.** Take call A with `current_schema 
=> 'a'` and call B with `current_schema => 'b'` on a session whose schema is 
`default`: A snapshots `default` and sets `a`; B snapshots `a` and sets `b`; A 
restores `default`; B restores `a`. The session ends on schema `a` although no 
`USE` ran, so later unqualified reads and writes resolve against `a`. The 
stored path and `spark.sql.path.enabled` lose updates the same way, and each 
call can analyze against the other's context.
   - **A failing restore skips the rest.** `setCurrentNamespace(prevNamespace)` 
re-validates the snapshot through `assertNamespaceExist`. If the invoker's 
current schema was dropped, or a v2 `SupportsNamespaces` catalog's implicit 
`defaultNamespace()` is not reported by `namespaceExists`, it throws: the path 
and `spark.sql.path.enabled` restores are skipped, a same-catalog 
`current_schema` override stays in place, and every statement's result is 
replaced with an unrelated `SCHEMA_NOT_FOUND` error.
   - **`spark.sql.path.enabled` is pinned.** When the key was never set, the 
restore calls `setConf(SQLConf.PATH_ENABLED, prevPathEnabled)` instead of 
unsetting it. The key stays explicitly `false`, `SET` lists it, and views and 
SQL functions created afterwards capture it through `CapturesConfig`.
   
   The usage text documents that concurrent queries can observe the override 
during the call, not that the session can stay changed after it. Only ANALYZE 
calls that pass at least one override are affected. No test covers overlapping 
calls or a failing restore, and both `current_path` tests enable the path 
feature first.
   
   **Recommended change:** Stop mutating the invoker session for 
current_catalog, current_schema and current_path: analyze the batch with a 
per-call analyzer obtained from Analyzer.withCatalogManager over a delegating 
CatalogManager that reports the requested catalog, namespace and SQL path 
(following the TransactionAwareCatalogManager pattern), run it under 
SQLConf.withExistingConf with a per-call copy of the conf that enables the path 
feature when current_path is given, use the same per-call manager for overlay 
canonicalization and reported-name qualification, delete the 
snapshot-and-restore and the documented caveat that concurrent queries can 
observe the override; add override tests for concurrent calls, a failing 
namespace check and the path feature left at its default.
   
   **Why this works:** Because the override lives only in objects owned by the 
call, nothing has to be restored: overlapping calls cannot capture each other's 
values, a failing namespace check cannot leave the session changed, other 
queries never observe the override, and spark.sql.path.enabled is never written 
to the session conf.
   
   **Scope:** Replace the session mutate-and-restore with analysis-scoped 
catalog, namespace, path and conf for ANALYZE; a small delegating 
CatalogManager is added for this.
   
   **Compatibility:** Statements of the batch still see the invoker's temp 
views, session variables and functions, and every override still applies to 
every statement of the call.
   
   **Risks:** Readers that bypass the analyzer's CatalogManager, such as 
SessionCatalog's own current database, would keep resolving against the invoker 
schema unless the per-call manager also covers them.
   
   **Constraints:** The override must apply to every resolution step of the 
call, including overlay canonicalization and reported-name qualification, while 
the invoker session is never modified.
   
   **Success:** After any ANALYZE call, including overlapping calls and calls 
whose context check fails, the invoker session's current catalog, namespace, 
SQL path and spark.sql.path.enabled entry are exactly as before the call. 
Within the call, statements still resolve and are reported against the 
requested catalog, schema and path.
   
   **Verification:**
   - **Regression:** Two overlapping ANALYZE calls with different 
current_schema values on one session each analyze against their own schema and 
leave the session's current schema unchanged.
   - **Regression:** An ANALYZE call with current_schema when the invoker's 
current namespace no longer exists returns its per-statement results and leaves 
the session unchanged.
   - **Regression:** An ANALYZE call with current_path in a session where 
spark.sql.path.enabled was never set leaves the key unset, and a view created 
afterwards does not capture it.
   - **Behavior:** current_catalog, current_schema and current_path still 
change resolution and reported qualification within the call, for the session 
catalog and for a v2 catalog.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +127,629 @@ 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 with optional context overrides.
+   * Statements are analyzed in order without being executed: each sees the
+   * variables, tables, and views declared or created by earlier statements
+   * (see [[AnalyzeState]]).
+   * 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) { env =>
+            val state = new AnalyzeState(env)
+            JArray(parsed.map {
+              case ParseFailed(errorObj) => errorObj
+              case Parsed(segment, plan) =>
+                state.statement(plan, 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)
+        }
+      }
+    }
+  }
+
+  /** 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))
+  }
+
+  /**
+   * Analysis state carried across the statements of one ANALYZE batch. It
+   * starts from the caller-supplied variables and relations and records the
+   * effect of each analyzed statement on later ones, without executing
+   * anything:
+   *  - DECLARE VARIABLE adds a session variable. DROP TEMPORARY VARIABLE
+   *    removes it.
+   *  - CREATE [OR REPLACE] TABLE / VIEW / TEMP VIEW (column list, AS query,
+   *    LIKE, or a temp view USING with a schema) adds a schema-only relation 
of
+   *    that identity. DROP TABLE / VIEW removes a relation known this way.
+   * Objects dropped that exist in the catalog stay visible, and statements 
that
+   * change a schema in place (for example ALTER TABLE) are not tracked.
+   */
+  private final class AnalyzeState(env: AnalyzeEnv) {
+    private val session = env.session
+    private def mgr: CatalogManager = session.sessionState.catalogManager
+    private def resolver: (String, String) => Boolean = 
session.sessionState.conf.resolver
+
+    private var variables: List[(Seq[String], DataType)] = env.variables.map {
+      case (name, dataType) => SuppliedVariableRegistry.canonicalize(name) -> 
dataType
+    }.toList
+    private var relations: List[(Seq[String], StructType)] = env.relations.map 
{
+      case (name, schema) => SuppliedRelationRegistry.canonicalize(mgr, name) 
-> schema
+    }.toList
+
+    /**
+     * Analyze one statement with the known objects, render its JSON, then
+     * record its effect.
+     */
+    def statement(plan: LogicalPlan, span: StatementSpan): JObject = {
+      SuppliedVariableRegistry.create(variables).runWith {
+        SuppliedRelationRegistry.create(relations).runWith {
+          val (analyzed, error) = analyze(plan)
+          val json = successJson(plan, analyzed, error, Some(session), span)
+          record(plan, analyzed)
+          json
+        }
+      }
+    }
+
+    private def analyze(
+        plan: LogicalPlan,
+        markers: Seq[(String, DataType)] = env.parameterMarkers)
+        : (Option[LogicalPlan], Option[SparkThrowable with Throwable]) = {
+      try {
+        val toAnalyze = bindSuppliedParameterMarkers(plan, markers)
+        val analyzed =
+          session.sessionState.executePlan(toAnalyze, 
CommandExecutionMode.SKIP).analyzed
+        (Some(analyzed), None)
+      } catch {
+        case e: SparkThrowable with Throwable if !isInternalError(e) => (None, 
Some(e))
+      }
+    }
+
+    private def record(plan: LogicalPlan, analyzed: Option[LogicalPlan]): Unit 
= plan match {
+      case c: CreateVariable =>
+        variableType(c, analyzed).foreach { dataType =>
+          c.names.foreach {
+            case UnresolvedIdentifier(nameParts, _) => 
declareVariable(nameParts, dataType)
+            case _ =>
+          }
+        }
+      case DropVariable(UnresolvedIdentifier(nameParts, _), _) =>
+        val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+        variables = variables.filterNot { case (name, _) => sameName(name, 
identity) }
+      case c: CreateTable =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach {
+          createRelation(_, StructType(c.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))
+        }
+      case r: ReplaceTable =>
+        nameOf(r.name).foreach {
+          createRelation(_, StructType(r.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))
+        }
+      case c: CreateTableAsSelect =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach { name =>
+          querySchema(c.query).foreach(createRelation(name, _))
+        }
+      case r: ReplaceTableAsSelect =>
+        nameOf(r.name).foreach(name => 
querySchema(r.query).foreach(createRelation(name, _)))
+      case c: CreateTableLike =>
+        for {
+          name <- nameOf(c.name) if !(c.ifNotExists && relationExists(name))
+          source <- nameOf(c.source)
+          schema <- querySchema(UnresolvedRelation(source))
+        } createRelation(name, schema)
+      case v: CreateView =>
+        nameOf(v.child).filterNot(n => v.allowExisting && 
relationExists(n)).foreach { name =>
+          querySchema(v.query).foreach(s =>
+            createRelation(name, withColumnNames(s, v.userSpecifiedColumns)))
+        }
+      case v: CreateViewCommand =>
+        val name = viewIdentity(v.name, v.viewType)
+        if (!(v.allowExisting && relationExists(name))) {
+          querySchema(v.plan).foreach(s =>
+            createRelation(name, withColumnNames(s, v.userSpecifiedColumns)))
+        }
+      case v: CreateTempViewUsing =>
+        val viewType = if (v.global) GlobalTempView else LocalTempView
+        
v.userSpecifiedSchema.foreach(createRelation(viewIdentity(v.tableIdent, 
viewType), _))
+      case d: DropTable => nameOf(d.child).foreach(dropRelation)
+      case d: DropView => nameOf(d.child).foreach(dropRelation)
+      case _ =>
+    }
+
+    /** The declared type, else the type of the analyzed DEFAULT expression. */
+    private def variableType(
+        c: CreateVariable,
+        analyzed: Option[LogicalPlan]): Option[DataType] = c.defaultExpr.child 
match {
+      case cast: Cast => Some(cast.dataType)
+      case Literal(null, dataType) => Some(dataType)
+      case _ =>
+        analyzed.flatMap(_.collectFirst {
+          case v: CreateVariable if v.defaultExpr.child.resolved => 
v.defaultExpr.child.dataType
+        })
+    }
+
+    private def declareVariable(nameParts: Seq[String], dataType: DataType): 
Unit = {
+      val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+      variables = (identity -> dataType) ::
+        variables.filterNot { case (name, _) => sameName(name, identity) }
+    }
+
+    private def nameOf(plan: LogicalPlan): Option[Seq[String]] = plan match {
+      case UnresolvedIdentifier(nameParts, _) => Some(nameParts)
+      case u: UnresolvedTableOrView => Some(u.multipartIdentifier)
+      case u: UnresolvedTable => Some(u.multipartIdentifier)
+      case u: UnresolvedView => Some(u.multipartIdentifier)
+      case u: UnresolvedRelation => Some(u.multipartIdentifier)
+      case _ => None
+    }
+
+    private def viewIdentity(name: TableIdentifier, viewType: ViewType): 
Seq[String] =
+      viewType match {
+        case LocalTempView => 
SuppliedRelationRegistry.sessionIdentity(name.table)
+        case GlobalTempView => 
Seq(session.sessionState.catalog.globalTempDatabase, name.table)
+        case _ => tableIdentifierParts(name)
+      }
+
+    private def withColumnNames(
+        schema: StructType,
+        columns: Seq[(String, Option[String])]): StructType = {
+      if (columns.nonEmpty && columns.length == schema.length) {
+        StructType(schema.fields.zip(columns).map { case (f, (name, _)) => 
f.copy(name = name) })
+      } else {
+        schema
+      }
+    }
+
+    private def querySchema(query: LogicalPlan): Option[StructType] =
+      analyze(query)._1.map(_.schema)
+
+    /**
+     * Whether an object of the identity `nameParts` names already exists, for
+     * IF NOT EXISTS. This checks that identity only, not what the name would
+     * resolve to through the path.
+     */
+    private def relationExists(nameParts: Seq[String]): Boolean = {
+      val identity = SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+      val catalog = session.sessionState.catalog
+      relations.exists(r => sameName(r._1, identity)) || {
+        try {
+          if (CatalogManager.isSessionQualifiedViewName(identity)) {
+            catalog.getRawTempView(identity.last).isDefined
+          } else if (identity.length == 2 && 
catalog.isGlobalTempViewDB(identity.head)) {
+            catalog.getRawGlobalTempView(identity.last).isDefined
+          } else {
+            mgr.catalog(identity.head) match {
+              case tables: TableCatalog =>
+                tables.tableExists(Identifier.of(identity.tail.init.toArray, 
identity.last))
+              case _ => false
+            }
+          }
+        } catch {
+          case NonFatal(_) => false
+        }
+      }
+    }
+
+    private def createRelation(nameParts: Seq[String], schema: StructType): 
Unit = {
+      if (schema.nonEmpty) {
+        val identity = SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+        relations = (identity -> schema) ::
+          relations.filterNot { case (name, _) => sameName(name, identity) }
+      }
+    }
+
+    /** A one-part name drops the temp view of that name first, as DROP does. 
*/
+    private def dropRelation(nameParts: Seq[String]): Unit = {
+      val sessionIdentity = 
SuppliedRelationRegistry.sessionIdentity(nameParts.last)
+      val identity =
+        if (nameParts.length == 1 && relations.exists(r => sameName(r._1, 
sessionIdentity))) {
+          sessionIdentity
+        } else {
+          SuppliedRelationRegistry.canonicalize(mgr, nameParts)
+        }
+      relations = relations.filterNot { case (name, _) => sameName(name, 
identity) }
+    }
+
+    private def sameName(left: Seq[String], right: Seq[String]): Boolean =
+      left.length == right.length && left.zip(right).forall { case (a, b) => 
resolver(a, b) }
   }
 
-  private def parseStatement(segment: PositionedSqlStatement): JObject = {
+  /** 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)))
+    }
+  }
+
+  /** Session and caller-supplied objects for one ANALYZE batch. */
+  private case class AnalyzeEnv(
+      session: ClassicSparkSession,
+      parameterMarkers: Seq[(String, DataType)],
+      variables: Seq[(Seq[String], DataType)],
+      relations: Seq[(Seq[String], StructType)])
+
+  /**
+   * Run `f` with optional catalog / schema / SQL path overrides, then restore
+   * the invoker session. `None` on a field keeps the current value. Supplied
+   * variables and relations are validated here and overlaid per statement.
+   */
+  private def withAnalyzeContext[T](
+      session: ClassicSparkSession,
+      context: ParseSqlAnalyzeContext)(f: AnalyzeEnv => T): T = {
+    val parsedVars = context.variables.map(parseKnownVariable)
+    val parsedRels = context.relations.map(parseKnownRelation)
+    val parsedMarkers = context.parameterMarkers.map(parseKnownParameterMarker)
+    val env = AnalyzeEnv(session, parsedMarkers, parsedVars, parsedRels)
+    val needsSessionOverride =
+      context.catalog.nonEmpty || context.schema.nonEmpty || 
context.path.nonEmpty
+    if (!needsSessionOverride) {
+      f(env)
+    } 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(env)
+      } 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)
+      }
+    }
+  }
+
+  private def parseKnownVariable(
+      spec: (String, String)): (Seq[String], DataType) = {
+    val (name, typeDdl) = spec
+    val parts = parseKnownName(name, "variables")
+    val dataType = try {
+      DataType.fromDDL(typeDdl)
+    } catch {
+      case _: ParseException | _: AnalysisException =>
+        throw QueryExecutionErrors.invalidParseSqlContext("variables", typeDdl)
+    }
+    (parts, dataType)
+  }
+
+  private def parseKnownRelation(
+      spec: (String, String)): (Seq[String], StructType) = {
+    val (name, schemaDdl) = spec
+    val parts = parseKnownName(name, "relations")
+    val dataType = try {
+      DataType.fromDDL(schemaDdl)
+    } catch {
+      case _: ParseException | _: AnalysisException =>
+        throw QueryExecutionErrors.invalidParseSqlContext("relations", 
schemaDdl)
+    }
+    dataType match {
+      case schema: StructType if schema.nonEmpty =>
+        (parts, CharVarcharUtils.replaceCharVarcharWithStringInSchema(schema))
+      case _ =>
+        throw QueryExecutionErrors.invalidParseSqlContext("relations", 
schemaDdl)
+    }
+  }
+
+  private def parseKnownName(name: String, parameter: String): Seq[String] = {
+    val parts = try {
+      parser.get().parseMultipartIdentifier(name)
+    } catch {
+      case _: ParseException =>
+        throw QueryExecutionErrors.invalidParseSqlContext(parameter, name)
+    }
+    if (parts.isEmpty) {
+      throw QueryExecutionErrors.invalidParseSqlContext(parameter, name)
+    }
+    parts
+  }
+
+  private def parseKnownParameterMarker(
+      spec: (String, String)): (String, DataType) = {
+    val (name, typeDdl) = spec
+    val dataType = try {
+      DataType.fromDDL(typeDdl)
+    } catch {
+      case _: ParseException | _: AnalysisException =>
+        throw QueryExecutionErrors.invalidParseSqlContext("parameter_markers", 
typeDdl)
+    }
+    val trimmed = name.trim
+    if (trimmed == "?") {
+      ("?", dataType)
+    } else {
+      val parts = parseKnownName(trimmed, "parameter_markers")
+      if (parts.length != 1) {
+        throw QueryExecutionErrors.invalidParseSqlContext("parameter_markers", 
name)
+      }
+      (parts.head, dataType)
+    }
+  }
+
+  /**
+   * Bind caller-supplied marker types using the same [[BindParameters]] rule 
as
+   * `sql(..., args)`, then analyze the bound plan.
+   */
+  private def bindSuppliedParameterMarkers(
+      plan: LogicalPlan,
+      markers: Seq[(String, DataType)]): LogicalPlan = {
+    // Markers may sit in subqueries and CTE definitions, so look deep.
+    val used = if (markers.isEmpty) None else 
collectPlanReferences(plan).parameterMarkers
+    if (used.isEmpty) {
+      plan
+    } else {
+      val namedArgs = markers.collect {
+        case (name, dataType) if name != "?" =>
+          name -> Literal.create(null, dataType)

Review Comment:
   **Nit (P3):** These literals are created while `ParseSql` is being 
constant-folded, outside any statement origin. `TreeNode` captures 
`CurrentOrigin` at construction and the optimizer sets it to the node being 
rewritten, so they carry the origin of the outer `parse_sql(...)` call, and 
`BindParameters` substitutes them unchanged. An analysis error attributed to a 
bound marker therefore reports the outer query's context, which 
`rewriteQueryContextIndexes` treats as statement-relative and re-slices or 
clamps to the whole statement.
   
   For example, `SELECT parse_sql('SELECT * FROM range(10) LIMIT :n', 
'ANALYZE', parameter_markers => array(named_struct('name', 'n', 'type', 
'INT')))` reports `INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL` with a `queryContext` 
derived from the position of `parse_sql` in the outer query rather than the 
span of `:n`. The error class and `analyze_success` are right; only the 
location violates the documented statement-relative code-point contract. Only 
ANALYZE calls with `parameter_markers` are affected.
   
   **Verification:**
   - **Regression:** SELECT parse_sql('SELECT * FROM range(10) LIMIT :n', 
'ANALYZE', parameter_markers => array(named_struct('name', 'n', 'type', 
'INT'))) reports INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL without a context 
derived from the outer query.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -43,18 +100,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):** The "Only `ParseException` / `SqlScriptingException` are 
converted to JSON; unexpected / internal failures propagate" sentence is no 
longer true once ANALYZE exists: `analyzeBatch` and `AnalyzeState.analyze` 
convert every non-`INTERNAL_ERROR` `SparkThrowable` into statement JSON, so for 
example `TABLE_OR_VIEW_NOT_FOUND` does not fail an ANALYZE call. The 
`bindSuppliedParameterMarkers` scaladoc (lines 607-608) is also inaccurate: it 
says the method binds "using the same `BindParameters` rule as `sql(..., 
args)`" and then analyzes the bound plan, but classic `SparkSession.sql(text, 
args)` substitutes parameters at parse time through `parsePlanWithParameters` 
unless the legacy constants-only mode is enabled, and the method returns an 
unanalyzed plan; analysis happens in `AnalyzeState.analyze`.
   
   **Verification:**
   - **Inspection:** Neither comment claims that only parse and scripting 
exceptions become JSON in ANALYZE or that marker binding matches sql(..., args).



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,71 +18,268 @@
 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.catalyst.util.ArrayData
+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.sql.types.{
+  AbstractDataType, ArrayType, DataType, StringType, StructField, StructType
+}
 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. Optional named
+ * `variables`, `relations`, and `parameter_markers` inject known session /
+ * procedure variables, non-persisted relations, and typed parameter markers
+ * so ANALYZE can resolve them.
  */
 // 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[, variables[, relations[, parameter_markers]]]]]]]) -
+    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. Statements are analyzed in order without being 
executed:
+    a later statement sees variables declared and tables or views created or 
dropped by

Review Comment:
   **Nit (P3):** "A later statement sees variables declared and tables or views 
created or dropped by earlier ones" overstates the drop side: 
`AnalyzeState.dropRelation` only removes overlays, so in `DROP TABLE 
existing_t; SELECT * FROM existing_t` the second statement still analyzes 
successfully against the catalog table. The internal `AnalyzeState` doc says 
so, but this usage text, which lists the other effects that are not simulated 
(lines 75-76), does not.
   
   Similarly, lines 95 and 184 promise statement-relative Unicode code-point 
indexes for nested error locations, but `rewriteErrorCharIndexes` rewrites only 
the statement's own contexts and leaves view and `EXECUTE IMMEDIATE` contexts 
as is. ANALYZE reaches those through persistent view bodies and `EXECUTE 
IMMEDIATE` payloads: `parse_sql('EXECUTE IMMEDIATE ''SELECT missing_col''', 
'ANALYZE')` reports indexes into the dynamic string, not into the statement.
   
   See **Shared repair plan 1** in the review body.



##########
docs/sql-ref-parameter-markers.md:
##########
@@ -58,6 +58,8 @@ Parameter markers can be provided by:
 - **Java** using its 
[org.apache.spark.sql.SparkSession.sql()](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/SparkSession.html)
 API.
 - **SQL** using the [`EXECUTE IMMEDIATE` 
statement](sql-ref-syntax-aux-exec-imm.html).
 
+To type-check SQL that contains markers without executing it, `parse_sql` 
`ANALYZE` accepts a `parameter_markers` array of names (`?` or a simple 
identifier) and DDL types. Markers are bound as typed `NULL` values for 
analysis only.

Review Comment:
   **Nit (P3):** This is the only mention of `parse_sql` under `docs/`, and it 
presents `parse_sql` `ANALYZE` as the way to type-check SQL with markers 
without saying that the function is disabled by default. On a default session 
`parse_sql` fails analysis with `FEATURE_NOT_ENABLED` unless 
`spark.sql.function.parseSql.enabled` is `true`, and the function's usage text 
says its JSON contract may still change while that flag stays off by default. 
Readers following this paragraph hit that error with no hint about the 
prerequisite.
   
   **Verification:**
   - **Inspection:** The parameter marker reference page names 
spark.sql.function.parseSql.enabled wherever it presents parse_sql ANALYZE.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +127,629 @@ 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 with optional context overrides.
+   * Statements are analyzed in order without being executed: each sees the
+   * variables, tables, and views declared or created by earlier statements
+   * (see [[AnalyzeState]]).
+   * 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) { env =>
+            val state = new AnalyzeState(env)
+            JArray(parsed.map {
+              case ParseFailed(errorObj) => errorObj
+              case Parsed(segment, plan) =>
+                state.statement(plan, 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)
+        }
+      }
+    }
+  }
+
+  /** 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))
+  }
+
+  /**
+   * Analysis state carried across the statements of one ANALYZE batch. It
+   * starts from the caller-supplied variables and relations and records the
+   * effect of each analyzed statement on later ones, without executing
+   * anything:
+   *  - DECLARE VARIABLE adds a session variable. DROP TEMPORARY VARIABLE
+   *    removes it.
+   *  - CREATE [OR REPLACE] TABLE / VIEW / TEMP VIEW (column list, AS query,
+   *    LIKE, or a temp view USING with a schema) adds a schema-only relation 
of
+   *    that identity. DROP TABLE / VIEW removes a relation known this way.
+   * Objects dropped that exist in the catalog stay visible, and statements 
that
+   * change a schema in place (for example ALTER TABLE) are not tracked.
+   */
+  private final class AnalyzeState(env: AnalyzeEnv) {
+    private val session = env.session
+    private def mgr: CatalogManager = session.sessionState.catalogManager
+    private def resolver: (String, String) => Boolean = 
session.sessionState.conf.resolver
+
+    private var variables: List[(Seq[String], DataType)] = env.variables.map {
+      case (name, dataType) => SuppliedVariableRegistry.canonicalize(name) -> 
dataType
+    }.toList
+    private var relations: List[(Seq[String], StructType)] = env.relations.map 
{
+      case (name, schema) => SuppliedRelationRegistry.canonicalize(mgr, name) 
-> schema
+    }.toList
+
+    /**
+     * Analyze one statement with the known objects, render its JSON, then
+     * record its effect.
+     */
+    def statement(plan: LogicalPlan, span: StatementSpan): JObject = {
+      SuppliedVariableRegistry.create(variables).runWith {
+        SuppliedRelationRegistry.create(relations).runWith {
+          val (analyzed, error) = analyze(plan)
+          val json = successJson(plan, analyzed, error, Some(session), span)
+          record(plan, analyzed)
+          json
+        }
+      }
+    }
+
+    private def analyze(
+        plan: LogicalPlan,
+        markers: Seq[(String, DataType)] = env.parameterMarkers)
+        : (Option[LogicalPlan], Option[SparkThrowable with Throwable]) = {
+      try {
+        val toAnalyze = bindSuppliedParameterMarkers(plan, markers)
+        val analyzed =
+          session.sessionState.executePlan(toAnalyze, 
CommandExecutionMode.SKIP).analyzed
+        (Some(analyzed), None)
+      } catch {
+        case e: SparkThrowable with Throwable if !isInternalError(e) => (None, 
Some(e))
+      }
+    }
+
+    private def record(plan: LogicalPlan, analyzed: Option[LogicalPlan]): Unit 
= plan match {
+      case c: CreateVariable =>
+        variableType(c, analyzed).foreach { dataType =>
+          c.names.foreach {
+            case UnresolvedIdentifier(nameParts, _) => 
declareVariable(nameParts, dataType)
+            case _ =>
+          }
+        }
+      case DropVariable(UnresolvedIdentifier(nameParts, _), _) =>
+        val identity = SuppliedVariableRegistry.canonicalize(nameParts)
+        variables = variables.filterNot { case (name, _) => sameName(name, 
identity) }
+      case c: CreateTable =>
+        nameOf(c.name).filterNot(n => c.ignoreIfExists && 
relationExists(n)).foreach {
+          createRelation(_, StructType(c.columns.map(col =>
+            StructField(col.name, col.dataType, col.nullable))))

Review Comment:
   **Non-blocking (P2):** The overlay schema here keeps the raw `col.dataType`, 
so `CHAR`/`VARCHAR` columns stay `CharType`/`VarcharType`. The same holds for 
`REPLACE TABLE`, and for `CREATE TEMP VIEW ... USING` with a schema, whose 
`userSpecifiedSchema` also keeps them. Caller-supplied relations go through 
`CharVarcharUtils.replaceCharVarcharWithStringInSchema` in 
`parseKnownRelation`; carried-forward ones do not, and `lookupSuppliedRelation` 
turns the stored schema directly into `LocalRelation` attributes. Under the 
default configuration (`spark.sql.preserveCharVarcharTypeInfo` and 
`spark.sql.charVarchar.standardSemantics.enabled` both false), `CheckAnalysis` 
raises `SparkException.internalError` for a leaf whose output has a 
char/varchar type, and `AnalyzeState.analyze` rethrows `INTERNAL_ERROR`.
   
   So `SELECT parse_sql('CREATE TABLE t (id INT, name VARCHAR(100)); SELECT 
name FROM t', 'ANALYZE')` fails the whole call with an internal error and loses 
every statement's result. `CREATE TABLE t2 LIKE t` fails the same way, because 
it analyzes `t` in `querySchema`. The same schema supplied through `relations` 
works. No carry-forward test uses CHAR or VARCHAR.
   
   **Verification:**
   - **Regression:** CREATE TABLE t (id INT, name VARCHAR(100)); SELECT name 
FROM t in one ANALYZE batch returns JSON for both statements, typing name as it 
is typed for a real table with a VARCHAR(100) column.
   - **Regression:** CREATE TABLE t2 LIKE t after such a CREATE TABLE in the 
same batch returns JSON.



##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,71 +18,268 @@
 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.catalyst.util.ArrayData
+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.sql.types.{
+  AbstractDataType, ArrayType, DataType, StringType, StructField, StructType
+}
 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. Optional named
+ * `variables`, `relations`, and `parameter_markers` inject known session /
+ * procedure variables, non-persisted relations, and typed parameter markers
+ * so ANALYZE can resolve them.
  */
 // 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[, variables[, relations[, parameter_markers]]]]]]]) -
+    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. Statements are analyzed in order without being 
executed:
+    a later statement sees variables declared and tables or views created or 
dropped by
+    earlier ones. Dynamic SQL (`EXECUTE IMMEDIATE`), in-place schema changes 
such as ALTER TABLE,
+    USE, SET PATH, SET VAR, and SET of a config are not simulated. `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.
+    Named `variables` is an array of structs (`name`, `type`) for session or 
procedure
+    variables not in the persisted schema. Named `relations` is an
+    array of structs (`name`, `schema`) whose `schema` is a DDL column list, 
parsed like
+    `from_json`. Named `parameter_markers` is an array of structs (`name`, 
`type`) whose
+    `name` is `?` for positional markers or a simple identifier for named 
markers (`:p`).
+    Omitted or NULL injects nothing. `PARSE` ignores the catalog, overlay, and 
marker
+    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`.
+      * variables - Optional ARRAY<STRUCT<name STRING, type STRING>> of known 
variables
+        for `ANALYZE` (for example `system.session.var` or `proc.parm1`). 
`type` is a DDL
+        data type string as accepted by `from_json`. NULL or omitted injects 
none.

Review Comment:
   **Nit (P3):** Several statements in the argument docs do not match the 
implementation:
   
   - The `variables` and `parameter_markers` types and the `relations` schema 
are documented as "as accepted by `from_json`" (here and at lines 124 and 134), 
but they are parsed with `DataType.fromDDL` (`parseKnownVariable`, 
`parseKnownRelation`, `parseKnownParameterMarker`). `from_json` rejects scalar 
schemas such as the documented `INT` and `STRING` with 
`DATATYPE_MISMATCH.INVALID_JSON_SCHEMA`, and accepts JSON-format schemas that 
`fromDDL` rejects, so a caller supplying one gets 
`INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT`.
   - Declared session variables are called persisted (lines 85 and 119, also a 
`parse-sql-analyze.sql` comment and a suite test name), although session 
variables are temporary objects in `system.session`.
   - "Session-shaped names" includes one-part names for `variables` (line 117), 
but the same term is used for `relations` (line 125), where one-part names take 
the analysis catalog and schema.
   - The note "ANALYZE qualifies persistent names, local temp views and temp 
functions as `system.session`" (lines 178-179) reads as if persistent names 
were reported under `system.session`.
   
   See **Shared repair plan 1** in the review body.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -184,23 +184,26 @@ class RelationResolution(
     val isTimeTravel = finalTimeTravelSpec.isDefined || 
hasTimeTravelWriteOptions
     val identifier = u.multipartIdentifier
 
-    // system.session.v (3 parts): only local temp view by name; same as 
SessionCatalog matching.
+    // system.session.v (3 parts): overlay, then local temp view by name.
     if (CatalogManager.isFullyQualifiedSystemSessionViewName(identifier)) {
       val normalized = normalizeSessionQualifiedViewIdentifier(identifier)
-      return resolveTempView(
-        normalized,
-        u.isStreaming,
-        isTimeTravel
-      )
+      return lookupSuppliedTempRelation(identifier, identifier).orElse(

Review Comment:
   **Blocking (P1):** The key passed here is the raw referenced identifier, 
while `SuppliedRelationRegistry.canonicalize` stored the overlay under its own 
qualification: a session-qualified name becomes lower-case 
`system.session.<name>`, a registered-catalog head is kept as the caller 
spelled it, and so on. `lookup` compares parts with `conf.resolver`. 
`isFullyQualifiedSystemSessionViewName` matches `system` and `session` 
case-insensitively, but the key comparison does not, so under 
`spark.sql.caseSensitive=true`:
   
   - With a temp view `v (a INT)` and `relations => array(named_struct('name', 
'system.session.v', 'schema', 'a STRING'))`, `SELECT a FROM SYSTEM.SESSION.v` 
misses the overlay, falls back to `resolveTempView`, and reports `a` as `int`.
   - An overlay named `SPARK_CATALOG.default.t` is stored with that spelling 
but looked up as `spark_catalog.default.t` (persistent lookups key on 
`ident.toQualifiedNameParts(catalog)`, i.e. `catalog.name()`), so the real 
table `t` is analyzed instead.
   
   The global temp branch also passes the raw identifier, and 
`AnalyzeState.relationExists` and `dropRelation` repeat the qualification 
rules, which then have to keep tracking `CatalogAndIdentifier` and 
`expandIdentifier`. A miss is silent: ANALYZE types and validates the statement 
against the object the caller asked to eclipse. The trigger needs 
case-sensitive analysis and a non-canonical spelling; default case-insensitive 
analysis is unaffected.
   
   **Verification:**
   - **Regression:** With spark.sql.caseSensitive=true, an overlay for 
system.session.v eclipses a temp view v when the query references 
SYSTEM.SESSION.v.
   - **Regression:** With spark.sql.caseSensitive=true, an overlay named with 
an upper-case spelling of the session catalog eclipses the table it names.
   - **Behavior:** Existing overlay tests for one-part, qualified and session 
names keep their results.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala:
##########
@@ -3704,4 +3704,28 @@ private[sql] object QueryExecutionErrors extends 
QueryErrorsBase with ExecutionE
       context = Array.empty,
       summary = "")
   }
+
+  def invalidParseSqlMode(mode: String): SparkRuntimeException = {
+    new SparkRuntimeException(
+      errorClass = "INVALID_PARAMETER_VALUE.PARSE_SQL_MODE",
+      messageParameters = Map(
+        "functionName" -> toSQLId("parse_sql"),
+        "parameter" -> toSQLId("mode"),
+        "invalidValue" -> toSQLValue(mode, StringType)))

Review Comment:
   **Nit (P3):** `INVALID_PARAMETER_VALUE.PARSE_SQL_MODE` now has two producers 
that format `invalidValue` differently. 
`QueryCompilationErrors.invalidParseSqlMode`, used when `mode` is constant, 
goes through `invalidParameter` and `toSQLExpr`, so `parse_sql('SELECT 1', 
'bogus')` reports the value double-quoted. This helper, used when a 
non-constant `mode` is evaluated per row, uses `toSQLValue` and renders the 
same value single-quoted. `invalidParseSqlContext` below also renders the 
sentinels that `decodeNameSpecArray` passes for a NULL element or field 
(`NULL`, `NULL name`) as string literals, so a NULL entry in `variables` 
reports exactly what a literal `'NULL'` string would. Clients and tests that 
match `messageParameters` see different values for the same mistake and cannot 
tell a missing value from a string.
   
   **Verification:**
   - **Compatibility:** For one invalid mode string, every path that can raise 
INVALID_PARAMETER_VALUE.PARSE_SQL_MODE reports the same invalidValue.
   - **Regression:** A NULL element in variables reports a value 
distinguishable from a literal 'NULL' entry.



-- 
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: reviews-unsubscr...@spark.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org