diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala
index 16112402350ef..6bdcab8c58675 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala
@@ -58,6 +58,26 @@ case class SqlStatementSplitResult(
   def isEmpty: Boolean = completeStatements.isEmpty && partialStatement.isEmpty
 }
 
+/** A split SQL statement together with its 0-based start in the original input. */
+private[sql] case class PositionedSqlStatement(
+    statement: String,
+    terminator: String,
+    start: Int) {
+  def length: Int = statement.length
+}
+
+/** Internal splitter result that retains source positions. */
+private[sql] case class PositionedSqlStatementSplitResult(
+    completeStatements: Seq[PositionedSqlStatement],
+    partialStatement: Option[PositionedSqlStatement],
+    hasUnclosedComment: Boolean) {
+
+  def withoutPositions: SqlStatementSplitResult = SqlStatementSplitResult(
+    completeStatements.map(s => SqlStatement(s.statement, s.terminator)),
+    partialStatement.map(_.statement).getOrElse(""),
+    hasUnclosedComment)
+}
+
 /**
  * A parser-based SQL statement splitter, inspired by Trino's
  * `io.trino.cli.lexer.StatementSplitter`.
@@ -111,7 +131,7 @@ object SqlStatementSplitter {
 
   /** Split the given SQL text into individual statements at `;` boundaries. */
   def split(sqlText: String): SqlStatementSplitResult =
-    split(sqlText, identity)
+    splitWithPositions(sqlText, identity).withoutPositions
 
   /**
    * Split the given SQL text, applying `validationPreprocess` to each candidate
@@ -121,10 +141,26 @@ object SqlStatementSplitter {
    *
    * Pass `identity` for a pure original-text splitter (the default).
    */
-  def split(sqlText: String, validationPreprocess: String => String): SqlStatementSplitResult = {
+  def split(
+      sqlText: String,
+      validationPreprocess: String => String): SqlStatementSplitResult =
+    splitWithPositions(sqlText, validationPreprocess).withoutPositions
+
+  /**
+   * Split SQL while retaining each statement's source position. This is used by
+   * parse-only tooling that reports spans into the original input.
+   */
+  private[sql] def splitWithPositions(
+      sqlText: String,
+      validationPreprocess: String => String): PositionedSqlStatementSplitResult = {
     require(sqlText != null, "sqlText must not be null")
     require(validationPreprocess != null, "validationPreprocess must not be null")
 
+    // CodePointCharStream token offsets count Unicode code points, while
+    // String offsets and lengths count UTF-16 code units. Map once so each
+    // token-boundary lookup is O(1) rather than rescanning the prefix.
+    val toUtf16 = utf16Offsets(sqlText)
+
     val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(sqlText)))
     lexer.removeErrorListeners()
     val tokenStream = new CommonTokenStream(lexer)
@@ -142,8 +178,9 @@ object SqlStatementSplitter {
       acc.toArray
     }
 
-    val completeStatements = mutable.ArrayBuffer.empty[SqlStatement]
+    val completeStatements = mutable.ArrayBuffer.empty[PositionedSqlStatement]
     val buffer = new StringBuilder()
+    var bufferStart = -1
     // Whether `buffer` contains any non-hidden token (i.e. any actual SQL content
     // beyond whitespace and comments). Chunks that only contain whitespace/comments
     // are dropped, matching the spark-sql CLI's long-standing behavior.
@@ -158,6 +195,31 @@ object SqlStatementSplitter {
     // interpretation (e.g. `double_quoted_identifiers`).
     val conf = SqlApiConf.get
 
+    def appendToken(token: Token): Unit = {
+      if (buffer.isEmpty) bufferStart = toUtf16(token.getStartIndex)
+      buffer.append(token.getText)
+    }
+
+    def resetBuffer(): Unit = {
+      buffer.setLength(0)
+      bufferStart = -1
+      bufferHasContent = false
+    }
+
+    def positionedStatement(terminator: String): Option[PositionedSqlStatement] = {
+      val raw = buffer.toString
+      val (leadingWhitespace, statement) = trimSqlWhitespace(raw)
+      if (statement.isEmpty) {
+        None
+      } else {
+        assert(bufferStart >= 0)
+        Some(PositionedSqlStatement(
+          statement,
+          terminator,
+          bufferStart + leadingWhitespace))
+      }
+    }
+
     while (!stopOuter && index < numTokens) {
       val startIdx = nextSignificantTokenIndex(tokenStream, index)
       if (startIdx < 0) {
@@ -167,7 +229,7 @@ object SqlStatementSplitter {
         while (index < numTokens) {
           val tok = tokenStream.get(index)
           index += 1
-          if (tok.getType != Token.EOF) buffer.append(tok.getText)
+          if (tok.getType != Token.EOF) appendToken(tok)
         }
         stopOuter = true
       } else if (tokenStream.get(startIdx).getType == SqlBaseLexer.SEMICOLON) {
@@ -192,7 +254,8 @@ object SqlStatementSplitter {
         while (!parsedOk && !failedNonEof && d < delimiterPositions.length) {
           val candidateEnd = delimiterPositions(d)
           tryParseRegion(
-            sqlText, tokenStream, startIdx, candidateEnd, validationPreprocess, conf) match {
+            sqlText, toUtf16, tokenStream, startIdx, candidateEnd,
+            validationPreprocess, conf) match {
             case ParsedOk =>
               parsedOk = true
               matchedDelimIdx = candidateEnd
@@ -216,17 +279,13 @@ object SqlStatementSplitter {
           while (index < matchedDelimIdx) {
             val tok = tokenStream.get(index)
             if (tok.getChannel != Token.HIDDEN_CHANNEL) bufferHasContent = true
-            buffer.append(tok.getText)
+            appendToken(tok)
             index += 1
           }
           if (bufferHasContent) {
-            val stmt = buffer.toString.trim
-            if (stmt.nonEmpty) {
-              completeStatements += SqlStatement(stmt, terminator)
-            }
+            positionedStatement(terminator).foreach(completeStatements += _)
           }
-          buffer.setLength(0)
-          bufferHasContent = false
+          resetBuffer()
           index = matchedDelimIdx + 1
           delimSearchStart = d + 1
         } else if (failedNonEof) {
@@ -253,17 +312,13 @@ object SqlStatementSplitter {
               stopInner = true
             } else if (token.getType == SqlBaseLexer.SEMICOLON) {
               if (bufferHasContent) {
-                val stmt = buffer.toString.trim
-                if (stmt.nonEmpty) {
-                  completeStatements += SqlStatement(stmt, token.getText)
-                }
+                positionedStatement(token.getText).foreach(completeStatements += _)
               }
-              buffer.setLength(0)
-              bufferHasContent = false
+              resetBuffer()
               stopInner = true
             } else {
               if (token.getChannel != Token.HIDDEN_CHANNEL) bufferHasContent = true
-              buffer.append(token.getText)
+              appendToken(token)
             }
           }
         } else {
@@ -275,7 +330,7 @@ object SqlStatementSplitter {
             index += 1
             if (tok.getType != Token.EOF) {
               if (tok.getChannel != Token.HIDDEN_CHANNEL) bufferHasContent = true
-              buffer.append(tok.getText)
+              appendToken(tok)
             }
           }
           stopOuter = true
@@ -285,8 +340,11 @@ object SqlStatementSplitter {
 
     val unclosed = lexer.has_unclosed_bracketed_comment
     val partial =
-      if (bufferHasContent || unclosed) buffer.toString.trim else ""
-    SqlStatementSplitResult(completeStatements.toSeq, partial, unclosed && partial.nonEmpty)
+      if (bufferHasContent || unclosed) positionedStatement("") else None
+    PositionedSqlStatementSplitResult(
+      completeStatements.toSeq,
+      partial,
+      unclosed && partial.nonEmpty)
   }
 
   /** Outcome of attempting to parse one statement candidate. */
@@ -301,13 +359,14 @@ object SqlStatementSplitter {
    * position of the trailing `;` token whose char range belongs to the
    * region) as a complete top-level Spark SQL statement.
    *
-   * The region is extracted from the original source by char-offset
-   * (`Token.getStartIndex` / `getStopIndex`), `validationPreprocess` is
-   * applied to it, and the result is re-lexed and parsed with a fresh
-   * [[SqlBaseParser]]. This isolation means the splitter's parser sees a
-   * sub-stream whose EOF lands right after the trailing `;`, so the existing
-   * `compoundOrSingleStatement` rule (which requires `SEMICOLON* EOF`) acts
-   * as the per-statement validator without any custom grammar rule.
+   * The region is extracted from the original source by converting ANTLR's
+   * Unicode code-point token offsets through a one-time UTF-16 map.
+   * `validationPreprocess` is applied to it, and the result is re-lexed and
+   * parsed with a fresh [[SqlBaseParser]]. This isolation means the splitter's
+   * parser sees a sub-stream whose EOF lands right after the trailing `;`, so
+   * the existing `compoundOrSingleStatement` rule (which requires
+   * `SEMICOLON* EOF`) acts as the per-statement validator without any custom
+   * grammar rule.
    *
    * Uses the same two-stage SLL -> LL prediction strategy as the main parser
    * for performance (most statements parse cleanly with the faster SLL stage).
@@ -326,6 +385,7 @@ object SqlStatementSplitter {
    */
   private def tryParseRegion(
       sqlText: String,
+      toUtf16: Array[Int],
       stream: CommonTokenStream,
       startIdx: Int,
       endIdx: Int,
@@ -333,9 +393,9 @@ object SqlStatementSplitter {
       conf: SqlApiConf): ParseOutcome = {
     val firstTok = stream.get(startIdx)
     val lastTok = stream.get(endIdx)
-    val regionStart = firstTok.getStartIndex
+    val regionStart = toUtf16(firstTok.getStartIndex)
     // Token.getStopIndex is inclusive, substring's upper bound is exclusive.
-    val regionEnd = lastTok.getStopIndex + 1
+    val regionEnd = toUtf16(lastTok.getStopIndex + 1)
     val original = sqlText.substring(regionStart, regionEnd)
     val preprocessed = validationPreprocess(original)
 
@@ -402,6 +462,45 @@ object SqlStatementSplitter {
     parser.setErrorHandler(new BailErrorStrategy)
   }
 
+  /**
+   * Maps each Unicode code-point index to a UTF-16 code-unit offset. The last
+   * entry is `sqlText.length`, so an inclusive ANTLR stop index converts with
+   * `toUtf16(stopIndex + 1)`.
+   */
+  private def utf16Offsets(sqlText: String): Array[Int] = {
+    val cuLen = sqlText.length
+    val offsets = new Array[Int](sqlText.codePointCount(0, cuLen) + 1)
+    var cu = 0
+    var cp = 0
+    while (cu < cuLen) {
+      offsets(cp) = cu
+      cu += Character.charCount(sqlText.codePointAt(cu))
+      cp += 1
+    }
+    offsets(cp) = cuLen
+    offsets
+  }
+
+  // Spark SQL WS token (SqlBaseLexer): ASCII space plus Unicode spaces the
+  // lexer hides. String.trim only strips characters <= U+0020.
+  // Hex literals keep the source ASCII (Scala unicode escapes are still non-ASCII).
+  private def isSqlWhitespace(c: Char): Boolean = c.toInt match {
+    case 0x20 | 0x09 | 0x0A | 0x0C | 0x0D | 0x0B | 0xA0 | 0x1680 |
+         0x2000 | 0x2001 | 0x2002 | 0x2003 | 0x2004 | 0x2005 |
+         0x2006 | 0x2007 | 0x2008 | 0x2009 | 0x200A | 0x2028 |
+         0x202F | 0x205F | 0x3000 => true
+    case _ => false
+  }
+
+  /** Trim lexer whitespace; return (leading UTF-16 count, trimmed text). */
+  private def trimSqlWhitespace(s: String): (Int, String) = {
+    var start = 0
+    var end = s.length
+    while (start < end && isSqlWhitespace(s.charAt(start))) start += 1
+    while (end > start && isSqlWhitespace(s.charAt(end - 1))) end -= 1
+    (start, s.substring(start, end))
+  }
+
   /** Returns the index of the next non-hidden, non-EOF token at or after `from`, or -1. */
   private def nextSignificantTokenIndex(stream: CommonTokenStream, from: Int): Int = {
     var i = from
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitterSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitterSuite.scala
index 2e091aeff974f..dfa07229901d5 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitterSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitterSuite.scala
@@ -71,6 +71,48 @@ class SqlStatementSplitterSuite extends SparkFunSuite {
     assert(result.partialStatement == "select * from")
   }
 
+  test("source positions skip comments attached to empty statements") {
+    val sql = "  select 1 ; /* select 2 */; select 2"
+    val result = SqlStatementSplitter.splitWithPositions(sql, identity)
+    val complete = result.completeStatements.head
+    assert(complete.statement == "select 1")
+    assert(complete.start == 2)
+    assert(complete.length == 8)
+
+    val partial = result.partialStatement.get
+    assert(partial.statement == "select 2")
+    assert(partial.start == sql.lastIndexOf("select 2"))
+    assert(partial.length == 8)
+  }
+
+  test("source positions use UTF-16 offsets for supplementary characters") {
+    val emoji = new String(Character.toChars(0x1F600))
+    val sql = s"SELECT '$emoji$emoji'; SELECT 2;"
+    val result = SqlStatementSplitter.splitWithPositions(sql, identity)
+
+    assert(result.completeStatements.map(_.statement) ==
+      Seq(s"SELECT '$emoji$emoji'", "SELECT 2"))
+    assert(result.completeStatements.map(_.start) == Seq(0, 15))
+    assert(result.completeStatements.map(_.length) == Seq(13, 8))
+    result.completeStatements.foreach { statement =>
+      assert(sql.substring(statement.start, statement.start + statement.length) ==
+        statement.statement)
+    }
+    assert(result.partialStatement.isEmpty)
+  }
+
+  test("source positions trim Spark SQL Unicode whitespace") {
+    val nbsp = 0xA0.toChar
+    val sql = s"${nbsp}SELECT 1$nbsp;"
+    val result = SqlStatementSplitter.splitWithPositions(sql, identity)
+    val complete = result.completeStatements.head
+    assert(complete.statement == "SELECT 1")
+    assert(complete.start == 1)
+    assert(complete.length == 8)
+    assert(sql.substring(complete.start, complete.start + complete.length) ==
+      complete.statement)
+  }
+
   // ----------------------------------------------------------------------------------
   // Error tolerance (mirrors Trino behavior)
   // ----------------------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala
index 649feeee9adf7..9551b9e10acd5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala
@@ -27,38 +27,46 @@ import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
 
 /**
- * Parses a SQL statement string and returns a compact JSON description of the
- * unresolved statement (identifier/code, lineage references, select-list names,
- * parameters), or a STANDARD-format error object when the statement does not
- * parse.
+ * 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.
  *
  * 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.
  */
 // scalastyle:off line.size.limit
+// scalastyle:off nonascii
 @ExpressionDescription(
-  usage = """_FUNC_(sqlStmt) - Parses `sqlStmt` with the stock Spark SQL parser and
-    returns a JSON string describing the statement (parse success, Table 39 statement
-    identifier/code, target and source table references for lineage, select-list column
-    names, and parameter markers). Session parser extensions are not applied.
+  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
-    with `parse_success` false, source location, and a nested STANDARD error object
-    instead of throwing.""",
+    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 `[]`.""",
   arguments = """
     Arguments:
-      * sqlStmt - A SQL statement string to parse.
+      * sqlStmt - A SQL batch string to split and parse.
         An expression that evaluates to a string.
   """,
   examples = """
     Examples:
-      > SELECT _FUNC_('SELECT a, b FROM t');
-       {"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}
-      > SELECT get_json_object(_FUNC_('SELEC'), '$.error.errorClass');
+      > SELECT _FUNC_('SELECT 1;SELECT 2');
+       [{"start":1,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]},{"start":10,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}]
+      > SELECT get_json_object(_FUNC_('SELECT ''😀'';SELECT 2'), '$[1].start');
+       13
+      > SELECT get_json_object(_FUNC_('SELEC'), '$[0].error.errorClass');
        PARSE_SYNTAX_ERROR
   """,
   group = "misc_funcs",
   since = "4.4.0")
+// scalastyle:on nonascii
 // scalastyle:on line.size.limit
 case class ParseSql(child: Expression)
   extends UnaryExpression
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala
index fbc4b96f9dd71..d246bdf3faf4c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala
@@ -33,8 +33,8 @@ import org.apache.spark.sql.execution.command.{CreateViewCommand, DescribeQueryC
 import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
 
 /**
- * Parses a SQL statement string and returns a compact JSON description of the
- * unresolved plan (parse-only; no catalog resolution).
+ * Parses a SQL batch string and returns a compact JSON array describing its
+ * unresolved statements (parse-only; no catalog resolution).
  *
  * Uses a stock [[SparkSqlParser]] (ThreadLocal) so statement coverage matches
  * the default production parser (EXPLAIN / SET / ADD JAR / temp views / etc.).
@@ -43,25 +43,38 @@ import org.apache.spark.sql.execution.datasources.CreateTempViewUsing
  * executors without a session, so only the stock parser is available under
  * distributed eval.
  *
- * On success the JSON always includes `parse_success`, the statement
+ * 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
+ * `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 it returns `parse_success: false` with source location and a nested
- * STANDARD-format error object, and does not throw. Only [[ParseException]] /
- * [[SqlScriptingException]] are converted to JSON; unexpected / internal
- * failures propagate so the function fails.
+ * `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
+ * [[ParseException]] / [[SqlScriptingException]] are converted to JSON;
+ * unexpected / internal failures propagate so the function fails.
  */
 object ParseSqlResult {
 
   private val parser: ThreadLocal[SparkSqlParser] =
     ThreadLocal.withInitial(() => new SparkSqlParser())
 
-  /** Parse `sql` and render the JSON result string. */
+  /** Parse `sql` as a batch and render the JSON result array. */
   def fromSql(sql: String): String = {
+    val split = parser.get().splitStatementsWithPositions(sql)
+    val statements = split.completeStatements ++ split.partialStatement
+    compact(render(JArray(statements.map(parseStatement).toList)))
+  }
+
+  private def parseStatement(segment: PositionedSqlStatement): JObject = {
+    val sql = segment.statement
     try {
       // Do not inherit the outer query's origin from the parse_sql expression.
-      // Errors and parsed nodes must refer to the SQL string passed to this function.
+      // 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))
       } else {
@@ -69,21 +82,23 @@ object ParseSqlResult {
       }
       CurrentOrigin.withOrigin(origin) {
         val plan = parser.get().parsePlan(sql)
-        fromPlan(plan)
+        successJson(plan, segment)
       }
     } catch {
       // User-facing parse / scripting failures become JSON; everything else fails.
       case e: ParseException =>
-        errorJson(e)
+        errorJson(e, segment)
       case e: SqlScriptingException =>
-        errorJson(e)
+        errorJson(e, segment)
     }
   }
 
   /** Build success JSON from an already-parsed unresolved plan. */
-  def fromPlan(plan: LogicalPlan): String = {
+  private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement): JObject = {
     val classification = SqlStatementCodes.classify(plan)
     val fields = mutable.ListBuffer.empty[JField]
+    fields += "start" -> JInt(segment.start + 1)
+    fields += "length" -> JInt(segment.length)
     fields += "parse_success" -> JBool(true)
     fields += "statement_identifier" -> JString(classification.statementIdentifier)
     fields += "statement_code" -> JInt(classification.statementCode)
@@ -105,10 +120,12 @@ object ParseSqlResult {
       fields += "select_list" -> JArray(selectList.toList)
     }
     refs.parameterMarkers.foreach(markers => fields += "parameter_markers" -> markers)
-    compact(render(JObject(fields.toList)))
+    JObject(fields.toList)
   }
 
-  private def errorJson(e: SparkThrowable with Throwable): String = {
+  private def errorJson(
+      e: SparkThrowable with Throwable,
+      segment: PositionedSqlStatement): JObject = {
     val errorObj = parseJson(
       SparkThrowableHelper.getMessage(e, ErrorMessageFormat.STANDARD)).asInstanceOf[JObject]
     val origin = e match {
@@ -122,10 +139,12 @@ object ParseSqlResult {
     } else {
       origin.toSeq.flatMap(queryContextField)
     }
-    compact(render(JObject(
+    JObject(
+      "start" -> JInt(segment.start + 1),
+      "length" -> JInt(segment.length),
       "parse_success" -> JBool(false),
       "error" -> JObject(errorObj.obj ++ contextFields ++ locationFields)
-    )))
+    )
   }
 
   private def queryContextField(origin: Origin): Option[JField] = origin.context match {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 5070a40259e37..bce4250c18975 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -123,6 +123,13 @@ class SparkSqlParser extends AbstractSqlParser {
   override def splitStatements(sqlText: String): SqlStatementSplitResult =
     SqlStatementSplitter.split(sqlText, SparkSqlParser.substituteVariablesForValidation)
 
+  /** Split statements while retaining their positions in the original SQL text. */
+  private[sql] def splitStatementsWithPositions(
+      sqlText: String): PositionedSqlStatementSplitResult =
+    SqlStatementSplitter.splitWithPositions(
+      sqlText,
+      SparkSqlParser.substituteVariablesForValidation)
+
   /**
    * Internal parse method that handles both parameter substitution and regular parsing.
    *
diff --git a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
index 6f11ca0435ae9..b80684843fa60 100644
--- a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
+++ b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
@@ -279,7 +279,7 @@
 | org.apache.spark.sql.catalyst.expressions.OctetLength | octet_length | SELECT octet_length('Spark SQL') | struct<octet_length(Spark SQL):int> |
 | org.apache.spark.sql.catalyst.expressions.Or | or | SELECT true or false | struct<(true OR false):boolean> |
 | org.apache.spark.sql.catalyst.expressions.Overlay | overlay | SELECT overlay('Spark SQL' PLACING '_' FROM 6) | struct<overlay(Spark SQL, _, 6, -1):string> |
-| org.apache.spark.sql.catalyst.expressions.ParseSql | parse_sql | SELECT parse_sql('SELECT a, b FROM t') | struct<parse_sql(SELECT a, b FROM t):string> |
+| org.apache.spark.sql.catalyst.expressions.ParseSql | parse_sql | SELECT parse_sql('SELECT 1;SELECT 2') | struct<parse_sql(SELECT 1;SELECT 2):string> |
 | org.apache.spark.sql.catalyst.expressions.ParseToDate | to_date | SELECT to_date('2009-07-30 04:17:52') | struct<to_date(2009-07-30 04:17:52):date> |
 | org.apache.spark.sql.catalyst.expressions.ParseToTimestamp | to_timestamp | SELECT to_timestamp('2016-12-31 00:12:00') | struct<to_timestamp(2016-12-31 00:12:00):timestamp> |
 | org.apache.spark.sql.catalyst.expressions.ParseToTimestampLTZExpressionBuilder | to_timestamp_ltz | SELECT to_timestamp_ltz('2016-12-31 00:12:00') | struct<to_timestamp_ltz(2016-12-31 00:12:00):timestamp> |
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/parse-sql.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/parse-sql.sql.out
index bc9bbba7c3c71..30d0d993176f9 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/parse-sql.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/parse-sql.sql.out
@@ -20,14 +20,35 @@ Project [parse_sql(SELECT db.my_func(a), count(b) FROM cat.ns.t1 JOIN t2) AS par
 +- OneRowRelation
 
 
+-- !query
+SELECT parse_sql('select 1; select 2')
+-- !query analysis
+Project [parse_sql(select 1; select 2) AS parse_sql(select 1; select 2)#x]
++- OneRowRelation
+
+
+-- !query
+SELECT parse_sql('select 1; select 2;')
+-- !query analysis
+Project [parse_sql(select 1; select 2;) AS parse_sql(select 1; select 2;)#x]
++- OneRowRelation
+
+
+-- !query
+SELECT parse_sql('SELECT 1; SELEC 2; SELECT 3')
+-- !query analysis
+Project [parse_sql(SELECT 1; SELEC 2; SELECT 3) AS parse_sql(SELECT 1; SELEC 2; SELECT 3)#x]
++- OneRowRelation
+
+
 -- !query
 SELECT
-  get_json_object(result, '$.statement_identifier') AS statement_identifier,
-  get_json_object(result, '$.source_table_references[0][0]') AS first_table,
-  get_json_object(result, '$.select_list[1].name[0]') AS second_column
+  get_json_object(result, '$[0].statement_identifier') AS statement_identifier,
+  get_json_object(result, '$[0].source_table_references[0][0]') AS first_table,
+  get_json_object(result, '$[0].select_list[1].name[0]') AS second_column
 FROM (SELECT parse_sql('SELECT a, b FROM t') AS result)
 -- !query analysis
-Project [get_json_object(result#x, $.statement_identifier) AS statement_identifier#x, get_json_object(result#x, $.source_table_references[0][0]) AS first_table#x, get_json_object(result#x, $.select_list[1].name[0]) AS second_column#x]
+Project [get_json_object(result#x, $[0].statement_identifier) AS statement_identifier#x, get_json_object(result#x, $[0].source_table_references[0][0]) AS first_table#x, get_json_object(result#x, $[0].select_list[1].name[0]) AS second_column#x]
 +- SubqueryAlias __auto_generated_subquery_name
    +- Project [parse_sql(SELECT a, b FROM t) AS result#x]
       +- OneRowRelation
@@ -251,12 +272,12 @@ Project [parse_sql(SELEC FROM t) AS parse_sql(SELEC FROM t)#x]
 
 -- !query
 SELECT
-  get_json_object(result, '$.parse_success') AS parse_success,
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[0].parse_success') AS parse_success,
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.queryContext[0].fragment') AS fragment
 FROM (SELECT parse_sql('SELEC FROM t') AS result)
 -- !query analysis
-Project [get_json_object(result#x, $.parse_success) AS parse_success#x, get_json_object(result#x, $.error.errorClass) AS error_class#x, get_json_object(result#x, $.error.queryContext[0].fragment) AS fragment#x]
+Project [get_json_object(result#x, $[0].parse_success) AS parse_success#x, get_json_object(result#x, $[0].error.errorClass) AS error_class#x, get_json_object(result#x, $[0].error.queryContext[0].fragment) AS fragment#x]
 +- SubqueryAlias __auto_generated_subquery_name
    +- Project [parse_sql(SELEC FROM t) AS result#x]
       +- OneRowRelation
@@ -281,10 +302,10 @@ Project [parse_sql(SELECT *
 
 -- !query
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].startIndex') AS start_index
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.line') AS line,
+  get_json_object(result, '$[0].error.position') AS position,
+  get_json_object(result, '$[0].error.queryContext[0].startIndex') AS start_index
 FROM (
   SELECT parse_sql(
 'SELECT *
@@ -293,7 +314,7 @@ FROM (
  CLUSTER BY b') AS result
 )
 -- !query analysis
-Project [get_json_object(result#x, $.error.errorClass) AS error_class#x, get_json_object(result#x, $.error.line) AS line#x, get_json_object(result#x, $.error.position) AS position#x, get_json_object(result#x, $.error.queryContext[0].startIndex) AS start_index#x]
+Project [get_json_object(result#x, $[0].error.errorClass) AS error_class#x, get_json_object(result#x, $[0].error.line) AS line#x, get_json_object(result#x, $[0].error.position) AS position#x, get_json_object(result#x, $[0].error.queryContext[0].startIndex) AS start_index#x]
 +- SubqueryAlias __auto_generated_subquery_name
    +- Project [parse_sql(SELECT *
  FROM t
@@ -391,10 +412,12 @@ Project [parse_sql(BEGIN
 
 -- !query
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[1].start') AS start,
+  get_json_object(result, '$[1].length') AS length,
+  get_json_object(result, '$[1].error.errorClass') AS error_class,
+  get_json_object(result, '$[1].error.line') AS line,
+  get_json_object(result, '$[1].error.position') AS position,
+  get_json_object(result, '$[1].error.queryContext[0].fragment') AS fragment
 FROM (
   SELECT parse_sql(
 'BEGIN
@@ -403,7 +426,7 @@ FROM (
  END') AS result
 )
 -- !query analysis
-Project [get_json_object(result#x, $.error.errorClass) AS error_class#x, get_json_object(result#x, $.error.line) AS line#x, get_json_object(result#x, $.error.position) AS position#x, get_json_object(result#x, $.error.queryContext[0].fragment) AS fragment#x]
+Project [get_json_object(result#x, $[1].start) AS start#x, get_json_object(result#x, $[1].length) AS length#x, get_json_object(result#x, $[1].error.errorClass) AS error_class#x, get_json_object(result#x, $[1].error.line) AS line#x, get_json_object(result#x, $[1].error.position) AS position#x, get_json_object(result#x, $[1].error.queryContext[0].fragment) AS fragment#x]
 +- SubqueryAlias __auto_generated_subquery_name
    +- Project [parse_sql(BEGIN
    SELECT 1;
@@ -434,10 +457,10 @@ Project [parse_sql(BEGIN
 
 -- !query
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.line') AS line,
+  get_json_object(result, '$[0].error.position') AS position,
+  get_json_object(result, '$[0].error.queryContext[0].fragment') AS fragment
 FROM (
   SELECT parse_sql(
 'BEGIN
@@ -447,7 +470,7 @@ FROM (
  END') AS result
 )
 -- !query analysis
-Project [get_json_object(result#x, $.error.errorClass) AS error_class#x, get_json_object(result#x, $.error.line) AS line#x, get_json_object(result#x, $.error.position) AS position#x, get_json_object(result#x, $.error.queryContext[0].fragment) AS fragment#x]
+Project [get_json_object(result#x, $[0].error.errorClass) AS error_class#x, get_json_object(result#x, $[0].error.line) AS line#x, get_json_object(result#x, $[0].error.position) AS position#x, get_json_object(result#x, $[0].error.queryContext[0].fragment) AS fragment#x]
 +- SubqueryAlias __auto_generated_subquery_name
    +- Project [parse_sql(BEGIN
    lbl_begin: BEGIN
diff --git a/sql/core/src/test/resources/sql-tests/inputs/parse-sql.sql b/sql/core/src/test/resources/sql-tests/inputs/parse-sql.sql
index b4eb15ffbc38e..6b73c8722c6bb 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/parse-sql.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/parse-sql.sql
@@ -1,5 +1,5 @@
--- End-to-end coverage for parse_sql (SPARK-58738).
--- Returns compact JSON for parse-only statement analysis via SparkSqlParser.
+-- End-to-end coverage for parse_sql (SPARK-59255).
+-- Returns a compact JSON array for parse-only SQL batch analysis via SparkSqlParser.
 -- Off by default while the JSON contract is still evolving.
 --SET spark.sql.function.parseSql.enabled=true
 
@@ -10,11 +10,24 @@ SELECT parse_sql(NULL);
 SELECT parse_sql('SELECT a, b FROM t');
 SELECT parse_sql('SELECT db.my_func(a), count(b) FROM cat.ns.t1 JOIN t2');
 
+-- batches with and without a final delimiter
+--QUERY-DELIMITER-START
+SELECT parse_sql('select 1; select 2');
+--QUERY-DELIMITER-END
+--QUERY-DELIMITER-START
+SELECT parse_sql('select 1; select 2;');
+--QUERY-DELIMITER-END
+
+-- a failing statement does not prevent later statements from being parsed
+--QUERY-DELIMITER-START
+SELECT parse_sql('SELECT 1; SELEC 2; SELECT 3');
+--QUERY-DELIMITER-END
+
 -- JSON-path access over one shared successful parse result
 SELECT
-  get_json_object(result, '$.statement_identifier') AS statement_identifier,
-  get_json_object(result, '$.source_table_references[0][0]') AS first_table,
-  get_json_object(result, '$.select_list[1].name[0]') AS second_column
+  get_json_object(result, '$[0].statement_identifier') AS statement_identifier,
+  get_json_object(result, '$[0].source_table_references[0][0]') AS first_table,
+  get_json_object(result, '$[0].select_list[1].name[0]') AS second_column
 FROM (SELECT parse_sql('SELECT a, b FROM t') AS result);
 
 -- DML
@@ -93,9 +106,9 @@ SELECT parse_sql('SELEC FROM t');
 
 -- JSON-path access over one shared parse result
 SELECT
-  get_json_object(result, '$.parse_success') AS parse_success,
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[0].parse_success') AS parse_success,
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.queryContext[0].fragment') AS fragment
 FROM (SELECT parse_sql('SELEC FROM t') AS result);
 
 -- full multiline parse-time validation error, including context and location
@@ -107,10 +120,10 @@ SELECT parse_sql(
 
 -- JSON-path access over one shared multiline parse result
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].startIndex') AS start_index
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.line') AS line,
+  get_json_object(result, '$[0].error.position') AS position,
+  get_json_object(result, '$[0].error.queryContext[0].startIndex') AS start_index
 FROM (
   SELECT parse_sql(
 'SELECT *
@@ -143,10 +156,12 @@ SELECT parse_sql(
 -- JSON-path access over one shared scripting parse result
 --QUERY-DELIMITER-START
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[1].start') AS start,
+  get_json_object(result, '$[1].length') AS length,
+  get_json_object(result, '$[1].error.errorClass') AS error_class,
+  get_json_object(result, '$[1].error.line') AS line,
+  get_json_object(result, '$[1].error.position') AS position,
+  get_json_object(result, '$[1].error.queryContext[0].fragment') AS fragment
 FROM (
   SELECT parse_sql(
 'BEGIN
@@ -169,10 +184,10 @@ SELECT parse_sql(
 -- JSON-path access over one shared scripting validation result
 --QUERY-DELIMITER-START
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.line') AS line,
+  get_json_object(result, '$[0].error.position') AS position,
+  get_json_object(result, '$[0].error.queryContext[0].fragment') AS fragment
 FROM (
   SELECT parse_sql(
 'BEGIN
diff --git a/sql/core/src/test/resources/sql-tests/results/parse-sql.sql.out b/sql/core/src/test/resources/sql-tests/results/parse-sql.sql.out
index 986ce7c4ae945..1310f7c30fffa 100644
--- a/sql/core/src/test/resources/sql-tests/results/parse-sql.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/parse-sql.sql.out
@@ -12,7 +12,7 @@ SELECT parse_sql('SELECT a, b FROM t')
 -- !query schema
 struct<parse_sql(SELECT a, b FROM t):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}
+[{"start":1,"length":18,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}]
 
 
 -- !query
@@ -20,14 +20,38 @@ SELECT parse_sql('SELECT db.my_func(a), count(b) FROM cat.ns.t1 JOIN t2')
 -- !query schema
 struct<parse_sql(SELECT db.my_func(a), count(b) FROM cat.ns.t1 JOIN t2):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["cat","ns","t1"],["t2"]],"function_references":[["db","my_func"],["count"]],"select_list":[{"name":[]},{"name":[]}]}
+[{"start":1,"length":53,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["cat","ns","t1"],["t2"]],"function_references":[["db","my_func"],["count"]],"select_list":[{"name":[]},{"name":[]}]}]
+
+
+-- !query
+SELECT parse_sql('select 1; select 2')
+-- !query schema
+struct<parse_sql(select 1; select 2):string>
+-- !query output
+[{"start":1,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]},{"start":11,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}]
+
+
+-- !query
+SELECT parse_sql('select 1; select 2;')
+-- !query schema
+struct<parse_sql(select 1; select 2;):string>
+-- !query output
+[{"start":1,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]},{"start":11,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}]
+
+
+-- !query
+SELECT parse_sql('SELECT 1; SELEC 2; SELECT 3')
+-- !query schema
+struct<parse_sql(SELECT 1; SELEC 2; SELECT 3):string>
+-- !query output
+[{"start":1,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]},{"start":11,"length":7,"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"'SELEC'","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":7,"fragment":"SELEC 2"}],"line":1,"position":0}},{"start":20,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}]
 
 
 -- !query
 SELECT
-  get_json_object(result, '$.statement_identifier') AS statement_identifier,
-  get_json_object(result, '$.source_table_references[0][0]') AS first_table,
-  get_json_object(result, '$.select_list[1].name[0]') AS second_column
+  get_json_object(result, '$[0].statement_identifier') AS statement_identifier,
+  get_json_object(result, '$[0].source_table_references[0][0]') AS first_table,
+  get_json_object(result, '$[0].select_list[1].name[0]') AS second_column
 FROM (SELECT parse_sql('SELECT a, b FROM t') AS result)
 -- !query schema
 struct<statement_identifier:string,first_table:string,second_column:string>
@@ -40,7 +64,7 @@ SELECT parse_sql('INSERT INTO t SELECT 1')
 -- !query schema
 struct<parse_sql(INSERT INTO t SELECT 1):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"INSERT","statement_code":50,"target_table_references":[["t"]],"select_list":[{"name":[]}]}
+[{"start":1,"length":22,"parse_success":true,"statement_identifier":"INSERT","statement_code":50,"target_table_references":[["t"]],"select_list":[{"name":[]}]}]
 
 
 -- !query
@@ -48,7 +72,7 @@ SELECT parse_sql('DELETE FROM t WHERE a = 1')
 -- !query schema
 struct<parse_sql(DELETE FROM t WHERE a = 1):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"DELETE WHERE","statement_code":19,"target_table_references":[["t"]]}
+[{"start":1,"length":25,"parse_success":true,"statement_identifier":"DELETE WHERE","statement_code":19,"target_table_references":[["t"]]}]
 
 
 -- !query
@@ -56,7 +80,7 @@ SELECT parse_sql('UPDATE t SET a = 1 WHERE b = 2')
 -- !query schema
 struct<parse_sql(UPDATE t SET a = 1 WHERE b = 2):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"UPDATE WHERE","statement_code":82,"target_table_references":[["t"]]}
+[{"start":1,"length":30,"parse_success":true,"statement_identifier":"UPDATE WHERE","statement_code":82,"target_table_references":[["t"]]}]
 
 
 -- !query
@@ -64,7 +88,7 @@ SELECT parse_sql('MERGE INTO t USING s ON t.id = s.id WHEN MATCHED THEN DELETE')
 -- !query schema
 struct<parse_sql(MERGE INTO t USING s ON t.id = s.id WHEN MATCHED THEN DELETE):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"MERGE","statement_code":128,"target_table_references":[["t"]],"source_table_references":[["s"]]}
+[{"start":1,"length":60,"parse_success":true,"statement_identifier":"MERGE","statement_code":128,"target_table_references":[["t"]],"source_table_references":[["s"]]}]
 
 
 -- !query
@@ -72,7 +96,7 @@ SELECT parse_sql('CREATE TABLE t (a INT)')
 -- !query schema
 struct<parse_sql(CREATE TABLE t (a INT)):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"CREATE TABLE","statement_code":77,"target_table_references":[["t"]]}
+[{"start":1,"length":22,"parse_success":true,"statement_identifier":"CREATE TABLE","statement_code":77,"target_table_references":[["t"]]}]
 
 
 -- !query
@@ -80,7 +104,7 @@ SELECT parse_sql('CREATE TABLE t AS SELECT 1 AS a')
 -- !query schema
 struct<parse_sql(CREATE TABLE t AS SELECT 1 AS a):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"CREATE TABLE","statement_code":77,"target_table_references":[["t"]],"select_list":[{"name":["a"]}]}
+[{"start":1,"length":31,"parse_success":true,"statement_identifier":"CREATE TABLE","statement_code":77,"target_table_references":[["t"]],"select_list":[{"name":["a"]}]}]
 
 
 -- !query
@@ -88,7 +112,7 @@ SELECT parse_sql('DROP TABLE t')
 -- !query schema
 struct<parse_sql(DROP TABLE t):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"DROP TABLE","statement_code":32,"target_table_references":[["t"]]}
+[{"start":1,"length":12,"parse_success":true,"statement_identifier":"DROP TABLE","statement_code":32,"target_table_references":[["t"]]}]
 
 
 -- !query
@@ -96,7 +120,7 @@ SELECT parse_sql('CACHE TABLE t')
 -- !query schema
 struct<parse_sql(CACHE TABLE t):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"CACHE TABLE","statement_code":-1,"target_table_references":[["t"]]}
+[{"start":1,"length":13,"parse_success":true,"statement_identifier":"CACHE TABLE","statement_code":-1,"target_table_references":[["t"]]}]
 
 
 -- !query
@@ -104,7 +128,7 @@ SELECT parse_sql('TABLE t')
 -- !query schema
 struct<parse_sql(TABLE t):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]]}
+[{"start":1,"length":7,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]]}]
 
 
 -- !query
@@ -112,7 +136,7 @@ SELECT parse_sql('VALUES (1), (2)')
 -- !query schema
 struct<parse_sql(VALUES (1), (2)):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21}
+[{"start":1,"length":15,"parse_success":true,"statement_identifier":"SELECT","statement_code":21}]
 
 
 -- !query
@@ -120,7 +144,7 @@ SELECT parse_sql('CREATE FUNCTION f AS ''x'' USING JAR ''y.jar''')
 -- !query schema
 struct<parse_sql(CREATE FUNCTION f AS 'x' USING JAR 'y.jar'):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"CREATE ROUTINE","statement_code":14}
+[{"start":1,"length":42,"parse_success":true,"statement_identifier":"CREATE ROUTINE","statement_code":14}]
 
 
 -- !query
@@ -128,7 +152,7 @@ SELECT parse_sql('DECLARE VARIABLE x INT')
 -- !query schema
 struct<parse_sql(DECLARE VARIABLE x INT):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"DECLARE VARIABLE","statement_code":-8}
+[{"start":1,"length":22,"parse_success":true,"statement_identifier":"DECLARE VARIABLE","statement_code":-8}]
 
 
 -- !query
@@ -136,7 +160,7 @@ SELECT parse_sql('SELECT * FROM t WHERE a = :foo AND b = ?')
 -- !query schema
 struct<parse_sql(SELECT * FROM t WHERE a = :foo AND b = ?):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["*"]}],"parameter_markers":{"named":["foo"],"unnamed_count":1}}
+[{"start":1,"length":40,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["t"]],"select_list":[{"name":["*"]}],"parameter_markers":{"named":["foo"],"unnamed_count":1}}]
 
 
 -- !query
@@ -144,7 +168,7 @@ SELECT parse_sql('WITH cte AS (SELECT a FROM hidden_base) SELECT a FROM cte')
 -- !query schema
 struct<parse_sql(WITH cte AS (SELECT a FROM hidden_base) SELECT a FROM cte):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["hidden_base"]],"select_list":[{"name":["a"]}]}
+[{"start":1,"length":57,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["hidden_base"]],"select_list":[{"name":["a"]}]}]
 
 
 -- !query
@@ -152,7 +176,7 @@ SELECT parse_sql('SELECT * FROM real_t WHERE EXISTS (WITH real_t AS (SELECT * FR
 -- !query schema
 struct<parse_sql(SELECT * FROM real_t WHERE EXISTS (WITH real_t AS (SELECT * FROM inner_base) SELECT * FROM real_t)):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["inner_base"],["real_t"]],"select_list":[{"name":["*"]}]}
+[{"start":1,"length":98,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["inner_base"],["real_t"]],"select_list":[{"name":["*"]}]}]
 
 
 -- !query
@@ -160,7 +184,7 @@ SELECT parse_sql('WITH a AS (SELECT * FROM b), b AS (SELECT 1 AS x) SELECT * FRO
 -- !query schema
 struct<parse_sql(WITH a AS (SELECT * FROM b), b AS (SELECT 1 AS x) SELECT * FROM a):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["b"]],"select_list":[{"name":["*"]}]}
+[{"start":1,"length":65,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["b"]],"select_list":[{"name":["*"]}]}]
 
 
 -- !query
@@ -168,7 +192,7 @@ SELECT parse_sql('SELECT (SELECT max(v) FROM scalar_src) AS m, t.a FROM outer_t
 -- !query schema
 struct<parse_sql(SELECT (SELECT max(v) FROM scalar_src) AS m, t.a FROM outer_t t WHERE EXISTS (SELECT 1 FROM exists_src e WHERE e.id = t.id)):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["scalar_src"],["exists_src"],["outer_t"]],"function_references":[["max"]],"select_list":[{"name":["m"]},{"name":["t","a"]}]}
+[{"start":1,"length":123,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["scalar_src"],["exists_src"],["outer_t"]],"function_references":[["max"]],"select_list":[{"name":["m"]},{"name":["t","a"]}]}]
 
 
 -- !query
@@ -195,7 +219,7 @@ struct<parse_sql(SELECT coalesce(t.a, 0), sum(abs(t.b)) OVER (
  HAVING count_if(t.b > 0) > 0
  ORDER BY greatest(t.a, 1)):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["scalar_t"],["left_t"],["right_t"]],"function_references":[["greatest"],["count_if"],["coalesce"],["sum"],["abs"],["lower"],["length"],["startswith"],["max"],["range"],["hash"]],"select_list":[{"name":[]},{"name":[]}]}
+[{"start":1,"length":390,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"source_table_references":[["scalar_t"],["left_t"],["right_t"]],"function_references":[["greatest"],["count_if"],["coalesce"],["sum"],["abs"],["lower"],["length"],["startswith"],["max"],["range"],["hash"]],"select_list":[{"name":[]},{"name":[]}]}]
 
 
 -- !query
@@ -224,7 +248,7 @@ struct<parse_sql(MERGE INTO target t
  WHEN NOT MATCHED THEN
    INSERT (id, name) VALUES (s.id, lower(s.name))):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"MERGE","statement_code":128,"target_table_references":[["target"]],"source_table_references":[["source"]],"function_references":[["hash"],["should_update"],["coalesce"],["upper"],["lower"],["normalize_name"],["is_valid"]]}
+[{"start":1,"length":320,"parse_success":true,"statement_identifier":"MERGE","statement_code":128,"target_table_references":[["target"]],"source_table_references":[["source"]],"function_references":[["hash"],["should_update"],["coalesce"],["upper"],["lower"],["normalize_name"],["is_valid"]]}]
 
 
 -- !query
@@ -239,7 +263,7 @@ struct<parse_sql(CREATE TABLE defaults (
    normalized STRING DEFAULT upper('x')
  )):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"CREATE TABLE","statement_code":77,"target_table_references":[["defaults"]],"function_references":[["current_date"],["upper"]]}
+[{"start":1,"length":106,"parse_success":true,"statement_identifier":"CREATE TABLE","statement_code":77,"target_table_references":[["defaults"]],"function_references":[["current_date"],["upper"]]}]
 
 
 -- !query
@@ -247,14 +271,14 @@ SELECT parse_sql('SELEC FROM t')
 -- !query schema
 struct<parse_sql(SELEC FROM t):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"'SELEC'","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":12,"fragment":"SELEC FROM t"}],"line":1,"position":0}}
+[{"start":1,"length":12,"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"'SELEC'","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":12,"fragment":"SELEC FROM t"}],"line":1,"position":0}}]
 
 
 -- !query
 SELECT
-  get_json_object(result, '$.parse_success') AS parse_success,
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[0].parse_success') AS parse_success,
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.queryContext[0].fragment') AS fragment
 FROM (SELECT parse_sql('SELEC FROM t') AS result)
 -- !query schema
 struct<parse_success:string,error_class:string,fragment:string>
@@ -274,15 +298,15 @@ struct<parse_sql(SELECT *
  ORDER BY a
  CLUSTER BY b):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"UNSUPPORTED_FEATURE.COMBINATION_QUERY_RESULT_CLAUSES","messageTemplate":"The feature is not supported: Combination of ORDER BY/SORT BY/DISTRIBUTE BY/CLUSTER BY.","sqlState":"0A000","queryContext":[{"objectType":"","objectName":"","startIndex":19,"stopIndex":42,"fragment":"ORDER BY a\n CLUSTER BY b"}],"line":3,"position":1}}
+[{"start":1,"length":42,"parse_success":false,"error":{"errorClass":"UNSUPPORTED_FEATURE.COMBINATION_QUERY_RESULT_CLAUSES","messageTemplate":"The feature is not supported: Combination of ORDER BY/SORT BY/DISTRIBUTE BY/CLUSTER BY.","sqlState":"0A000","queryContext":[{"objectType":"","objectName":"","startIndex":19,"stopIndex":42,"fragment":"ORDER BY a\n CLUSTER BY b"}],"line":3,"position":1}}]
 
 
 -- !query
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].startIndex') AS start_index
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.line') AS line,
+  get_json_object(result, '$[0].error.position') AS position,
+  get_json_object(result, '$[0].error.queryContext[0].startIndex') AS start_index
 FROM (
   SELECT parse_sql(
 'SELECT *
@@ -301,7 +325,7 @@ SELECT parse_sql('')
 -- !query schema
 struct<parse_sql():string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"PARSE_EMPTY_STATEMENT","messageTemplate":"Syntax error, unexpected empty statement.","sqlState":"42617","line":1,"position":0}}
+[]
 
 
 -- !query
@@ -309,7 +333,7 @@ SELECT parse_sql('USE bad-name')
 -- !query schema
 struct<parse_sql(USE bad-name):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"INVALID_IDENTIFIER","messageTemplate":"The unquoted identifier <ident> is invalid and must be back quoted as: `<ident>`.\nUnquoted identifiers can only contain ASCII letters ('a' - 'z', 'A' - 'Z'), digits ('0' - '9'), and underbar ('_').\nUnquoted identifiers must also not start with a digit.\nDifferent data sources and meta stores may impose additional restrictions on valid identifiers.","sqlState":"42602","messageParameters":{"ident":"bad-name"},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":12,"fragment":"USE bad-name"}],"line":1,"position":7}}
+[{"start":1,"length":12,"parse_success":false,"error":{"errorClass":"INVALID_IDENTIFIER","messageTemplate":"The unquoted identifier <ident> is invalid and must be back quoted as: `<ident>`.\nUnquoted identifiers can only contain ASCII letters ('a' - 'z', 'A' - 'Z'), digits ('0' - '9'), and underbar ('_').\nUnquoted identifiers must also not start with a digit.\nDifferent data sources and meta stores may impose additional restrictions on valid identifiers.","sqlState":"42602","messageParameters":{"ident":"bad-name"},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":12,"fragment":"USE bad-name"}],"line":1,"position":7}}]
 
 
 -- !query
@@ -317,7 +341,7 @@ SELECT parse_sql('WITH c AS (SELECT 1), c AS (SELECT 2) SELECT * FROM c')
 -- !query schema
 struct<parse_sql(WITH c AS (SELECT 1), c AS (SELECT 2) SELECT * FROM c):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"DUPLICATED_CTE_NAMES","messageTemplate":"CTE definition can't have duplicate names: <duplicateNames>.","sqlState":"42602","messageParameters":{"duplicateNames":"`c`"},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":53,"fragment":"WITH c AS (SELECT 1), c AS (SELECT 2) SELECT * FROM c"}],"line":1,"position":0}}
+[{"start":1,"length":53,"parse_success":false,"error":{"errorClass":"DUPLICATED_CTE_NAMES","messageTemplate":"CTE definition can't have duplicate names: <duplicateNames>.","sqlState":"42602","messageParameters":{"duplicateNames":"`c`"},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":53,"fragment":"WITH c AS (SELECT 1), c AS (SELECT 2) SELECT * FROM c"}],"line":1,"position":0}}]
 
 
 -- !query
@@ -325,7 +349,7 @@ SELECT parse_sql('MERGE INTO target USING source ON target.id = source.id')
 -- !query schema
 struct<parse_sql(MERGE INTO target USING source ON target.id = source.id):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"MERGE_WITHOUT_WHEN","messageTemplate":"There must be at least one WHEN clause in a MERGE statement.","sqlState":"42601","queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":55,"fragment":"MERGE INTO target USING source ON target.id = source.id"}],"line":1,"position":0}}
+[{"start":1,"length":55,"parse_success":false,"error":{"errorClass":"MERGE_WITHOUT_WHEN","messageTemplate":"There must be at least one WHEN clause in a MERGE statement.","sqlState":"42601","queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":55,"fragment":"MERGE INTO target USING source ON target.id = source.id"}],"line":1,"position":0}}]
 
 
 -- !query
@@ -333,7 +357,7 @@ SELECT parse_sql('EXPLAIN SELECT 1')
 -- !query schema
 struct<parse_sql(EXPLAIN SELECT 1):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"EXPLAIN","statement_code":-23,"select_list":[{"name":[]}]}
+[{"start":1,"length":16,"parse_success":true,"statement_identifier":"EXPLAIN","statement_code":-23,"select_list":[{"name":[]}]}]
 
 
 -- !query
@@ -341,7 +365,7 @@ SELECT parse_sql('SET spark.sql.adaptive.enabled=true')
 -- !query schema
 struct<parse_sql(SET spark.sql.adaptive.enabled=true):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"SET","statement_code":-24}
+[{"start":1,"length":35,"parse_success":true,"statement_identifier":"SET","statement_code":-24}]
 
 
 -- !query
@@ -349,7 +373,7 @@ SELECT parse_sql('ADD JAR /tmp/x.jar')
 -- !query schema
 struct<parse_sql(ADD JAR /tmp/x.jar):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"ADD JAR","statement_code":-26}
+[{"start":1,"length":18,"parse_success":true,"statement_identifier":"ADD JAR","statement_code":-26}]
 
 
 -- !query
@@ -357,7 +381,7 @@ SELECT parse_sql('CREATE VIEW v AS SELECT a, b FROM t')
 -- !query schema
 struct<parse_sql(CREATE VIEW v AS SELECT a, b FROM t):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"CREATE VIEW","statement_code":84,"target_table_references":[["v"]],"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}
+[{"start":1,"length":35,"parse_success":true,"statement_identifier":"CREATE VIEW","statement_code":84,"target_table_references":[["v"]],"source_table_references":[["t"]],"select_list":[{"name":["a"]},{"name":["b"]}]}]
 
 
 -- !query
@@ -365,7 +389,7 @@ SELECT parse_sql('SELECT 1 AS IDENTIFIER(''alias.field'')')
 -- !query schema
 struct<parse_sql(SELECT 1 AS IDENTIFIER('alias.field')):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"IDENTIFIER_TOO_MANY_NAME_PARTS","messageTemplate":"<identifier> is not a valid identifier as it has more than <limit> name parts.","sqlState":"42601","messageParameters":{"identifier":"`alias`.`field`","limit":"1"},"queryContext":[{"objectType":"","objectName":"","startIndex":8,"stopIndex":37,"fragment":"1 AS IDENTIFIER('alias.field')"}],"line":1,"position":12}}
+[{"start":1,"length":37,"parse_success":false,"error":{"errorClass":"IDENTIFIER_TOO_MANY_NAME_PARTS","messageTemplate":"<identifier> is not a valid identifier as it has more than <limit> name parts.","sqlState":"42601","messageParameters":{"identifier":"`alias`.`field`","limit":"1"},"queryContext":[{"objectType":"","objectName":"","startIndex":8,"stopIndex":37,"fragment":"1 AS IDENTIFIER('alias.field')"}],"line":1,"position":12}}]
 
 
 -- !query
@@ -373,7 +397,7 @@ SELECT parse_sql('SELECT DATE ''not-a-date''')
 -- !query schema
 struct<parse_sql(SELECT DATE 'not-a-date'):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"INVALID_TYPED_LITERAL","messageTemplate":"The value of the typed literal <valueType> is invalid: <value>.","sqlState":"42604","messageParameters":{"value":"'not-a-date'","valueType":"\"DATE\""},"queryContext":[{"objectType":"","objectName":"","startIndex":8,"stopIndex":24,"fragment":"DATE 'not-a-date'"}],"line":1,"position":7}}
+[{"start":1,"length":24,"parse_success":false,"error":{"errorClass":"INVALID_TYPED_LITERAL","messageTemplate":"The value of the typed literal <valueType> is invalid: <value>.","sqlState":"42604","messageParameters":{"value":"'not-a-date'","valueType":"\"DATE\""},"queryContext":[{"objectType":"","objectName":"","startIndex":8,"stopIndex":24,"fragment":"DATE 'not-a-date'"}],"line":1,"position":7}}]
 
 
 -- !query
@@ -388,15 +412,17 @@ struct<parse_sql(BEGIN
    SELEC 2;
  END):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"'2'","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":35,"fragment":"BEGIN\n   SELECT 1;\n   SELEC 2;\n END"}],"line":3,"position":9}}
+[{"start":1,"length":17,"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"end of input","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":17,"fragment":"BEGIN\n   SELECT 1"}],"line":2,"position":11}},{"start":23,"length":7,"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"'SELEC'","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":7,"fragment":"SELEC 2"}],"line":1,"position":0}},{"start":33,"length":3,"parse_success":false,"error":{"errorClass":"PARSE_SYNTAX_ERROR","messageTemplate":"Syntax error at or near <error><hint>.","sqlState":"42601","messageParameters":{"error":"'END'","hint":""},"queryContext":[{"objectType":"","objectName":"","startIndex":1,"stopIndex":3,"fragment":"END"}],"line":1,"position":0}}]
 
 
 -- !query
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[1].start') AS start,
+  get_json_object(result, '$[1].length') AS length,
+  get_json_object(result, '$[1].error.errorClass') AS error_class,
+  get_json_object(result, '$[1].error.line') AS line,
+  get_json_object(result, '$[1].error.position') AS position,
+  get_json_object(result, '$[1].error.queryContext[0].fragment') AS fragment
 FROM (
   SELECT parse_sql(
 'BEGIN
@@ -405,12 +431,9 @@ FROM (
  END') AS result
 )
 -- !query schema
-struct<error_class:string,line:string,position:string,fragment:string>
+struct<start:string,length:string,error_class:string,line:string,position:string,fragment:string>
 -- !query output
-PARSE_SYNTAX_ERROR	3	9	BEGIN
-   SELECT 1;
-   SELEC 2;
- END
+23	7	PARSE_SYNTAX_ERROR	1	0	SELEC 2
 
 
 -- !query
@@ -427,15 +450,15 @@ struct<parse_sql(BEGIN
    END lbl_end;
  END):string>
 -- !query output
-{"parse_success":false,"error":{"errorClass":"LABELS_MISMATCH","messageTemplate":"Begin label <beginLabel> does not match the end label <endLabel>.","sqlState":"42K0L","messageParameters":{"beginLabel":"`lbl_begin`","endLabel":"`lbl_end`"},"queryContext":[{"objectType":"","objectName":"","startIndex":10,"stopIndex":19,"fragment":"lbl_begin:"}],"line":2,"position":3}}
+[{"start":1,"length":61,"parse_success":false,"error":{"errorClass":"LABELS_MISMATCH","messageTemplate":"Begin label <beginLabel> does not match the end label <endLabel>.","sqlState":"42K0L","messageParameters":{"beginLabel":"`lbl_begin`","endLabel":"`lbl_end`"},"queryContext":[{"objectType":"","objectName":"","startIndex":10,"stopIndex":19,"fragment":"lbl_begin:"}],"line":2,"position":3}}]
 
 
 -- !query
 SELECT
-  get_json_object(result, '$.error.errorClass') AS error_class,
-  get_json_object(result, '$.error.line') AS line,
-  get_json_object(result, '$.error.position') AS position,
-  get_json_object(result, '$.error.queryContext[0].fragment') AS fragment
+  get_json_object(result, '$[0].error.errorClass') AS error_class,
+  get_json_object(result, '$[0].error.line') AS line,
+  get_json_object(result, '$[0].error.position') AS position,
+  get_json_object(result, '$[0].error.queryContext[0].fragment') AS fragment
 FROM (
   SELECT parse_sql(
 'BEGIN
@@ -459,9 +482,9 @@ AS t(sql_text)
 -- !query schema
 struct<sql_text:string,parse_sql(sql_text):string>
 -- !query output
-CACHE TABLE t	{"parse_success":true,"statement_identifier":"CACHE TABLE","statement_code":-1,"target_table_references":[["t"]]}
-INSERT INTO t SELECT 1	{"parse_success":true,"statement_identifier":"INSERT","statement_code":50,"target_table_references":[["t"]],"select_list":[{"name":[]}]}
-SELECT 1	{"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}
+CACHE TABLE t	[{"start":1,"length":13,"parse_success":true,"statement_identifier":"CACHE TABLE","statement_code":-1,"target_table_references":[["t"]]}]
+INSERT INTO t SELECT 1	[{"start":1,"length":22,"parse_success":true,"statement_identifier":"INSERT","statement_code":50,"target_table_references":[["t"]],"select_list":[{"name":[]}]}]
+SELECT 1	[{"start":1,"length":8,"parse_success":true,"statement_identifier":"SELECT","statement_code":21,"select_list":[{"name":[]}]}]
 
 
 -- !query
@@ -469,7 +492,7 @@ SELECT parse_sql('BEGIN SELECT 1; END')
 -- !query schema
 struct<parse_sql(BEGIN SELECT 1; END):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22}
+[{"start":1,"length":19,"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22}]
 
 
 -- !query
@@ -477,7 +500,7 @@ SELECT parse_sql('BEGIN SELECT count(a) FROM script_t WHERE c = :p; END')
 -- !query schema
 struct<parse_sql(BEGIN SELECT count(a) FROM script_t WHERE c = :p; END):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"source_table_references":[["script_t"]],"function_references":[["count"]],"parameter_markers":{"named":["p"]}}
+[{"start":1,"length":53,"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"source_table_references":[["script_t"]],"function_references":[["count"]],"parameter_markers":{"named":["p"]}}]
 
 
 -- !query
@@ -485,7 +508,7 @@ SELECT parse_sql('BEGIN SELECT * FROM t WHERE a = ?; END')
 -- !query schema
 struct<parse_sql(BEGIN SELECT * FROM t WHERE a = ?; END):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"source_table_references":[["t"]],"parameter_markers":{"unnamed_count":1}}
+[{"start":1,"length":38,"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"source_table_references":[["t"]],"parameter_markers":{"unnamed_count":1}}]
 
 
 -- !query
@@ -493,7 +516,7 @@ SELECT parse_sql('BEGIN IF (SELECT flag FROM gate) THEN INSERT INTO dest SELECT
 -- !query schema
 struct<parse_sql(BEGIN IF (SELECT flag FROM gate) THEN INSERT INTO dest SELECT * FROM src_if; ELSE DELETE FROM src_else; END IF; END):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"target_table_references":[["dest"],["src_else"]],"source_table_references":[["gate"],["src_if"]]}
+[{"start":1,"length":115,"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"target_table_references":[["dest"],["src_else"]],"source_table_references":[["gate"],["src_if"]]}]
 
 
 -- !query
@@ -501,7 +524,7 @@ SELECT parse_sql('BEGIN DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN INSERT INTO
 -- !query schema
 struct<parse_sql(BEGIN DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN INSERT INTO err_log SELECT * FROM failing_row; END; SELECT a FROM main_t; END):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"target_table_references":[["err_log"]],"source_table_references":[["failing_row"],["main_t"]]}
+[{"start":1,"length":127,"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"target_table_references":[["err_log"]],"source_table_references":[["failing_row"],["main_t"]]}]
 
 
 -- !query
@@ -568,4 +591,4 @@ struct<parse_sql(BEGIN
    END FOR;
  END):string>
 -- !query output
-{"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"target_table_references":[["error_log"],["output_names"],["update_target"],["delete_target"]],"source_table_references":[["error_source"],["input_names"],["control_flags"],["update_source"],["delete_source"],["loop_source"],["loop_body"]],"function_references":[["format_string"],["normalize_name"],["is_valid"],["upper"],["enabled"],["coalesce"],["should_update"],["max"],["expired"],["ready"],["audit"],["count"]]}
+[{"start":1,"length":783,"parse_success":true,"statement_identifier":"BEGIN END","statement_code":-22,"target_table_references":[["error_log"],["output_names"],["update_target"],["delete_target"]],"source_table_references":[["error_source"],["input_names"],["control_flags"],["update_source"],["delete_source"],["loop_source"],["loop_body"]],"function_references":[["format_string"],["normalize_name"],["is_valid"],["upper"],["enabled"],["coalesce"],["should_update"],["max"],["expired"],["ready"],["audit"],["count"]]}]
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlSuite.scala
index f44a409a0d641..0322c01c4dbca 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlSuite.scala
@@ -36,6 +36,16 @@ class ParseSqlSuite extends SparkFunSuite with ExpressionEvalHelper with SQLHelp
     parse(result)
   }
 
+  private def evalStatements(sql: String): List[JObject] = evalJson(sql) match {
+    case JArray(values) => values.map(_.asInstanceOf[JObject])
+    case other => fail(s"expected JSON array, got: $other")
+  }
+
+  private def evalStatement(sql: String): JObject = evalStatements(sql) match {
+    case value :: Nil => value
+    case other => fail(s"expected one statement, got ${other.size}: $other")
+  }
+
   test("parse_sql is disabled by default") {
     assert(!SQLConf.get.parseSqlEnabled)
     checkError(
@@ -58,7 +68,9 @@ class ParseSqlSuite extends SparkFunSuite with ExpressionEvalHelper with SQLHelp
 
   test("parse_sql returns JSON for a valid SELECT") {
     withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
-      val j = evalJson("SELECT 1 AS a")
+      val j = evalStatement("SELECT 1 AS a")
+      assert(j \ "start" === JInt(1))
+      assert(j \ "length" === JInt(13))
       assert(j \ "parse_success" === JBool(true))
       assert(j \ "statement_identifier" === JString("SELECT"))
       assert(j \ "statement_code" === JInt(21))
@@ -73,7 +85,7 @@ class ParseSqlSuite extends SparkFunSuite with ExpressionEvalHelper with SQLHelp
 
   test("parse_sql does not throw on syntax error") {
     withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
-      val j = evalJson("NOT A STATEMENT !!!")
+      val j = evalStatement("NOT A STATEMENT !!!")
       assert(j \ "parse_success" === JBool(false))
       assert(j \ "error" \ "errorClass" === JString("PARSE_SYNTAX_ERROR"))
     }
@@ -83,8 +95,17 @@ class ParseSqlSuite extends SparkFunSuite with ExpressionEvalHelper with SQLHelp
     withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
       val expr = ParseSql(Literal("INSERT INTO t SELECT 1"))
       assert(expr.isInstanceOf[CodegenFallback])
-      val j = evalJson("INSERT INTO t SELECT 1")
+      val j = evalStatement("INSERT INTO t SELECT 1")
       assert(j \ "statement_identifier" === JString("INSERT"))
     }
   }
+
+  test("parse_sql returns a JSON array for a SQL batch") {
+    withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+      val statements = evalStatements("SELECT 1; SELECT 2")
+      assert(statements.size === 2)
+      assert(statements.map(_ \ "start") === Seq(JInt(1), JInt(11)))
+      assert(statements.map(_ \ "length") === Seq(JInt(8), JInt(8)))
+    }
+  }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala
index 397584cb0c8db..a04c8b7753854 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala
@@ -29,8 +29,16 @@ import org.apache.spark.sql.internal.SQLConf
  */
 class ParseSqlResultSuite extends SparkFunSuite {
 
-  private def obj(sql: String): JObject =
-    parse(ParseSqlResult.fromSql(sql)).asInstanceOf[JObject]
+  private def objs(sql: String): List[JObject] =
+    parse(ParseSqlResult.fromSql(sql)) match {
+      case JArray(values) => values.map(_.asInstanceOf[JObject])
+      case other => fail(s"expected JSON array, got: $other")
+    }
+
+  private def obj(sql: String): JObject = objs(sql) match {
+    case value :: Nil => value
+    case other => fail(s"expected one statement, got ${other.size}: $other")
+  }
 
   private def tableRefs(sql: String, field: String): Set[Seq[String]] =
     obj(sql) \ field match {
@@ -67,6 +75,77 @@ class ParseSqlResultSuite extends SparkFunSuite {
     assert(SqlStatementCodes.CreateMetricViewStmt.statementCode === -37)
   }
 
+  test("batches return one statement object per statement with source spans") {
+    Seq("select 1; select 2", "select 1; select 2;").foreach { sql =>
+      val statements = objs(sql)
+      assert(statements.size === 2)
+      assert(statements.map(_ \ "start") === Seq(JInt(1), JInt(11)))
+      assert(statements.map(_ \ "length") === Seq(JInt(8), JInt(8)))
+      assert(statements.map(_ \ "parse_success") === Seq(JBool(true), JBool(true)))
+      assert(statements.map(_ \ "statement_identifier") ===
+        Seq(JString("SELECT"), JString("SELECT")))
+    }
+
+    val statements = objs("  SELECT 1 ;\n SELECT 2;  ")
+    assert(statements.map(_ \ "start") === Seq(JInt(3), JInt(15)))
+    assert(statements.map(_ \ "length") === Seq(JInt(8), JInt(8)))
+
+    val sqlWithDroppedComment = "SELECT 1; /* SELECT 2 */; SELECT 2"
+    val commentStatements = objs(sqlWithDroppedComment)
+    assert(commentStatements.map(_ \ "start") ===
+      Seq(JInt(1), JInt(sqlWithDroppedComment.lastIndexOf("SELECT 2") + 1)))
+
+    val emoji = new String(Character.toChars(0x1F600))
+    val unicodeSql = s"SELECT '$emoji$emoji'; SELECT 2;"
+    val unicodeStatements = objs(unicodeSql)
+    val spans = unicodeStatements.map { statement =>
+      val JInt(start) = statement \ "start"
+      val JInt(length) = statement \ "length"
+      (start.toInt, length.toInt)
+    }
+    assert(spans === Seq((1, 13), (16, 8)))
+    assert(spans.map { case (start, length) =>
+      unicodeSql.substring(start - 1, start - 1 + length)
+    } === Seq(s"SELECT '$emoji$emoji'", "SELECT 2"))
+
+    val nbsp = 0xA0.toChar
+    val nbspSql = s"${nbsp}SELECT 1$nbsp;"
+    val nbspStmt = objs(nbspSql).head
+    val JInt(nbspStart) = nbspStmt \ "start"
+    val JInt(nbspLength) = nbspStmt \ "length"
+    assert((nbspStart.toInt, nbspLength.toInt) === (2, 8))
+    assert(nbspSql.substring(nbspStart.toInt - 1, nbspStart.toInt - 1 + nbspLength.toInt) ===
+      "SELECT 1")
+  }
+
+  test("batch errors are isolated and preserve statement order") {
+    val statements = objs("SELECT 1; SELEC 2; SELECT 3")
+    assert(statements.map(_ \ "start") === Seq(JInt(1), JInt(11), JInt(20)))
+    assert(statements.map(_ \ "length") === Seq(JInt(8), JInt(7), JInt(8)))
+    assert(statements.map(_ \ "parse_success") ===
+      Seq(JBool(true), JBool(false), JBool(true)))
+    assert(statements(1) \ "error" \ "errorClass" === JString("PARSE_SYNTAX_ERROR"))
+  }
+
+  test("SQL scripts remain one statement in a batch") {
+    val statements = objs("BEGIN SELECT 1; SELECT 2; END; SELECT 3")
+    assert(statements.size === 2)
+    assert(statements.map(_ \ "start") === Seq(JInt(1), JInt(32)))
+    assert(statements.map(_ \ "length") === Seq(JInt(29), JInt(8)))
+    assert(statements.map(_ \ "statement_identifier") ===
+      Seq(JString("BEGIN END"), JString("SELECT")))
+  }
+
+  test("empty and closed-comment-only batches contain no statements") {
+    Seq("", "  ", ";;", "-- comment", "/* closed */").foreach { sql =>
+      assert(objs(sql).isEmpty, sql)
+    }
+
+    val unclosed = objs("/* unclosed")
+    assert(unclosed.size === 1)
+    assert(unclosed.head \ "parse_success" === JBool(false))
+  }
+
   test("TABLE and VALUES classify as SELECT") {
     val table = obj("TABLE t")
     assert(table \ "statement_identifier" === JString("SELECT"))