srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4429936897
##########
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
Review Comment:
Dropped current_catalog / current_schema / current_path from this PR in
37f4a1d. ANALYZE now only inherits the invoker session and never mutates it.
Isolated overlays belong in SPARK-60622.
##########
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 {
Review Comment:
Relation identities now come from the analyzed plan (DataSourceV2Relation,
LogicalRelation, View, ResolvedTable / ResolvedTempView / ResolvedIdentifier,
Create*Command, DropTempViewCommand, and so on). The parse-time re-resolve path
is gone. Function names still come from the parse tree plus the function
registries (builtin, temp, session extras).
--
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]
