srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4487002141
##########
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(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ if (session.sessionState.catalog.isTempView(parts)) {
+ val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
+ val last = parts.last
+ if (parts.exists(_.equalsIgnoreCase(globalTemp))) {
+ globalTempIdent(last)
+ } else {
+ sessionIdent(last)
+ }
+ } else {
+ qualifyWithCurrentNamespace(parts, session)
+ }
+ }
+
+ private def qualifyRoutine(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ val catalog = session.sessionState.catalog
+ // Same order as FunctionResolution: `first` is session, builtin,
persistent;
+ // `second` (default) is builtin, session, persistent; `last` is builtin,
persistent, session.
+ val sessionOrder =
+
session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER)
+ def isTemp(name: String): Boolean =
+ catalog.isTemporaryFunction(FunctionIdentifier(name))
+ def isPersistent(name: String): Boolean =
+ catalog.isPersistentFunction(FunctionIdentifier(name))
+ def isBuiltin(name: String): Boolean = {
+ val id = FunctionIdentifier(name)
+ FunctionRegistry.builtin.functionExists(id) ||
+ TableFunctionRegistry.builtin.functionExists(id)
+ }
+ def isSessionBuiltin(name: String): Boolean = {
+ val id = FunctionRegistry.builtinFunctionIdentifier(name)
+ !isTemp(name) && !isBuiltin(name) &&
+ (session.sessionState.functionRegistry.functionExists(id) ||
+ session.sessionState.tableFunctionRegistry.functionExists(id))
+ }
+ def builtinIdent(name: String): Seq[String] =
+ FunctionRegistry.builtinFunctionIdentifier(name).nameParts
+ parts match {
+ case Seq(name) if sessionOrder == "first" && isTemp(name) =>
+ sessionIdent(name)
+ case Seq(name) if isBuiltin(name) || isSessionBuiltin(name) =>
+ builtinIdent(name)
+ case Seq(name) if sessionOrder == "last" && isPersistent(name) =>
Review Comment:
Fixed in f9390c2225c. Under sessionOrder=last, the persistent-function probe
now asks the current FunctionCatalog through functionExists when it is not the
session catalog. isPersistentFunction is used only for the session catalog. The
sessionOrder=last owner test covers the session case; the non-session catalog
probe has no test yet.
##########
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 =>
addTarget(c.multipartIdentifier)
case u: UnresolvedTableValuedFunction => addFunction(u.name)
+ case w: ReplaceData => rewriteTargets ++= rewriteTargetParts(w.table)
Review Comment:
Fixed in f9390c2225c. InsertOnlyMerge is a row-level rewrite now, so its
query reads the target as RewriteRead and the target is not reported as a
source. The rewrite filter matches the resolved identity parts of the target,
not the bare name, so a different table with the same name is not filtered.
Covered by 'ANALYZE of row-level DELETE and insert-only MERGE keeps real source
reads'.
##########
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(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ if (session.sessionState.catalog.isTempView(parts)) {
+ val globalTemp = SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
+ val last = parts.last
+ if (parts.exists(_.equalsIgnoreCase(globalTemp))) {
+ globalTempIdent(last)
+ } else {
+ sessionIdent(last)
+ }
+ } else {
+ qualifyWithCurrentNamespace(parts, session)
+ }
+ }
+
+ private def qualifyRoutine(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ val catalog = session.sessionState.catalog
+ // Same order as FunctionResolution: `first` is session, builtin,
persistent;
+ // `second` (default) is builtin, session, persistent; `last` is builtin,
persistent, session.
+ val sessionOrder =
+
session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER)
+ def isTemp(name: String): Boolean =
+ catalog.isTemporaryFunction(FunctionIdentifier(name))
+ def isPersistent(name: String): Boolean =
+ catalog.isPersistentFunction(FunctionIdentifier(name))
+ def isBuiltin(name: String): Boolean = {
+ val id = FunctionIdentifier(name)
+ FunctionRegistry.builtin.functionExists(id) ||
+ TableFunctionRegistry.builtin.functionExists(id)
+ }
+ def isSessionBuiltin(name: String): Boolean = {
+ val id = FunctionRegistry.builtinFunctionIdentifier(name)
+ !isTemp(name) && !isBuiltin(name) &&
+ (session.sessionState.functionRegistry.functionExists(id) ||
+ session.sessionState.tableFunctionRegistry.functionExists(id))
+ }
+ def builtinIdent(name: String): Seq[String] =
+ FunctionRegistry.builtinFunctionIdentifier(name).nameParts
+ parts match {
+ case Seq(name) if sessionOrder == "first" && isTemp(name) =>
+ sessionIdent(name)
+ case Seq(name) if isBuiltin(name) || isSessionBuiltin(name) =>
+ builtinIdent(name)
+ case Seq(name) if sessionOrder == "last" && isPersistent(name) =>
+ qualifyWithCurrentNamespace(parts, session)
+ case Seq(name) if isTemp(name) =>
+ sessionIdent(name)
+ case Seq(ns, name) if isSystemNamespace(ns) =>
+ Seq(CatalogManager.SYSTEM_CATALOG_NAME, ns.toLowerCase(Locale.ROOT),
name)
+ case Seq(cat, ns, name)
+ if cat.equalsIgnoreCase(CatalogManager.SYSTEM_CATALOG_NAME) &&
+ isSystemNamespace(ns) =>
+ Seq(CatalogManager.SYSTEM_CATALOG_NAME, ns.toLowerCase(Locale.ROOT),
name)
+ case other =>
+ qualifyWithCurrentNamespace(other, session)
+ }
+ }
+
+ /**
+ * Qualify `parts` using the current catalog/namespace without loading a
+ * catalog plugin. `isCatalogRegistered` instantiates the plugin.
+ */
+ private def qualifyWithCurrentNamespace(
+ parts: Seq[String],
+ session: ClassicSparkSession): Seq[String] = {
+ val mgr = session.sessionState.catalogManager
+ val catalogName = mgr.currentCatalog.name()
+ val namespace = mgr.currentNamespace.toSeq
+ parts match {
+ case Seq(name) => Seq(catalogName) ++ namespace :+ name
+ case other if isKnownCatalogName(other.head, session) => other
+ case other => catalogName +: other
+ }
+ }
+
+ private def isKnownCatalogName(
Review Comment:
Fixed in 59cccb92091. The current catalog is now compared exactly, since
custom catalogs are loaded by their exact name. The session catalog keeps the
case-insensitive match that CatalogManager.catalog uses. The conf lookup was
already exact.
##########
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 =>
Review Comment:
Fixed in f9390c2225c. The unreachable v1 CreateTable and PersistedView arms
and their imports are removed.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/expressions/ParseSql.scala:
##########
@@ -64,26 +94,66 @@ import org.apache.spark.unsafe.types.UTF8String
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 `sqlStmt` and `mode` must be constant
for the
+ query (DATATYPE_MISMATCH.NON_FOLDABLE_INPUT otherwise); a DataFrame column
of
+ statements can use PARSE only. Each JSON object has `start` (1-based
character
+ offset in the batch), `length`, and `parse_success`. Parse success also has
+ `statement_identifier` / `statement_code` (ISO/IEC 9075-2:2023 Table 39).
ANALYZE
+ adds `analyze_success`. Optional fields are omitted when empty:
+ `target_table_references`, `source_table_references`,
`function_references` (arrays
+ of name parts; ANALYZE qualifies persistent names to
catalog.schema.object, local
+ temp views and temp functions as `system.session`, builtins as
`system.builtin`;
+ CTE aliases in scope are not listed as tables), `select_list` (`{name}`
parts;
+ ANALYZE lists the output columns, with `*` expanded, and adds `type` as a
+ DESCRIBE TABLE EXTENDED AS JSON type object), `parameter_markers` (`named`
and/or
+ `unnamed_count`), and `error` (STANDARD format). Nested error locations are
+ statement-relative Unicode code points. Later statements are still
processed
+ after a per-statement parse or analysis failure. ANALYZE of a SQL script
reports
+ `analyze_success` false without executing the script. Path-based relations
keep
+ `[format, path]` as written. Analysis may still run time-travel subqueries,
+ infer file schemas, and call Python UDTF `analyze`.
""",
group = "misc_funcs",
since = "4.4.0")
// scalastyle:on nonascii
// scalastyle:on line.size.limit
-case class ParseSql(child: Expression)
- extends UnaryExpression
+case class ParseSql(
+ sqlStmt: Expression,
+ mode: Expression)
+ extends Expression
with ImplicitCastInputTypes
- with CodegenFallback {
+ with CodegenFallback
+ with QueryErrorsBase {
+
+ def this(sqlStmt: Expression) = this(sqlStmt, Literal("PARSE"))
override def prettyName: String = "parse_sql"
- override def nullable: Boolean = true
+ override def children: Seq[Expression] = Seq(sqlStmt, mode)
+
+ override def nullable: Boolean = sqlStmt.nullable || mode.nullable
+
+ // Foldable so constant SQL is evaluated on the driver, where ANALYZE can
see the session.
Review Comment:
Fixed in f9390c2225c. The comment now says that foldable lets
ConstantFolding fold constant PARSE calls, and that EvaluateParseSql evaluates
constant ANALYZE calls. The usage text is unchanged, since it feeds the
generated docs and golden output.
--
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]
