srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4429938789
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -5070,6 +5070,16 @@
"expects a positive integer, but got <actual>."
]
},
+ "PARSE_SQL_CONTEXT" : {
+ "message" : [
+ "expects a valid catalog, schema, or SQL path, but got
<invalidValue>."
Review Comment:
PARSE_SQL_CONTEXT is removed with the context arguments.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -93,22 +194,114 @@ case class ParseSql(child: Expression)
"configKey" -> SQLConf.PARSE_SQL_ENABLED.key,
"configValue" -> "true"))
}
- super.checkInputDataTypes()
+ val typeCheck = super.checkInputDataTypes()
+ if (!typeCheck.isSuccess) {
+ typeCheck
+ } else {
+ if (!mode.foldable) {
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> toSQLId("mode"),
+ "inputType" -> toSQLType(mode.dataType),
+ "inputExpr" -> toSQLExpr(mode)))
+ } else {
+ val parsedMode = Option(mode.eval()).map { value =>
+ val modeStr = value.asInstanceOf[UTF8String].toString
+ ParseSqlMode.parse(modeStr).getOrElse {
+ throw QueryCompilationErrors.invalidParseSqlMode(mode)
+ }
+ }
+ // ANALYZE needs the session, which only the driver has: every argument
+ // must fold, or be current_catalog() / current_schema() /
current_path()
+ // / current_user(), so the optimizer can replace them with literals.
+ def constant(e: Expression): Boolean = e.foldable || (e match {
+ case _: CurrentDatabase | _: CurrentCatalog | _: CurrentPath | _:
CurrentUser =>
+ true
+ case _ => false
+ })
+ val nonFoldable = ParseSql.parameterNames.zip(children).find(c =>
!constant(c._2))
+ (parsedMode, nonFoldable) match {
+ case (Some(ParseSqlMode.Analyze), Some((name, child))) =>
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> toSQLId(name),
+ "inputType" -> toSQLType(child.dataType),
+ "inputExpr" -> toSQLExpr(child)))
+ case _ => TypeCheckResult.TypeCheckSuccess
+ }
+ }
+ }
}
- override def nullSafeEval(input: Any): Any = {
- val sql = input.asInstanceOf[UTF8String].toString
- UTF8String.fromString(ParseSqlResult.fromSql(sql))
+ override def eval(input: InternalRow): Any = {
+ val sqlInput = sqlStmt.eval(input)
+ if (sqlInput == null) {
+ return null
+ }
+ val modeInput = mode.eval(input)
+ if (modeInput == null) {
+ return null
+ }
+ val sqlText = sqlInput.asInstanceOf[UTF8String].toString
+ val modeStr = modeInput.asInstanceOf[UTF8String].toString
+ val parsedMode = ParseSqlMode.parse(modeStr).getOrElse {
+ throw QueryExecutionErrors.invalidParseSqlMode(modeStr)
+ }
+ val context = if (parsedMode == ParseSqlMode.Analyze) {
+ ParseSqlAnalyzeContext(
+ catalog = utf8Option(currentCatalog.eval(input)),
+ schema = utf8Option(currentSchema.eval(input)),
+ path = utf8Option(currentPath.eval(input)))
+ } else {
+ ParseSqlAnalyzeContext.empty
+ }
+ UTF8String.fromString(ParseSqlResult.fromSql(sqlText, parsedMode, context))
}
- override protected def withNewChildInternal(newChild: Expression): ParseSql =
- copy(child = newChild)
+ private def utf8Option(value: Any): Option[String] =
+ Option(value).map(_.asInstanceOf[UTF8String].toString)
+
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): ParseSql =
+ copy(
+ sqlStmt = newChildren(0),
+ mode = newChildren(1),
+ currentCatalog = newChildren(2),
+ currentSchema = newChildren(3),
+ currentPath = newChildren(4))
}
-object ParseSql {
+object ParseSql extends ExpressionBuilder {
+ def apply(sqlStmt: Expression): ParseSql = new ParseSql(sqlStmt)
+
+ def apply(sqlStmt: Expression, mode: Expression): ParseSql = new
ParseSql(sqlStmt, mode)
+
+ /** Parameter names in argument order. */
+ val parameterNames: Seq[String] =
+ Seq("sqlStmt", "mode", "current_catalog", "current_schema", "current_path")
+
+ // All five parameters are named-invocation eligible, including required
`sqlStmt`.
+ override def functionSignature: Option[FunctionSignature] =
Some(FunctionSignature(Seq(
+ InputParameter("sqlStmt"),
Review Comment:
Keeping sqlStmt to match the existing ExpressionBuilder named-parameter
style (sqlStmt => ...). Can rename to sql_stmt if you prefer snake_case for SQL
named args.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/InvokeProcedures.scala:
##########
@@ -83,3 +87,18 @@ class InvokeProcedures(session: SparkSession) extends
Rule[LogicalPlan] {
new GenericInternalRow(values)
}
}
+
+object InvokeProcedures {
+ private val suppressed: ThreadLocal[Boolean] = ThreadLocal.withInitial(() =>
false)
+
+ /**
+ * Run `f` with CALL left unexecuted by analysis on this thread, including
+ * CALL inside EXECUTE IMMEDIATE. For callers that analyze statements without
+ * running them.
+ */
+ def withoutInvocation[T](f: => T): T = {
Review Comment:
withoutInvocation is now private[sql].
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,70 +122,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, spans, context)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false).
+ * After a context override failure, references stay unqualified parse names
+ * rather than the invoker or the attempted context.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ successJson(plan, None, Some(e), Some(session), span, qualify
= false)
+ }.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 in the original batch. */
+ private case class StatementSpan(start: Int, length: Int)
+
+ /**
+ * Code-point `start` for each statement in one forward pass over `batch`.
+ * Statements are in source order, so prefix counts are not repeated.
+ */
+ private def statementSpans(
+ batch: String,
+ statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = {
+ var lastUtf16 = 0
+ var lastCp = 0
+ statements.map { segment =>
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ if (utf16 >= lastUtf16) {
+ lastCp += batch.codePointCount(lastUtf16, utf16)
+ lastUtf16 = utf16
+ } else {
+ lastCp = batch.codePointCount(0, utf16)
+ lastUtf16 = utf16
+ }
+ StatementSpan(lastCp + 1, charLength(segment))
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): JObject = {
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ segment: PositionedSqlStatement,
+ span: StatementSpan): 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, span))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, span))
+ }
+ }
+
+ /** Analyze `plan` without executing it. Scripts never reach QueryExecution.
*/
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ plan match {
+ case _: CompoundBody =>
+ (None, Some(QueryExecutionErrors.parseSqlAnalyzeUnsupportedScript()))
+ case _ =>
+ try {
+ val analyzed =
+ session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
(None, Some(e))
+ }
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ * Restore writes back the stored catalog name, namespace and PATH flag
+ * without re-validating, and unsets PATH when it was unset.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.storedCurrentCatalogName
+ val prevNamespace = mgr.storedCurrentNamespace
+ val prevPath = mgr.storedSessionPathEntries
+ val pathEnabledWasSet = conf.contains(SQLConf.PATH_ENABLED.key)
+ val prevPathEnabled =
+ if (pathEnabledWasSet) Some(conf.getConf(SQLConf.PATH_ENABLED)) else
None
+ try {
+ context.catalog.foreach { catalog =>
+ asContextError("current_catalog", catalog) {
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ }
+ context.schema.foreach { schema =>
+ asContextError("current_schema", 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 =>
+ asContextError("current_path", path) {
+ if (!conf.getConf(SQLConf.PATH_ENABLED)) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ }
+ f
+ } finally {
+ try {
+ mgr.restoreStoredCurrentCatalogName(prevCatalog)
+ mgr.restoreStoredCurrentNamespace(prevNamespace)
+ } finally {
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ prevPathEnabled match {
+ case Some(value) => conf.setConf(SQLConf.PATH_ENABLED, value)
+ case None => conf.unsetConf(SQLConf.PATH_ENABLED)
+ }
+ }
+ }
+ }
+ }
+
+ private def asContextError(parameter: String, value: String)(f: => Unit):
Unit = {
+ try {
+ f
+ } catch {
+ case e: SparkThrowable with Throwable
+ if !isInternalError(e) &&
+ !Option(e.getCondition).exists(_.endsWith("PARSE_SQL_CONTEXT")) =>
+ throw QueryExecutionErrors.invalidParseSqlContext(parameter, value, e)
+ }
+ }
+
+ 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(
Review Comment:
applyPathOverride is gone with current_path.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -407,19 +720,167 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(
+ analyzed: LogicalPlan,
+ parsed: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(analyzed)
+ 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(parsed)
+ }
+ }
+
+ private def uniqueIdentities(
+ ids: Seq[Seq[String]],
+ session: ClassicSparkSession): Seq[Seq[String]] = {
+ val sensitive = session.sessionState.conf.caseSensitiveAnalysis
+ val seen = mutable.LinkedHashSet.empty[Seq[String]]
+ ids.filter { parts =>
+ val key = if (sensitive) parts else parts.map(_.toLowerCase(Locale.ROOT))
+ seen.add(key)
+ }
+ }
+
+ private def sessionIdent(name: String): Seq[String] =
+ Seq(CatalogManager.SYSTEM_CATALOG_NAME, CatalogManager.SESSION_NAMESPACE,
name)
+
+ private def isTempViewCreate(plan: LogicalPlan): Boolean = plan match {
+ case c: CreateViewCommand =>
+ c.viewType == LocalTempView || c.viewType == GlobalTempView
+ case _: CreateTempViewUsing => true
+ case _ => false
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession,
+ dml: Boolean): Seq[String] = {
+ if (isTempViewCreate(plan)) {
+ sessionIdent(parts.last)
+ } else if (dml) {
+ qualifyRelation(parts, session)
+ } else {
+ qualifyWithCurrentNamespace(parts, session)
+ }
+ }
+
+ private def qualifyRelation(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ val catalog = session.sessionState.catalog
+ if (CatalogManager.isSessionQualifiedViewName(parts) &&
catalog.isTempView(parts)) {
+ sessionIdent(parts.last)
+ } else {
+ resolvedRelationIdentity(parts, session)
+ .getOrElse(qualifyWithCurrentNamespace(parts, session))
+ }
+ }
+
+ /**
+ * Identity of the relation `parts` resolves to, so a name found through the
+ * path is reported where it was found. None when it does not resolve.
+ */
+ private def resolvedRelationIdentity(
+ parts: Seq[String],
+ session: ClassicSparkSession): Option[Seq[String]] = {
+ try {
+ val catalog = session.sessionState.catalog
+ session.sessionState.analyzer.execute(UnresolvedRelation(parts)) match {
+ case s: SubqueryAlias =>
+ val name = s.identifier.name
+ val qualifier = s.identifier.qualifier
+ if (catalog.isTempView(qualifier :+ name) ||
+ (parts.size == 1 && catalog.getRawTempView(name).isDefined)) {
+ Some(sessionIdent(name))
+ } else if (qualifier.nonEmpty) {
+ Some(qualifier :+ name)
+ } else {
+ None
+ }
+ case _ => None
+ }
+ } catch {
+ case NonFatal(_) => None
Review Comment:
resolvedRelationIdentity and the NonFatal swallow are gone; we no longer
re-analyze names.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,70 +122,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, spans, context)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false).
+ * After a context override failure, references stay unqualified parse names
+ * rather than the invoker or the attempted context.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ successJson(plan, None, Some(e), Some(session), span, qualify
= false)
+ }.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 in the original batch. */
+ private case class StatementSpan(start: Int, length: Int)
+
+ /**
+ * Code-point `start` for each statement in one forward pass over `batch`.
+ * Statements are in source order, so prefix counts are not repeated.
+ */
+ private def statementSpans(
Review Comment:
JSON start / length are Unicode code points; Origin indexes stay UTF-16
until SPARK-60672. Calling that out in the PR description.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,70 +122,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val spans = statementSpans(sql, statements)
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, spans, context)
+ } else {
+ JArray(statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(_, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None, span)
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false).
+ * After a context override failure, references stay unqualified parse names
+ * rather than the invoker or the attempted context.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ spans: Seq[StatementSpan],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.zip(spans).map { case (segment, span) =>
+ parsePlanOrError(segment, span)
+ }
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session), span)
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.zip(spans).map {
+ case (ParseFailed(errorObj), _) => errorObj
+ case (Parsed(_, plan), span) =>
+ successJson(plan, None, Some(e), Some(session), span, qualify
= false)
+ }.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 in the original batch. */
+ private case class StatementSpan(start: Int, length: Int)
+
+ /**
+ * Code-point `start` for each statement in one forward pass over `batch`.
+ * Statements are in source order, so prefix counts are not repeated.
+ */
+ private def statementSpans(
+ batch: String,
+ statements: Seq[PositionedSqlStatement]): Seq[StatementSpan] = {
+ var lastUtf16 = 0
+ var lastCp = 0
+ statements.map { segment =>
+ val utf16 = math.max(0, math.min(segment.start, batch.length))
+ if (utf16 >= lastUtf16) {
+ lastCp += batch.codePointCount(lastUtf16, utf16)
+ lastUtf16 = utf16
+ } else {
+ lastCp = batch.codePointCount(0, utf16)
+ lastUtf16 = utf16
+ }
+ StatementSpan(lastCp + 1, charLength(segment))
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): JObject = {
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ segment: PositionedSqlStatement,
+ span: StatementSpan): 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, span))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, span))
+ }
+ }
+
+ /** Analyze `plan` without executing it. Scripts never reach QueryExecution.
*/
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ plan match {
+ case _: CompoundBody =>
+ (None, Some(QueryExecutionErrors.parseSqlAnalyzeUnsupportedScript()))
+ case _ =>
+ try {
+ val analyzed =
+ session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
(None, Some(e))
+ }
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ * Restore writes back the stored catalog name, namespace and PATH flag
+ * without re-validating, and unsets PATH when it was unset.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.storedCurrentCatalogName
+ val prevNamespace = mgr.storedCurrentNamespace
+ val prevPath = mgr.storedSessionPathEntries
+ val pathEnabledWasSet = conf.contains(SQLConf.PATH_ENABLED.key)
+ val prevPathEnabled =
+ if (pathEnabledWasSet) Some(conf.getConf(SQLConf.PATH_ENABLED)) else
None
+ try {
+ context.catalog.foreach { catalog =>
+ asContextError("current_catalog", catalog) {
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ }
+ context.schema.foreach { schema =>
+ asContextError("current_schema", 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 =>
+ asContextError("current_path", path) {
+ if (!conf.getConf(SQLConf.PATH_ENABLED)) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ }
+ f
+ } finally {
+ try {
+ mgr.restoreStoredCurrentCatalogName(prevCatalog)
+ mgr.restoreStoredCurrentNamespace(prevNamespace)
+ } finally {
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ prevPathEnabled match {
+ case Some(value) => conf.setConf(SQLConf.PATH_ENABLED, value)
+ case None => conf.unsetConf(SQLConf.PATH_ENABLED)
+ }
+ }
+ }
+ }
+ }
+
+ private def asContextError(parameter: String, value: String)(f: => Unit):
Unit = {
+ try {
+ f
+ } catch {
+ case e: SparkThrowable with Throwable
+ if !isInternalError(e) &&
+ !Option(e.getCondition).exists(_.endsWith("PARSE_SQL_CONTEXT")) =>
+ throw QueryExecutionErrors.invalidParseSqlContext(parameter, value, e)
+ }
+ }
+
+ private def applyPathOverride(
+ mgr: CatalogManager,
+ conf: SQLConf,
+ path: String): Unit = {
+ if (path.trim.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ val elements = try {
+ parser.get().parsePathElements(path)
+ } catch {
+ case _: ParseException =>
+ throw QueryExecutionErrors.invalidParseSqlContext("current_path",
path)
+ }
+ val expanded = PathElement.validateNoStaticDuplicates(
+ PathElement.expand(elements, conf, mgr),
+ conf.caseSensitiveAnalysis)
+ if (expanded.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ mgr.setSessionPath(expanded)
+ }
}
}
/** Build success JSON from an already-parsed unresolved plan. */
- private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement):
JObject = {
+ private def successJson(
+ plan: LogicalPlan,
+ analyzed: Option[LogicalPlan],
+ analyzeError: Option[SparkThrowable with Throwable],
+ session: Option[ClassicSparkSession],
+ span: StatementSpan,
+ qualify: Boolean = true): 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) {
+ val dmlTarget = plan match {
+ case _: InsertIntoStatement | _: UnresolvedInsert | _: UpdateTable | _:
DeleteFromTable |
+ _: MergeIntoTable => true
+ case _ => false
+ }
+ val targetTables = session.filter(_ => qualify).map { s =>
Review Comment:
Dropped the dead context helpers. Remaining walk-style nits are leftover
from PARSE collection; I would rather not reshuffle this file again in the same
PR.
--
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]
