srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4487000191
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -452,11 +616,41 @@ object ParseSqlResult {
case u: UnresolvedView => add(u.multipartIdentifier)
case u: UnresolvedTableOrView => add(u.multipartIdentifier)
case u: UnresolvedIdentifier if role == TableRefRole.Target =>
add(u.nameParts)
- case c: CreateViewCommand => addTarget(tableIdentifierParts(c.name))
- case c: CreateTempViewUsing =>
addTarget(tableIdentifierParts(c.tableIdent))
+ case c: CreateViewCommand => addTarget(Seq(c.name.table))
+ case c: CreateTempViewUsing => addTarget(Seq(c.tableIdent.table))
+ case c: CreateDataSourceTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateDataSourceTableAsSelectCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: CreateTableCommand =>
+ addTarget(persistedTableParts(c.table.identifier))
+ case c: DropTableCommand =>
+ addTarget(tableIdentifierParts(c.tableName))
+ case c: DropTempViewCommand =>
+ addTarget(tempViewParts(c.ident))
case c: CacheTable if c.multipartIdentifier.nonEmpty =>
Review Comment:
Fixed in f9390c2225c. A CACHE TABLE name is collected only when the parse
name is not already analyzed, and the analyzed table is reported as the target,
so `CACHE TABLE db.t` no longer reports `["db","t"]`. Covered by the 'ANALYZE
qualifies CACHE and UNCACHE TABLE targets' test.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +683,307 @@ 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 rewriteTargetParts(table: NamedRelation): Option[Seq[String]] =
table match {
+ case d: DataSourceV2Relation => v2RelationParts(d)
+ 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]],
+ analysisSucceeded: Boolean,
+ session: ClassicSparkSession,
+ 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) ||
+ (analysisSucceeded && isFileReference(parts, session))) {
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ private def isPathReference(parts: Seq[String]): Boolean = parts match {
+ case Seq(_, path) => path.contains("/") || path.contains(":")
+ case Seq(path) => path.contains("/") || path.startsWith("file:")
+ case _ => false
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ case _ => false
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = plan match {
+ case c: CreateViewCommand if c.viewType == LocalTempView =>
+ sessionIdent(parts.last)
+ case c: CreateViewCommand if c.viewType == GlobalTempView =>
+ globalTempIdent(parts.last)
+ case c: CreateTempViewUsing if c.global =>
+ globalTempIdent(parts.last)
+ case _: CreateTempViewUsing =>
+ sessionIdent(parts.last)
+ case _: CreateTableAsSelect | _: ReplaceTableAsSelect | _: CreateTable |
+ _: ReplaceTable | _: CreateView |
+ _: org.apache.spark.sql.execution.datasources.CreateTable =>
+ // Persistent DDL never names a temp view, even if one shares the name.
+ qualifyWithCurrentNamespace(parts, session)
+ case c: CreateViewCommand if c.viewType == PersistedView =>
+ qualifyWithCurrentNamespace(parts, session)
+ case _ =>
+ qualifySource(parts, session)
+ }
+
+ private def qualifySource(
Review Comment:
Fixed in f9390c2225c. The analyzed walk now covers CREATE VIEW, ALTER VIEW,
CREATE TABLE AS SELECT, v1 INSERT, and the SQL function and view bodies, so the
analyzed pool is no longer empty for these statements. The CREATE VIEW and
ALTER VIEW case has a test. The CTAS and v1 INSERT arms do not have a dedicated
test yet.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala:
##########
@@ -489,19 +683,307 @@ 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 rewriteTargetParts(table: NamedRelation): Option[Seq[String]] =
table match {
+ case d: DataSourceV2Relation => v2RelationParts(d)
+ 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]],
+ analysisSucceeded: Boolean,
+ session: ClassicSparkSession,
+ 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) ||
+ (analysisSucceeded && isFileReference(parts, session))) {
+ 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)
+ }
+ }
+
+ /** Absolute or URI-like path written in the statement text. */
+ private def isPathReference(parts: Seq[String]): Boolean = parts match {
+ case Seq(_, path) => path.contains("/") || path.contains(":")
+ case Seq(path) => path.contains("/") || path.startsWith("file:")
+ case _ => false
+ }
+
+ /**
+ * A `format`.`path` reference that analysis resolved to a file relation
without a catalog
+ * table. With `spark.sql.runSQLOnFiles` (default on), any two-part name
whose first part is a
+ * data source and not a database is read as a path, relative or not. The
analyzed plan carries
+ * no identity for it, so the name is reported as written.
+ */
+ private def isFileReference(parts: Seq[String], session:
ClassicSparkSession): Boolean =
+ parts match {
+ case Seq(format, _) =>
+ Try(DataSource.lookupDataSource(format,
session.sessionState.conf)).isSuccess &&
+ !session.sessionState.catalog.databaseExists(format)
+ case _ => false
+ }
+
+ private def qualifyTarget(
+ parts: Seq[String],
+ plan: LogicalPlan,
+ session: ClassicSparkSession): Seq[String] = plan match {
+ case c: CreateViewCommand if c.viewType == LocalTempView =>
+ sessionIdent(parts.last)
+ case c: CreateViewCommand if c.viewType == GlobalTempView =>
Review Comment:
Fixed in f9390c2225c with tests. 'ANALYZE reports temp view targets for
global, USING and EXPLAIN forms' covers CREATE GLOBAL TEMP VIEW (qualified to
the global temp database), CREATE TEMPORARY VIEW ... USING, CREATE GLOBAL
TEMPORARY VIEW ... USING, and EXPLAIN CREATE TEMPORARY VIEW. Each asserts the
reported target and that no table was created.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EvaluateParseSql.scala:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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 scala.util.control.NonFatal
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal,
ParseSql}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.PARSE_SQL
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Evaluates constant ANALYZE `parse_sql` calls on the driver, where the
session catalog is
+ * available. ANALYZE must run here even when `ConstantFolding` is excluded.
PARSE calls are
+ * left to `ConstantFolding`, which keeps its tolerance for failing constants
in untaken
+ * branches.
+ *
+ * A constant ANALYZE call whose SQL argument fails to evaluate is left in
place. The error
+ * then surfaces when the expression runs, so a failure in an untaken branch
stays unobserved.
+ */
+object EvaluateParseSql extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan =
+ plan.transformDownWithSubqueriesAndPruning(_.containsPattern(PARSE_SQL)) {
+ case subQuery =>
+
subQuery.transformAllExpressionsWithPruning(_.containsPattern(PARSE_SQL)) {
+ case p: ParseSql if p.isConstantAnalyze => evaluate(p)
+ }
+ }
+
+ private def evaluate(p: ParseSql): Expression =
Review Comment:
Fixed in f9390c2225c. The rule now falls back only when the SQL argument
itself cannot be evaluated. An ANALYZE failure is reported in the JSON, and any
other exception propagates instead of being swallowed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala:
##########
@@ -64,6 +64,7 @@ class SparkOptimizer(
override def defaultBatches: Seq[Batch] = flattenBatches(Seq(
preOptimizationBatches,
+ Batch("Evaluate parse_sql", Once, EvaluateParseSql),
Review Comment:
Fixed in f9390c2225c. Evaluate parse_sql is now its own batch placed right
after Finish Analysis, so SpecialDatetimeValues and ComputeCurrentTime run
first. Covered by 'ANALYZE of a constant whose SQL text depends on a special
datetime value' (ANSI off, CAST('epoch' AS DATE)).
--
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]
