srielau commented on code in PR #59873:
URL: https://github.com/apache/spark/pull/59873#discussion_r4479045586
##########
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(":")
+ case Seq(path) => path.contains("/") || path.startsWith("file:")
+ 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
+ val tempFirst =
+
session.sessionState.conf.getConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER) ==
"first"
+ def isTemp(name: String): Boolean =
+ catalog.isTemporaryFunction(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 tempFirst && isTemp(name) =>
+ sessionIdent(name)
+ case Seq(name) if isBuiltin(name) || isSessionBuiltin(name) =>
+ builtinIdent(name)
+ 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(
+ name: String,
+ session: ClassicSparkSession): Boolean = {
+ val mgr = session.sessionState.catalogManager
+ name.equalsIgnoreCase(CatalogManager.SESSION_CATALOG_NAME) ||
+ name.equalsIgnoreCase(mgr.currentCatalog.name()) ||
+ session.sessionState.conf.getAllConfs.exists { case (key, _) =>
+ key.equalsIgnoreCase("spark.sql.catalog." + name)
+ }
+ }
+
+ private def isSystemNamespace(ns: String): Boolean =
+ ns.equalsIgnoreCase(CatalogManager.BUILTIN_NAMESPACE) ||
+ ns.equalsIgnoreCase(CatalogManager.SESSION_NAMESPACE)
+
private def primaryQueryPlan(plan: LogicalPlan): LogicalPlan = plan match {
case UnresolvedWith(child, _, _) => primaryQueryPlan(child)
+ case WithCTE(child, _) => primaryQueryPlan(child)
case UnresolvedInsert(_, _, _, query, _, _, _, _, _) =>
primaryQueryPlan(query)
- case InsertIntoStatement(_, _, _, query, _, _, _, _, _) =>
- primaryQueryPlan(query)
+ case i: InsertIntoStatement => primaryQueryPlan(i.query)
+ case d: DataWritingCommand => primaryQueryPlan(d.query)
+ // Row-level rewrites add internal columns; leave them out of select_list.
+ case _: ReplaceData | _: WriteDelta => plan
Review Comment:
Added in 6fc55256046: `CHAR(3)` and `VARCHAR(10)` columns report `char` /
`varchar` with their lengths in select_list, and row-level DELETE, UPDATE and
MERGE on `InMemoryRowLevelOperationTableCatalog` report no select_list and no
rewrite columns.
--
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]
