srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4408723449
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): JObject = {
+ /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the
call. */
+ private def isInternalError(e: SparkThrowable): Boolean =
+ Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR"))
+
+ /**
+ * Character span of a statement object; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): ParsedStatement = {
val sql = segment.statement
try {
// Do not inherit the outer query's origin from the parse_sql expression.
// Errors and parsed nodes refer to the individual statement.
val origin = if (sql.nonEmpty) {
- Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText
= Some(sql))
+ Origin(
+ startIndex = Some(0),
+ stopIndex = Some(sql.codePointCount(0, sql.length) - 1),
+ sqlText = Some(sql))
} else {
Origin(sqlText = Some(sql))
}
CurrentOrigin.withOrigin(origin) {
- val plan = parser.get().parsePlan(sql)
- successJson(plan, segment)
+ Parsed(segment, parser.get().parsePlan(sql))
}
} catch {
- // User-facing parse / scripting failures become JSON; everything else
fails.
case e: ParseException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
Review Comment:
Fixed in 2c8af4ccad9. ANALYZE now rejects a top-level CompoundBody before
QueryExecution, with PARSE_SQL_ANALYZE_UNSUPPORTED_SCRIPT and analyze_success
false. The regression test inserts/drops inside BEGIN ... END and checks that
table data, the dropped table, and the current schema are unchanged.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -407,12 +721,109 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(plan: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(plan)
+ if (query.resolved && query.output.nonEmpty) {
+ query.output.map { attr =>
+ JObject(
+ "name" -> partsToJArray(Seq(attr.name)),
+ "type" -> DescribeRelationJsonCommand.jsonType(
+
CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType)))
+ }
+ } else {
+ collectSelectList(plan)
+ }
+ }
+
+ private def qualifyRelation(
+ parts: Seq[String],
+ session: ClassicSparkSession,
+ resolve: Boolean = true): Seq[String] = {
+ val catalog = session.sessionState.catalog
+ val sessionIdent = Seq(CatalogManager.SYSTEM_CATALOG_NAME,
+ CatalogManager.SESSION_NAMESPACE, parts.last)
+ parts match {
+ case Seq(name) if catalog.getRawTempView(name).isDefined =>
+ sessionIdent
+ case _ if CatalogManager.isSessionQualifiedViewName(parts) &&
catalog.isTempView(parts) =>
+ sessionIdent
+ case _ if catalog.isTempView(parts) =>
+ parts match {
+ case Seq(db, name) => Seq(db, name)
+ case other => other
+ }
+ case _ =>
+ Option.when(resolve)(resolvedRelationIdentity(parts, session)).flatten
+ .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 {
+ session.sessionState.analyzer.execute(UnresolvedRelation(parts)) match {
+ case s: SubqueryAlias if s.identifier.qualifier.nonEmpty =>
+ Some(s.identifier.qualifier :+ s.identifier.name)
+ case _ => None
+ }
+ } catch {
+ case NonFatal(_) => None
+ }
+ }
+
+ private def qualifyRoutine(
Review Comment:
Fixed in 2c8af4ccad9 for the cases that blocked landing: qualifyRoutine now
consults TableFunctionRegistry.builtin (so range is system.builtin.range) and
respects spark.sql.functionResolution.sessionOrder for temp vs builtin.
Path-resolved persistent functions still fall back to the current namespace
when analysis identities are not available; reading those from the analyzed
plan is the remaining follow-up.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -64,68 +119,325 @@ object ParseSqlResult {
ThreadLocal.withInitial(() => new SparkSqlParser())
/** Parse `sql` as a batch and render the JSON result array. */
- def fromSql(sql: String): String = {
+ def fromSql(
+ sql: String,
+ mode: String = ParseSqlMode.Parse,
+ context: ParseSqlAnalyzeContext = ParseSqlAnalyzeContext.empty): String
= {
val split = parser.get().splitStatementsWithPositions(sql)
val statements = split.completeStatements ++ split.partialStatement
- compact(render(JArray(statements.map(parseStatement).toList)))
+ val json = if (mode == ParseSqlMode.Analyze) {
+ analyzeBatch(sql, statements, context)
+ } else {
+ JArray(statements.map { segment =>
+ parsePlanOrError(sql, segment) match {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(seg, plan) =>
+ successJson(plan, analyzed = None, analyzeError = None, None,
segmentSpan(sql, seg))
+ }
+ }.toList)
+ }
+ compact(render(json))
+ }
+
+ private sealed trait ParsedStatement
+ private case class Parsed(segment: PositionedSqlStatement, plan: LogicalPlan)
+ extends ParsedStatement
+ private case class ParseFailed(errorObj: JObject) extends ParsedStatement
+
+ /**
+ * Parse every statement, then analyze each with optional context overrides.
+ * Statements are analyzed against the session as it is, without being
+ * executed, so one does not see the effect of an earlier one.
+ * Analysis / context failures stay JSON (`analyze_success` false) with
+ * whatever parse and catalog qualification is already known.
+ */
+ private def analyzeBatch(
+ batch: String,
+ statements: Seq[PositionedSqlStatement],
+ context: ParseSqlAnalyzeContext): JArray = {
+ val parsed = statements.map(parsePlanOrError(batch, _))
+ val session = ClassicSparkSession.getActiveSession
+ .getOrElse(throw QueryExecutionErrors.parseSqlAnalyzeRequiresSession())
+ // The batch is analyzed on its own: a SQL script that calls parse_sql does
+ // not lend it its locals, and CALL is not run.
+ SqlScriptingContextManager.create(null).runWith {
+ InvokeProcedures.withoutInvocation {
+ try {
+ withAnalyzeContext(session, context) {
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ val (analyzed, error) = analyze(session, plan)
+ successJson(plan, analyzed, error, Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) =>
+ JArray(parsed.map {
+ case ParseFailed(errorObj) => errorObj
+ case Parsed(segment, plan) =>
+ successJson(plan, None, Some(e), Some(session),
segmentSpan(batch, segment))
+ }.toList)
+ }
+ }
+ }
}
- private def parseStatement(segment: PositionedSqlStatement): JObject = {
+ /** INTERNAL_ERROR is a Spark bug, not an analysis result: it fails the
call. */
+ private def isInternalError(e: SparkThrowable): Boolean =
+ Option(e.getCondition).exists(_.startsWith("INTERNAL_ERROR"))
+
+ /**
+ * Character span of a statement object; `text` is the statement's own text,
+ * which its error indexes are relative to.
+ */
+ private case class StatementSpan(start: Int, length: Int, text: String)
+
+ private def segmentSpan(batch: String, segment: PositionedSqlStatement):
StatementSpan =
+ StatementSpan(charStart(batch, segment), charLength(segment),
segment.statement)
+
+ /** Code points `[from, until)` of `text`, clamped to its bounds. */
+ private def codePointSlice(text: String, from: Int, until: Int): String = {
+ val count = text.codePointCount(0, text.length)
+ val begin = math.max(0, math.min(from, count))
+ val end = math.max(begin, math.min(until, count))
+ text.substring(text.offsetByCodePoints(0, begin),
text.offsetByCodePoints(0, end))
+ }
+
+ /** Parsed plan, or parse-failure JSON for this statement. */
+ private def parsePlanOrError(
+ batch: String,
+ segment: PositionedSqlStatement): ParsedStatement = {
val sql = segment.statement
try {
// Do not inherit the outer query's origin from the parse_sql expression.
// Errors and parsed nodes refer to the individual statement.
val origin = if (sql.nonEmpty) {
- Origin(startIndex = Some(0), stopIndex = Some(sql.length - 1), sqlText
= Some(sql))
+ Origin(
+ startIndex = Some(0),
+ stopIndex = Some(sql.codePointCount(0, sql.length) - 1),
+ sqlText = Some(sql))
} else {
Origin(sqlText = Some(sql))
}
CurrentOrigin.withOrigin(origin) {
- val plan = parser.get().parsePlan(sql)
- successJson(plan, segment)
+ Parsed(segment, parser.get().parsePlan(sql))
}
} catch {
- // User-facing parse / scripting failures become JSON; everything else
fails.
case e: ParseException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
case e: SqlScriptingException =>
- errorJson(e, segment)
+ ParseFailed(errorJson(e, segmentSpan(batch, segment)))
+ }
+ }
+
+ /** Analyze `plan` without executing it. */
+ private def analyze(
+ session: ClassicSparkSession,
+ plan: LogicalPlan): (Option[LogicalPlan], Option[SparkThrowable with
Throwable]) = {
+ try {
+ val analyzed = session.sessionState.executePlan(plan,
CommandExecutionMode.SKIP).analyzed
+ (Some(analyzed), None)
+ } catch {
+ case e: SparkThrowable with Throwable if !isInternalError(e) => (None,
Some(e))
+ }
+ }
+
+ /**
+ * Run `f` with optional catalog / schema / SQL path overrides, then restore
+ * the invoker session. `None` on a field keeps the current value.
+ */
+ private def withAnalyzeContext[T](
+ session: ClassicSparkSession,
+ context: ParseSqlAnalyzeContext)(f: => T): T = {
+ if (context.isEmpty) {
+ f
+ } else {
+ val mgr = session.sessionState.catalogManager
+ val conf = session.sessionState.conf
+ val prevCatalog = mgr.currentCatalog.name()
+ val prevNamespace = mgr.currentNamespace.clone()
+ val prevPath = mgr.storedSessionPathEntries
+ val prevPathEnabled = conf.getConf(SQLConf.PATH_ENABLED)
+ try {
+ context.catalog.foreach { catalog =>
+ if (catalog.trim.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_catalog", catalog)
+ }
+ mgr.setCurrentCatalog(catalog)
+ }
+ context.schema.foreach { schema =>
+ val parts = try {
+ parser.get().parseMultipartIdentifier(schema)
+ } catch {
+ case _: ParseException =>
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ if (parts.isEmpty) {
+ throw
QueryExecutionErrors.invalidParseSqlContext("current_schema", schema)
+ }
+ mgr.setCurrentNamespace(parts.toArray)
+ }
+ context.path.foreach { path =>
+ if (!prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, true)
+ }
+ applyPathOverride(mgr, conf, path)
+ }
+ f
+ } finally {
+ mgr.setCurrentCatalog(prevCatalog)
+ mgr.setCurrentNamespace(prevNamespace)
+ prevPath match {
+ case Some(entries) => mgr.setSessionPath(entries)
+ case None => mgr.clearSessionPath()
+ }
+ if (conf.getConf(SQLConf.PATH_ENABLED) != prevPathEnabled) {
+ conf.setConf(SQLConf.PATH_ENABLED, prevPathEnabled)
+ }
+ }
+ }
+ }
+
+ private def applyPathOverride(
+ mgr: CatalogManager,
+ conf: SQLConf,
+ path: String): Unit = {
+ if (path.trim.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ val elements = try {
+ parser.get().parsePathElements(path)
+ } catch {
+ case _: ParseException =>
+ throw QueryExecutionErrors.invalidParseSqlContext("current_path",
path)
+ }
+ val expanded = PathElement.validateNoStaticDuplicates(
+ PathElement.expand(elements, conf, mgr),
+ conf.caseSensitiveAnalysis)
+ if (expanded.isEmpty) {
+ mgr.clearSessionPath()
+ } else {
+ mgr.setSessionPath(expanded)
+ }
}
}
/** Build success JSON from an already-parsed unresolved plan. */
- private def successJson(plan: LogicalPlan, segment: PositionedSqlStatement):
JObject = {
+ private def successJson(
+ plan: LogicalPlan,
+ analyzed: Option[LogicalPlan],
+ analyzeError: Option[SparkThrowable with Throwable],
+ session: Option[ClassicSparkSession],
+ span: StatementSpan): JObject = {
val classification = SqlStatementCodes.classify(plan)
val fields = mutable.ListBuffer.empty[JField]
- fields += "start" -> JInt(segment.start + 1)
- fields += "length" -> JInt(segment.length)
+ fields += "start" -> JInt(span.start)
+ fields += "length" -> JInt(span.length)
fields += "parse_success" -> JBool(true)
fields += "statement_identifier" ->
JString(classification.statementIdentifier)
fields += "statement_code" -> JInt(classification.statementCode)
+ if (session.nonEmpty) {
+ fields += "analyze_success" -> JBool(analyzed.isDefined)
+ }
// Omit unused collections / markers so consumers can treat absence as
empty.
val refs = collectPlanReferences(plan)
- if (refs.targetTables.nonEmpty) {
+ // Qualify from the ANALYZE session even when analysis failed: names and
+ // builtins are known from parse plus current catalog / schema / path.
+ // A DML target is the existing object its name resolves to; a DDL target
+ // is named in the current schema.
+ val dmlTarget = plan match {
+ case _: InsertIntoStatement | _: UnresolvedInsert | _: UpdateTable | _:
DeleteFromTable |
+ _: MergeIntoTable => true
+ case _ => false
+ }
+ val targetTables = session.map(s =>
refs.targetTables.map(qualifyRelation(_, s, dmlTarget)))
Review Comment:
Fixed in 2c8af4ccad9. CREATE TEMPORARY VIEW / CreateTempViewUsing always
report system.session, independent of whether the view already exists. Other
DDL targets use the current catalog/schema and ignore a same-named temp view.
Covered by ANALYZE qualifies DDL targets by statement kind.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -407,12 +721,109 @@ object ParseSqlResult {
named.map(selectListItem)
}
+ /** Select-list items with SQL types from an analyzed query's output. */
+ private def collectTypedSelectList(plan: LogicalPlan): Seq[JObject] = {
+ val query = primaryQueryPlan(plan)
+ if (query.resolved && query.output.nonEmpty) {
+ query.output.map { attr =>
+ JObject(
+ "name" -> partsToJArray(Seq(attr.name)),
+ "type" -> DescribeRelationJsonCommand.jsonType(
+
CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType)))
+ }
+ } else {
+ collectSelectList(plan)
Review Comment:
Fixed in 2c8af4ccad9. primaryQueryPlan unwraps
CreateDataSourceTableAsSelectCommand, InsertIntoDataSourceCommand, and other
LeafRunnableCommand query children (Hive CTAS included). V2WriteCommand /
ReplaceData / WriteDelta are not unwrapped, so row-level rewrite columns stay
out of select_list. Session-catalog CTAS is covered by ANALYZE types
session-catalog CTAS query columns.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -93,22 +190,109 @@ case class ParseSql(child: Expression)
"configKey" -> SQLConf.PARSE_SQL_ENABLED.key,
"configValue" -> "true"))
}
- super.checkInputDataTypes()
+ val typeCheck = super.checkInputDataTypes()
+ if (!typeCheck.isSuccess) {
+ typeCheck
+ } else {
+ val parsedMode = if (mode.foldable) {
Review Comment:
Fixed in 2c8af4ccad9. A non-foldable mode now fails analysis with
NON_FOLDABLE_INPUT. ANALYZE arguments must be foldable or one of
current_catalog() / current_schema() / current_path() / current_user(); the old
deterministic-UDF walk is gone. PARSE with a column of statements and a literal
mode still works.
--
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]
