cloud-fan commented on code in PR #58530:
URL: https://github.com/apache/spark/pull/58530#discussion_r3991389357
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala:
##########
@@ -399,7 +548,48 @@ object SqlStatementSplitter {
parser.single_character_pipe_operator_enabled =
conf.singleCharacterPipeOperatorEnabled
parser.removeErrorListeners()
- parser.setErrorHandler(new BailErrorStrategy)
+ if (bailOnError) {
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962199863","thread_id":"inline:3962199863","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala:
##########
@@ -158,6 +190,36 @@ object SqlStatementSplitter {
// interpretation (e.g. `double_quoted_identifiers`).
val conf = SqlApiConf.get
+ def appendToken(token: Token): Unit = {
+ if (buffer.isEmpty) {
+ // CodePointCharStream token offsets count Unicode code points, while
+ // String offsets and lengths count UTF-16 code units.
+ bufferStart = sqlText.offsetByCodePoints(0, token.getStartIndex)
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3940933503","thread_id":"inline:3940933503","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -27,34 +27,37 @@ 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
@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 start
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3940933500","thread_id":"inline:3940933500","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala:
##########
@@ -285,8 +370,155 @@ 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)
+ }
+
+ /**
+ * Returns the delimiter-array index and ending token index of a real outer
END for a malformed
+ * compound statement. Error recovery may repair the body, but a missing END
is synthetic and
+ * has token index -1.
+ */
+ private def findMalformedCompoundEnd(
+ sqlText: String,
+ toUtf16: Array[Int],
+ stream: CommonTokenStream,
+ startIdx: Int,
+ delimiterPositions: Array[Int],
+ fromDelimiter: Int,
+ validationPreprocess: String => String,
+ conf: SqlApiConf): Option[(Int, Int)] = {
+ if (stream.get(startIdx).getType != SqlBaseLexer.BEGIN) {
+ return None
+ }
+
+ var delimiter = fromDelimiter
+ while (delimiter <= delimiterPositions.length) {
+ val endIdx = if (delimiter < delimiterPositions.length) {
+ delimiterPositions(delimiter)
+ } else {
+ stream.size() - 1
+ }
+ val firstTok = stream.get(startIdx)
+ val lastTok = stream.get(endIdx)
+ val regionStart = toUtf16(firstTok.getStartIndex)
+ val regionEnd = if (lastTok.getType == Token.EOF) {
+ sqlText.length
+ } else {
+ toUtf16(lastTok.getStopIndex + 1)
+ }
+ val candidate = validationPreprocess(sqlText.substring(regionStart,
regionEnd))
+ val lexer = new SqlBaseLexer(
+ new UpperCaseCharStream(CharStreams.fromString(candidate)))
+ lexer.removeErrorListeners()
+ val tokens = new CommonTokenStream(lexer)
+ tokens.fill()
+ val parser = new SqlBaseParser(tokens)
+ configureSplitterParser(parser, conf, bailOnError = false)
+ parser.getInterpreter.setPredictionMode(PredictionMode.LL)
+ try {
+ val context = parser.singleCompoundStatement()
+ val end = context.END()
+ if (end != null && isOuterCompoundEnd(tokens, end.getSymbol)) {
+ return Some((delimiter, endIdx))
+ }
+ } catch {
+ case _: StackOverflowError => return None
+ }
+ delimiter += 1
+ }
+ None
+ }
+
+ /** Returns true only when a real recovered END closes the candidate's outer
BEGIN. */
+ private def isOuterCompoundEnd(tokens: CommonTokenStream, recoveredEnd:
Token): Boolean = {
+ if (recoveredEnd.getTokenIndex < 0) return false
+ val suffixEnd = trailingEndToken(tokens)
+ suffixEnd != null && closesOuterBegin(tokens, suffixEnd)
+ }
+
+ private def closesOuterBegin(tokens: CommonTokenStream, suffixEnd: Token):
Boolean = {
+ var depth = 0
+ var index = 0
+ val limit = suffixEnd.getTokenIndex
+ while (index <= limit) {
+ val token = tokens.get(index)
+ if (token.getChannel != Token.HIDDEN_CHANNEL) {
+ token.getType match {
+ case SqlBaseLexer.BEGIN =>
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3971952378","thread_id":"inline:3971952378","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala:
##########
@@ -285,8 +370,67 @@ 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)
+ }
+
+ /**
+ * Returns the delimiter-array index and ending token index of a real outer
END for a malformed
+ * compound statement. Error recovery may repair the body, but a missing END
is synthetic and
+ * has token index -1.
+ */
+ private def findMalformedCompoundEnd(
+ sqlText: String,
+ toUtf16: Array[Int],
+ stream: CommonTokenStream,
+ startIdx: Int,
+ delimiterPositions: Array[Int],
+ fromDelimiter: Int,
+ validationPreprocess: String => String,
+ conf: SqlApiConf): Option[(Int, Int)] = {
+ if (stream.get(startIdx).getType != SqlBaseLexer.BEGIN) {
+ return None
+ }
+
+ var delimiter = fromDelimiter
+ while (delimiter <= delimiterPositions.length) {
+ val endIdx = if (delimiter < delimiterPositions.length) {
+ delimiterPositions(delimiter)
+ } else {
+ stream.size() - 1
+ }
+ val firstTok = stream.get(startIdx)
+ val lastTok = stream.get(endIdx)
+ val regionStart = toUtf16(firstTok.getStartIndex)
+ val regionEnd = if (lastTok.getType == Token.EOF) {
+ sqlText.length
+ } else {
+ toUtf16(lastTok.getStopIndex + 1)
+ }
+ val candidate = validationPreprocess(sqlText.substring(regionStart,
regionEnd))
+ val lexer = new SqlBaseLexer(
+ new UpperCaseCharStream(CharStreams.fromString(candidate)))
+ lexer.removeErrorListeners()
+ val tokens = new CommonTokenStream(lexer)
+ tokens.fill()
+ val parser = new SqlBaseParser(tokens)
+ configureSplitterParser(parser, conf, bailOnError = false)
+ parser.getInterpreter.setPredictionMode(PredictionMode.LL)
+ try {
+ val context = parser.singleCompoundStatement()
+ val end = context.END()
+ if (end != null && end.getSymbol.getTokenIndex >= 0 && tokens.LA(1) ==
Token.EOF) {
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962199855","thread_id":"inline:3962199855","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala:
##########
@@ -158,6 +190,36 @@ object SqlStatementSplitter {
// interpretation (e.g. `double_quoted_identifiers`).
val conf = SqlApiConf.get
+ def appendToken(token: Token): Unit = {
+ if (buffer.isEmpty) {
+ // CodePointCharStream token offsets count Unicode code points, while
+ // String offsets and lengths count UTF-16 code units.
+ bufferStart = sqlText.offsetByCodePoints(0, 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 statement = raw.trim
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3940933506","thread_id":"inline:3940933506","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementSplitter.scala:
##########
@@ -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
+= _)
Review Comment:
Confirmed in the pinned head that this exact issue is resolved. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3945872256","thread_id":"inline:3945872256","verdict_sha256":"1c873ce5350b0afbb843eeabb7da0c9a30fa6a9bfbebf518124010218135d0ad"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
