srielau commented on code in PR #59873: URL: https://github.com/apache/spark/pull/59873#discussion_r4479040768
########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala: ########## @@ -0,0 +1,35 @@ +/* + * 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.optimizer + +import org.apache.spark.sql.catalyst.expressions.{Literal, ParseSql} +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.types.StringType + +/** + * Evaluates constant `parse_sql` calls on the driver. ANALYZE needs the + * session catalog, so this must run even when `ConstantFolding` is excluded. + */ +object EvaluateParseSql extends Rule[LogicalPlan] { + override def apply(plan: LogicalPlan): LogicalPlan = + plan.transformAllExpressionsWithSubqueries { + case p: ParseSql if p.children.forall(_.foldable) => + Literal.create(p.eval(), StringType) Review Comment: Fixed in 6fc55256046. The rule now folds only constant ANALYZE calls (`ParseSql.isConstantAnalyze`), so PARSE stays with ConstantFolding and keeps its tolerance for failing constants in untaken branches. A constant ANALYZE whose SQL argument fails to evaluate is left in place, and the error surfaces only if that branch runs. Tests: PARSE under ANSI returns NULL for `CASE WHEN id > 0 THEN parse_sql(CAST(1/0 AS STRING)) END`, and the ANALYZE variant also returns NULL. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala: ########## @@ -0,0 +1,35 @@ +/* + * 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.optimizer + +import org.apache.spark.sql.catalyst.expressions.{Literal, ParseSql} +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.types.StringType + +/** + * Evaluates constant `parse_sql` calls on the driver. ANALYZE needs the + * session catalog, so this must run even when `ConstantFolding` is excluded. + */ +object EvaluateParseSql extends Rule[LogicalPlan] { + override def apply(plan: LogicalPlan): LogicalPlan = + plan.transformAllExpressionsWithSubqueries { Review Comment: Fixed in 6fc55256046. The rule uses `transformDownWithSubqueriesAndPruning` and `transformAllExpressionsWithPruning` on a new `PARSE_SQL` tree pattern that `ParseSql` declares, so plans without parse_sql are not walked. Subquery calls are still folded; the scalar-subquery ANALYZE test still passes with ConstantFolding excluded. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -489,19 +652,274 @@ object ParseSqlResult { named.map(selectListItem) } + /** + * Select-list items with SQL types from the analyzed query output, so `*` + * is expanded and names match analysis. INSERT reports the write projection + * (target columns), not the user's SELECT aliases. + */ + 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 catalogIdentParts(catalog: String, ident: Identifier): Seq[String] = + Seq(catalog) ++ ident.namespace() :+ ident.name() + + private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] = + (r.catalog, r.identifier) match { + case (Some(catalog), Some(ident)) => Some(catalogIdentParts(catalog.name(), ident)) + case _ => None + } + + private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] = + r.catalogTable.map { t => + val id = t.identifier + if (id.catalog.isEmpty && id.database.isDefined) { + CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id) + } else { + tableIdentifierParts(id) + } + } + + private def viewParts(v: View): Seq[String] = { + val id = v.desc.identifier + if (v.isTempView) { + val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) + if (id.database.exists(_.equalsIgnoreCase(globalTemp))) { + Seq(id.catalog.getOrElse(CatalogManager.SESSION_CATALOG_NAME)) ++ + id.database.toSeq :+ id.table + } else { + sessionIdent(id.table) + } + } else { + tableIdentifierParts(id) + } + } + + private def tempViewParts(ident: Identifier): Seq[String] = { + val parts = ident.namespace().toSeq :+ ident.name() + val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) + val ns = ident.namespace() + if (ns.length == 1 && ns.head.equalsIgnoreCase(globalTemp)) { + Seq(CatalogManager.SESSION_CATALOG_NAME, ns.head, ident.name()) + } else if (ns.isEmpty || CatalogManager.isSessionQualifiedViewName(parts)) { + sessionIdent(ident.name()) + } else { + catalogIdentParts(CatalogManager.SESSION_CATALOG_NAME, ident) + } + } + + private def globalTempIdent(name: String): Seq[String] = + Seq( + CatalogManager.SESSION_CATALOG_NAME, + SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE), + name) + + /** + * Match each parse name to an analyzed identity at most once. Longer parse + * names claim first so `spark_catalog.db.t` is not stolen by a later bare + * `t`. Equal-length matches count: a 3-part parse name is already complete. + */ + private def qualifyParseIdentities( + parseIds: Seq[Seq[String]], + analyzedIds: Seq[Seq[String]], + fallback: Seq[String] => Seq[String]): Seq[Seq[String]] = { + val remaining = analyzedIds.to(mutable.ArrayBuffer) + val assigned = Array.fill[Option[Seq[String]]](parseIds.length)(None) + parseIds.zipWithIndex.sortBy { case (parts, _) => -parts.length }.foreach { + case (parts, i) => + matchingIdentity(parts, remaining.toSeq) match { + case Some(matched) => + val matchedKey = matched.map(_.toLowerCase(Locale.ROOT)) + val idx = remaining.indexWhere { analyzed => + analyzed.map(_.toLowerCase(Locale.ROOT)) == matchedKey + } + if (idx >= 0) { + remaining.remove(idx) + } + assigned(i) = Some(matched) + case None => + val ident = + if (isPathReference(parts)) parts else fallback(parts) + assigned(i) = Some(ident) + } + } + assigned.map(_.get).toSeq + } + + private def matchingIdentity( + parseParts: Seq[String], + analyzedIds: Seq[Seq[String]]): Option[Seq[String]] = { + val parsed = parseParts.map(_.toLowerCase(Locale.ROOT)) + analyzedIds.find { analyzed => Review Comment: Fixed in 6fc55256046. Two sources of the misattribution are closed. (1) A row-level rewrite (`ReplaceData` / `WriteDelta`) records its write target, and a source read of that same identity is skipped as the rewrite's re-read. (2) `SQLFunctionNode` bodies are not walked. Tests: `MERGE INTO prod.t USING t` under `USE staging` reports `staging.t` as the source, and `SELECT * FROM parse_sql_tvf() JOIN t` reports only the user's `t`. With the fix disabled, both tests fail. Known limitation: when a statement reads its own write target (self-merge, or a subquery re-reading the target), that source now falls back to name-based qualification, which matches the written name in the current schema. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -489,19 +652,274 @@ object ParseSqlResult { named.map(selectListItem) } + /** + * Select-list items with SQL types from the analyzed query output, so `*` + * is expanded and names match analysis. INSERT reports the write projection + * (target columns), not the user's SELECT aliases. + */ + 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 catalogIdentParts(catalog: String, ident: Identifier): Seq[String] = + Seq(catalog) ++ ident.namespace() :+ ident.name() + + private def v2RelationParts(r: DataSourceV2Relation): Option[Seq[String]] = + (r.catalog, r.identifier) match { + case (Some(catalog), Some(ident)) => Some(catalogIdentParts(catalog.name(), ident)) + case _ => None + } + + private def logicalRelationParts(r: LogicalRelation): Option[Seq[String]] = + r.catalogTable.map { t => + val id = t.identifier + if (id.catalog.isEmpty && id.database.isDefined) { + CatalogManager.SESSION_CATALOG_NAME +: tableIdentifierParts(id) + } else { + tableIdentifierParts(id) + } + } + + private def viewParts(v: View): Seq[String] = { + val id = v.desc.identifier + if (v.isTempView) { + val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) + if (id.database.exists(_.equalsIgnoreCase(globalTemp))) { + Seq(id.catalog.getOrElse(CatalogManager.SESSION_CATALOG_NAME)) ++ + id.database.toSeq :+ id.table + } else { + sessionIdent(id.table) + } + } else { + tableIdentifierParts(id) + } + } + + private def tempViewParts(ident: Identifier): Seq[String] = { + val parts = ident.namespace().toSeq :+ ident.name() + val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) + val ns = ident.namespace() + if (ns.length == 1 && ns.head.equalsIgnoreCase(globalTemp)) { + Seq(CatalogManager.SESSION_CATALOG_NAME, ns.head, ident.name()) + } else if (ns.isEmpty || CatalogManager.isSessionQualifiedViewName(parts)) { + sessionIdent(ident.name()) + } else { + catalogIdentParts(CatalogManager.SESSION_CATALOG_NAME, ident) + } + } + + private def globalTempIdent(name: String): Seq[String] = + Seq( + CatalogManager.SESSION_CATALOG_NAME, + SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE), + name) + + /** + * Match each parse name to an analyzed identity at most once. Longer parse + * names claim first so `spark_catalog.db.t` is not stolen by a later bare + * `t`. Equal-length matches count: a 3-part parse name is already complete. + */ + private def qualifyParseIdentities( + parseIds: Seq[Seq[String]], + analyzedIds: Seq[Seq[String]], + fallback: Seq[String] => Seq[String]): Seq[Seq[String]] = { + val remaining = analyzedIds.to(mutable.ArrayBuffer) + val assigned = Array.fill[Option[Seq[String]]](parseIds.length)(None) + parseIds.zipWithIndex.sortBy { case (parts, _) => -parts.length }.foreach { + case (parts, i) => + matchingIdentity(parts, remaining.toSeq) match { + case Some(matched) => + val matchedKey = matched.map(_.toLowerCase(Locale.ROOT)) + val idx = remaining.indexWhere { analyzed => + analyzed.map(_.toLowerCase(Locale.ROOT)) == matchedKey + } + if (idx >= 0) { + remaining.remove(idx) + } + assigned(i) = Some(matched) + case None => + val ident = + if (isPathReference(parts)) parts else fallback(parts) + assigned(i) = Some(ident) + } + } + assigned.map(_.get).toSeq + } + + private def matchingIdentity( + parseParts: Seq[String], + analyzedIds: Seq[Seq[String]]): Option[Seq[String]] = { + val parsed = parseParts.map(_.toLowerCase(Locale.ROOT)) + analyzedIds.find { analyzed => + val ident = analyzed.map(_.toLowerCase(Locale.ROOT)) + ident.endsWith(parsed) && + (ident.length > parsed.length || + ident.length == parsed.length && parsed.length >= 2) + } + } + + private def isPathReference(parts: Seq[String]): Boolean = parts match { + case Seq(_, path) => path.contains("/") || path.contains(":") Review Comment: Fixed in 6fc55256046. When analysis succeeded and a two-part name matches no analyzed identity, it is reported as written if its first part is a data source and not a database (`isFileReference`). So a relative `parquet.`data`` reports `[parquet, data]` with `analyze_success` true. The check uses the analysis result and the format lookup, not the file relation itself, since the written name is not recoverable from the relation. Compatibility: absolute and URI paths keep the path rule, `CREATE VIEW v AS SELECT a FROM db.s` still reports `spark_catalog.db.s`, and failed statements keep the fallback. Tests: relative path, and the existing absolute path test. ########## sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala: ########## @@ -43,170 +85,251 @@ import org.apache.spark.sql.execution.datasources.CreateTempViewUsing * executors without a session, so only the stock parser is available under * distributed eval. * - * Every statement object includes its 1-based Unicode code-point `start` in the - * original batch and its code-point `length`, excluding surrounding whitespace - * and the terminating semicolon. On success it also includes + * Every statement object includes its 1-based character `start` in the + * original batch and its character `length`, excluding surrounding + * whitespace and the terminating semicolon. Characters are Unicode code + * points (Spark SQL `length()` / UTF-8 strings): a supplementary-plane + * emoji is one character. Java `String` offsets are UTF-16 (emoji = 2); + * ANTLR `CodePointCharStream` offsets are already code points (emoji = 1). + * The splitter maps ANTLR indexes to UTF-16 for `substring`; JSON converts + * those spans back to code points. On success it also includes * `parse_success`, the statement * identifier/code (ISO/IEC 9075-2:2023 Table 39), and omits unused optional * fields (`target_table_references`, `source_table_references`, * `function_references`, `select_list`, `parameter_markers`) when empty. On * parse failure the statement object contains `parse_success: false` with * source location and a nested STANDARD-format error object, and parsing - * continues with later statements. Nested error locations are relative to the - * individual statement, while `start` is relative to the original batch; error - * `queryContext` indexes are code points as well. An - * empty or closed-comment-only batch produces an empty array. Only - * [[ParseException]] / [[SqlScriptingException]] are converted to JSON; + * continues with later statements. Nested `queryContext` is the STANDARD + * renderer output for that error. View and EXECUTE IMMEDIATE bodies keep + * their own text; `start` is relative to the original batch. An empty or + * closed-comment-only batch produces an empty array. PARSE converts + * [[ParseException]] / [[SqlScriptingException]] to JSON. + * ANALYZE also reports non-internal analysis errors as JSON; * unexpected / internal failures propagate so the function fails. + * Nested error locations are statement-relative Unicode code points. */ object ParseSqlResult { private val parser: ThreadLocal[SparkSqlParser] = 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): String = { val split = parser.get().splitStatementsWithPositions(sql) val statements = split.completeStatements ++ split.partialStatement - val starts = charStarts(sql, statements) - compact(render(JArray(statements.zip(starts).map { - case (segment, start) => parseStatement(segment, start) - }.toList))) + val spans = statementSpans(sql, statements) + val json = if (mode == ParseSqlMode.Analyze) { + analyzeBatch(statements, spans) + } 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) Review Comment: Done in 6fc55256046: `Parsed` now holds only the plan. -- 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]
