srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4408725345
##########
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)
Review Comment:
Fixed in 2c8af4ccad9. Restore now writes back storedCurrentCatalogName /
storedCurrentNamespace without existence checks, so a v2 catalog whose default
namespace does not namespaceExists, or a dropped current namespace, no longer
throws in finally and leak the override. Path restore still runs in an inner
finally.
##########
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)
Review Comment:
Acknowledged. This PR still applies overrides on the invoker session (tools
and CALL suppression share that session). Restore is now a snapshot of stored
catalog/namespace/PATH/PATH-flag, so we no longer re-validate or materialize
unset keys. A concurrent USE or SET PATH that commits during the call can still
be overwritten by that snapshot. A per-call CatalogManager overlay (shared
repair plan 1) is the complete isolation fix; I would rather land that as a
follow-up than grow this PR further with a new CatalogManager implementation.
##########
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)
Review Comment:
Fixed in 2c8af4ccad9. Restore unsets spark.sql.path.enabled when it was
unset, and restores storedCurrentCatalogName to None when the current catalog
still followed spark.sql.defaultCatalog.
##########
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)))
+ .getOrElse(refs.targetTables)
+ val sourceTables = session.map(s =>
refs.sourceTables.map(qualifyRelation(_, s)))
Review Comment:
Fixed in 2c8af4ccad9. After qualification, reference arrays de-duplicate
case-insensitively in first-seen order, so count(a) with COUNT(b) is one
system.builtin.count.
##########
sql/core/src/test/scala/org/apache/spark/sql/catalyst/expressions/ParseSqlAnalyzeSuite.scala:
##########
@@ -0,0 +1,343 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.procedures.{
+ BoundProcedure,
+ ProcedureParameter,
+ UnboundProcedure
+}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataTypes, StructType}
+
+class ParseSqlAnalyzeSuite extends QueryTest with SharedSparkSession {
+
+ private def evalJson(sql: String, mode: String = "ANALYZE"): JValue = {
+ val result = spark.sql(s"SELECT parse_sql($sql,
'$mode')").head().getString(0)
+ parse(result)
+ }
+
+ private def evalStatement(sql: String, mode: String = "ANALYZE"): JObject = {
+ evalJson("'" + sql.replace("'", "''") + "'", mode) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ }
+
+ private def jsonType(name: String): JObject = JObject("name" ->
JString(name))
+ private val jsonInt: JObject = jsonType("int")
+ private val jsonBigint: JObject = jsonType("bigint")
+ private val jsonBoolean: JObject = jsonType("boolean")
+ private val jsonString: JObject = JObject(
+ "name" -> JString("string"),
+ "collation" -> JString("UTF8_BINARY"))
+
+ test("ANALYZE types selected columns and PARSE does not") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val analyzed = evalStatement("SELECT 1 AS a", "ANALYZE")
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+
+ val parsed = evalStatement("SELECT 1 AS a", "PARSE")
+ assert(parsed \ "analyze_success" === JNothing)
+ assert(parsed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a")))))))
+ }
+ }
+
+ test("ANALYZE fully qualifies tables and builtin routines") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTable("parse_sql_analyze_t") {
+ sql("CREATE TABLE parse_sql_analyze_t (a INT, b STRING) USING parquet")
+ val j = evalStatement(
+ "SELECT count(a) AS c FROM parse_sql_analyze_t", "analyze")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"), JString("default"),
JString("parse_sql_analyze_t"))))))
+ assert(j \ "function_references" === JArray(List(JArray(List(
+ JString("system"), JString("builtin"), JString("count"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("c"))), "type" ->
jsonBigint))))
+ }
+ }
+ }
+
+ test("ANALYZE qualifies local temp views as system.session") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ withTempView("parse_sql_analyze_v") {
+ sql("CREATE OR REPLACE TEMP VIEW parse_sql_analyze_v AS SELECT 1 AS a,
'x' AS b")
+ val j = evalStatement("SELECT a, b FROM parse_sql_analyze_v")
+ assert(j \ "analyze_success" === JBool(true))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("system"), JString("session"),
JString("parse_sql_analyze_v"))))))
+ assert(j \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt),
+ JObject("name" -> JArray(List(JString("b"))), "type" ->
jsonString))))
+ }
+ }
+ }
+
+ test("ANALYZE keeps parse metadata when analysis fails") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val j = evalStatement("SELECT a FROM missing_parse_sql_analyze_t")
+ assert(j \ "parse_success" === JBool(true))
+ assert(j \ "analyze_success" === JBool(false))
+ assert(j \ "statement_identifier" === JString("SELECT"))
+ assert(j \ "source_table_references" === JArray(List(JArray(List(
+ JString("spark_catalog"),
+ JString("default"),
+ JString("missing_parse_sql_analyze_t"))))))
+ val errorClass = (j \ "error" \ "errorClass").asInstanceOf[JString].s
+ assert(errorClass == "TABLE_OR_VIEW_NOT_FOUND" ||
+ errorClass.contains("TABLE_OR_VIEW_NOT_FOUND"))
+ }
+ }
+
+ test("ANALYZE inherits invoker schema and can override current_schema") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ sql("CREATE DATABASE parse_sql_ctx_a")
+ sql("CREATE DATABASE parse_sql_ctx_b")
+ try {
+ sql("CREATE TABLE parse_sql_ctx_a.t (a INT) USING parquet")
+ sql("USE parse_sql_ctx_b")
+ try {
+ val missing = evalStatement("SELECT a FROM t")
+ assert(missing \ "analyze_success" === JBool(false))
+ assert(missing \ "source_table_references" ===
JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_b"),
JString("t"))))))
+
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT a FROM t',
+ | 'ANALYZE',
+ | current_schema =>
'parse_sql_ctx_a')""".stripMargin).head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "source_table_references" ===
JArray(List(JArray(List(
+ JString("spark_catalog"), JString("parse_sql_ctx_a"),
JString("t"))))))
+ assert(spark.sql("SELECT current_schema()").head().getString(0) ===
+ "parse_sql_ctx_b")
+ } finally {
+ sql("USE default")
+ }
+ } finally {
+ sql("DROP TABLE IF EXISTS parse_sql_ctx_a.t")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_a CASCADE")
+ sql("DROP DATABASE IF EXISTS parse_sql_ctx_b CASCADE")
+ }
+ }
+ }
+
+ test("ANALYZE named mode and current_catalog inherit from invoker") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | mode => 'ANALYZE',
+ | current_catalog =>
current_catalog())""".stripMargin).head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a"))), "type" -> jsonInt))))
+ }
+ }
+
+ test("ANALYZE current_path override is restored on the invoker") {
+ withSQLConf(
+ SQLConf.PARSE_SQL_ENABLED.key -> "true",
+ SQLConf.PATH_ENABLED.key -> "true") {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ try {
+ val before = spark.sql("SELECT current_path()").head().getString(0)
+ val json = spark.sql(
+ """SELECT parse_sql(
+ | 'SELECT 1 AS a',
+ | 'ANALYZE',
+ | current_path => 'system.builtin,
spark_catalog.default')""".stripMargin)
+ .head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "analyze_success" === JBool(true))
+ assert(spark.sql("SELECT current_path()").head().getString(0) ===
before)
+ } finally {
+ sql("SET PATH = DEFAULT_PATH")
+ }
+ }
+ }
+
+ test("ANALYZE reports invalid current_schema as JSON instead of failing the
query") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql(
+ "SELECT parse_sql('SELECT 1 AS a', 'ANALYZE', current_schema =>
'!!!')")
+ .head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(false))
+ assert(analyzed \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("a")))))))
+ assert((analyzed \ "error" \ "errorClass").asInstanceOf[JString].s ===
+ "INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT")
+ }
+ }
+
+ test("ANALYZE without parameter marker types leaves the query unresolved") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ val json = spark.sql("SELECT parse_sql('SELECT abs(?) AS x', 'ANALYZE')")
+ .head().getString(0)
+ val analyzed = parse(json) match {
+ case JArray(List(value: JObject)) => value
+ case other => fail(s"expected one statement object, got: $other")
+ }
+ assert(analyzed \ "parse_success" === JBool(true))
+ assert(analyzed \ "analyze_success" === JBool(false))
+ }
+ }
+
+ private def evalBatch(sql: String): List[JObject] =
+ evalJson("'" + sql.replace("'", "''") + "'") match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def typedColumns(columns: (String, JObject)*): JArray =
JArray(columns.map {
+ case (name, dataType) => JObject("name" -> JArray(List(JString(name))),
"type" -> dataType)
+ }.toList)
+
+ private def ref(parts: String*): JArray = JArray(parts.map(JString).toList)
+
+ /** Run a full `SELECT parse_sql(...)` and return its statement objects. */
+ private def evalCall(query: String): List[JObject] =
+ parse(spark.sql(query).head().getString(0)) match {
+ case JArray(values) => values.map(_.asInstanceOf[JObject])
+ case other => fail(s"expected a JSON array, got: $other")
+ }
+
+ private def errorClass(statement: JObject): JValue = statement \ "error" \
"errorClass"
+
+ private def queryContext(statement: JObject): JObject =
+ statement \ "error" \ "queryContext" match {
+ case JArray(List(ctx: JObject)) => ctx
+ case other => fail(s"expected one query context, got: $other")
+ }
+
+ private val emoji = new String(Character.toChars(0x1F600))
+
+ test("ANALYZE does not run CALL, directly or through EXECUTE IMMEDIATE") {
+ withSQLConf(SQLConf.PARSE_SQL_ENABLED.key -> "true") {
+ spark.conf.set("spark.sql.catalog.parse_sql_cat",
classOf[InMemoryCatalog].getName)
+ try {
+ val catalog =
spark.sessionState.catalogManager.catalog("parse_sql_cat")
+ .asInstanceOf[InMemoryCatalog]
+ catalog.createProcedure(
+ Identifier.of(Array("ns"), "record"),
ParseSqlAnalyzeSuite.UnboundRecord)
+ ParseSqlAnalyzeSuite.calls.set(0)
+ val results = evalBatch(
+ """CALL parse_sql_cat.ns.record(1);
+ |EXECUTE IMMEDIATE 'CALL
parse_sql_cat.ns.record(2)'""".stripMargin)
+ assert(results.map(_ \ "parse_success") === List(JBool(true),
JBool(true)))
Review Comment:
Tests in 2c8af4ccad9 now assert analyze_success on CALL, a real error class
on hidden script locals, a non-foldable mode column, script non-execution
(data/catalog/schema unchanged), range as system.builtin, DDL temp vs
persistent targets, session-catalog CTAS types, and PARSE_SQL_CONTEXT for an
unknown catalog. current_path still uses SELECT 1; a path-resolved function
case remains a follow-up.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -18,71 +18,168 @@
package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry,
FunctionRegistryBase, TypeCheckResult}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{
+ ExpressionBuilder, FunctionRegistry, FunctionRegistryBase, TypeCheckResult
+}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.catalyst.parser.ParseSqlResult
+import org.apache.spark.sql.catalyst.parser.{
+ ParseSqlAnalyzeContext, ParseSqlMode, ParseSqlResult
+}
+import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature,
InputParameter}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase,
QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.StringTypeWithCollation
import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
import org.apache.spark.unsafe.types.UTF8String
/**
* 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.
+ * statements (source position, identifier/code, lineage references,
+ * select-list names, parameters). `PARSE` mode is unresolved. `ANALYZE`
+ * mode also types the select list and fully qualifies resolved relations
+ * and routines. 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.
+ * User-facing parse / analyze errors become JSON; unexpected internal
+ * failures propagate. `ANALYZE` needs an active Spark session (catalog)
+ * and inherits the invoker's current catalog, schema, and SQL path.
+ * Optional named `current_catalog`, `current_schema`, and `current_path`
+ * override that context for the duration of analysis.
*/
// scalastyle:off line.size.limit
// scalastyle:off nonascii
@ExpressionDescription(
- 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
- 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 `[]`.""",
+ usage = """_FUNC_(sqlStmt[, mode[, current_catalog[, current_schema[,
current_path]]]]) -
+ Splits `sqlStmt` into SQL statements, parses each with the stock Spark SQL
parser, and
+ returns a JSON array describing them so callers can inspect SQL text
(lineage, statement
+ kind, select-list types) without running the statements. Each object has
1-based
+ character `start` and character `length` (Unicode code points; an emoji is
one
+ character), parse success, Table 39 statement identifier/code, target and
source
+ table references, select-list column names, and parameter markers. Every
parameter
+ may be passed by name (`sqlStmt => ...`, `mode => ...`).
+ `mode` is `PARSE` (default) or `ANALYZE` (case-insensitive). `PARSE` is
unresolved and
+ does not need a catalog. `ANALYZE` needs an active session; it types
selected columns,
+ fully qualifies resolved relations and routines, and reports analysis
failures as JSON
+ without aborting the batch. Each statement is analyzed against the session
as it is,
+ without being executed, so a DECLARE, CREATE, USE, or SET does not affect
later
+ statements. `CALL` is analyzed but the procedure is not run. A SQL script
calling
+ `parse_sql` does not lend it its locals or cursors. `ANALYZE` inherits the
invoker's
+ current catalog, schema, and SQL path. Named `current_catalog`,
`current_schema`, and
+ `current_path` override that context for analysis only; omitted or NULL
keeps the
+ invoker value, which is restored after the call. An override changes the
session
+ for the duration of the call, so other queries running concurrently on the
same
+ session can observe it; `current_path` also enables the SQL path for the
call.
+ `PARSE` ignores the catalog arguments. Statement length excludes
surrounding
+ whitespace and the terminating semicolon. Session parser extensions are
not applied.
+ The JSON contract may still change while
spark.sql.function.parseSql.enabled remains
+ off by default. On syntax / parse error returns JSON for that statement
with
+ `parse_success` false, source location, and a nested STANDARD error object
instead of
+ throwing or stopping later statements. Nested error locations are
statement-relative
+ Unicode code points. Unexpected internal failures still fail the outer
query. An
+ empty or closed-comment-only batch returns `[]`. NULL `sqlStmt` or `mode`
returns
+ NULL.""",
arguments = """
Arguments:
* sqlStmt - A SQL batch string to split and parse.
- An expression that evaluates to a string.
+ An expression that evaluates to a string. NULL returns NULL.
+ * mode - Optional `PARSE` or `ANALYZE` (case-insensitive). Defaults to
`PARSE`.
+ An expression that evaluates to a string. NULL returns NULL. Any other
value
+ fails the outer query with INVALID_PARAMETER_VALUE.PARSE_SQL_MODE.
+ * current_catalog - Optional catalog name used as current catalog during
`ANALYZE`.
+ NULL or omitted inherits the invoker session. Ignored in `PARSE`. The
invoker
+ catalog, schema, and path are restored after analysis.
+ * current_schema - Optional schema / namespace used as current schema
during `ANALYZE`.
+ NULL or omitted inherits the invoker session. Ignored in `PARSE`.
+ * current_path - Optional SQL PATH used during `ANALYZE` (SET PATH
grammar, or the
+ string returned by `current_path()`). NULL or omitted inherits the
invoker session.
+ Ignored in `PARSE`.
""",
examples = """
Examples:
> 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
+ 12
> SELECT get_json_object(_FUNC_('SELEC'), '$[0].error.errorClass');
PARSE_SYNTAX_ERROR
+ > SELECT get_json_object(_FUNC_('SELECT 1 AS a', 'ANALYZE'),
'$[0].select_list[0].type.name');
+ int
+ > SELECT get_json_object(_FUNC_('SELECT a FROM missing_t', 'ANALYZE'),
'$[0].analyze_success');
+ false
+ """,
+ note = """
+ Requires spark.sql.function.parseSql.enabled=true. The result type is
STRING JSON.
+ ANALYZE needs an active session, and every argument must be constant for
the query
+ (DATATYPE_MISMATCH.NON_FOLDABLE_INPUT otherwise); a DataFrame column of
statements
+ can use PARSE only. An invalid `current_catalog`, `current_schema`, or
`current_path`
+ is reported as JSON (`analyze_success` false,
INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT)
Review Comment:
Fixed in 2c8af4ccad9. Failures while applying current_catalog /
current_schema / current_path (unknown catalog, missing schema, duplicate path,
parse errors) are mapped to INVALID_PARAMETER_VALUE.PARSE_SQL_CONTEXT with the
parameter name and original cause.
--
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]
