[SPARK-30799][SQL] "spark_catalog.t" should not be resolved to temp view #27550
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier} | |
| import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogStorageFormat, CatalogTable, CatalogTableType, CatalogUtils} | ||
| import org.apache.spark.sql.catalyst.plans.logical._ | ||
| import org.apache.spark.sql.catalyst.rules.Rule | ||
| import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, CatalogV2Util, LookupCatalog, SupportsNamespaces, TableCatalog, TableChange, V1Table} | ||
| import org.apache.spark.sql.connector.expressions.Transform | ||
| import org.apache.spark.sql.execution.command._ | ||
| import org.apache.spark.sql.execution.datasources.{CreateTable, DataSource, RefreshTable} | ||
|
|
@@ -39,7 +39,8 @@ import org.apache.spark.sql.types.{HIVE_TYPE_STRING, HiveStringType, MetadataBui | |
| class ResolveSessionCatalog( | ||
| val catalogManager: CatalogManager, | ||
| conf: SQLConf, | ||
| isTempView: Seq[String] => Boolean, | ||
| isTempFunction: String => Boolean) | ||
| extends Rule[LogicalPlan] with LookupCatalog { | ||
| import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ | ||
| import org.apache.spark.sql.connector.catalog.CatalogV2Util._ | ||
|
|
@@ -215,8 +216,9 @@ class ResolveSessionCatalog( | |
| } | ||
| AlterDatabaseSetLocationCommand(ns.head, location) | ||
|
|
||
| // v1 RENAME TABLE supports temp view. | ||
| case RenameTableStatement(TempViewOrV1Table(oldName), newName, isView) => | ||
| AlterTableRenameCommand(oldName.asTableIdentifier, newName.asTableIdentifier, isView) | ||
|
|
||
| case DescribeRelation(ResolvedTable(_, ident, _: V1Table), partitionSpec, isExtended) => | ||
| DescribeTableCommand(ident.asTableIdentifier, partitionSpec, isExtended) | ||
|
|
@@ -228,10 +230,12 @@ class ResolveSessionCatalog( | |
| case DescribeColumnStatement( | ||
| SessionCatalogAndTable(catalog, tbl), colNameParts, isExtended) => | ||
| loadTable(catalog, tbl.asIdentifier).collect { | ||
| // `V1Table` also includes permanent views. | ||
| case v1Table: V1Table => | ||
| DescribeColumnCommand(tbl.asTableIdentifier, colNameParts, isExtended) | ||
| }.getOrElse { | ||
| if (isTempView(tbl)) { | ||
|
Contributor
There was a problem hiding this comment. Since
Contributor
Author
There was a problem hiding this comment. You have a good point here. I think this is an existing bug. Here we look up table first (call
Contributor
There was a problem hiding this comment. Yes, will do after this PR. |
||
| // v1 DESCRIBE COLUMN supports temp view. | ||
| DescribeColumnCommand(tbl.asTableIdentifier, colNameParts, isExtended) | ||
| } else { | ||
| throw new AnalysisException("Describing columns is not supported for v2 tables.") | ||
|
|
@@ -279,8 +283,9 @@ class ResolveSessionCatalog( | |
| ignoreIfExists = c.ifNotExists) | ||
| } | ||
|
|
||
| // v1 REFRESH TABLE supports temp view. | ||
| case RefreshTableStatement(TempViewOrV1Table(name)) => | ||
| RefreshTable(name.asTableIdentifier) | ||
|
|
||
| // For REPLACE TABLE [AS SELECT], we should fail if the catalog is resolved to the | ||
| // session catalog and the table provider is not v2. | ||
|
|
@@ -315,11 +320,13 @@ class ResolveSessionCatalog( | |
| orCreate = c.orCreate) | ||
| } | ||
|
|
||
| // v1 DROP TABLE supports temp view. | ||
| case DropTableStatement(TempViewOrV1Table(name), ifExists, purge) => | ||
| DropTableCommand(name.asTableIdentifier, ifExists, isView = false, purge = purge) | ||
|
|
||
| // v1 DROP TABLE supports temp view. | ||
| case DropViewStatement(TempViewOrV1Table(name), ifExists) => | ||
| DropTableCommand(name.asTableIdentifier, ifExists, isView = true, purge = false) | ||
|
|
||
| case c @ CreateNamespaceStatement(CatalogAndNamespace(catalog, ns), _, _) | ||
| if isSessionCatalog(catalog) => | ||
|
|
@@ -393,12 +400,18 @@ class ResolveSessionCatalog( | |
| ShowCreateTableAsSerdeCommand(v1TableName.asTableIdentifier) | ||
|
|
||
| case CacheTableStatement(tbl, plan, isLazy, options) => | ||
| val name = if (plan.isDefined) { | ||
| // CACHE TABLE ... AS SELECT creates a temp view with the input query. | ||
| // Temp view doesn't belong to any catalog and we shouldn't resolve catalog in the name. | ||
| tbl | ||
| } else { | ||
| parseTempViewOrV1Table(tbl, "CACHE TABLE") | ||
| } | ||
| CacheTableCommand(name.asTableIdentifier, plan, isLazy, options) | ||
|
|
||
| case UncacheTableStatement(tbl, ifExists) => | ||
| val name = parseTempViewOrV1Table(tbl, "UNCACHE TABLE") | ||
| UncacheTableCommand(name.asTableIdentifier, ifExists) | ||
|
|
||
| case TruncateTableStatement(tbl, partitionSpec) => | ||
| val v1TableName = parseV1Table(tbl, "TRUNCATE TABLE") | ||
|
|
@@ -427,10 +440,6 @@ class ResolveSessionCatalog( | |
| throw new AnalysisException( | ||
| s"Namespace name should have only one part if specified: ${ns.get.quoted}") | ||
| } | ||
|
||
| ShowColumnsCommand(db, v1TableName) | ||
|
|
||
| case AlterTableRecoverPartitionsStatement(tbl) => | ||
|
|
@@ -470,18 +479,23 @@ class ResolveSessionCatalog( | |
| serdeProperties, | ||
| partitionSpec) | ||
|
|
||
| case AlterViewAsStatement(name, originalText, query) => | ||
| val viewName = parseTempViewOrV1Table(name, "ALTER VIEW QUERY") | ||
| AlterViewAsCommand( | ||
| viewName.asTableIdentifier, | ||
| originalText, | ||
| query) | ||
|
|
||
| case CreateViewStatement( | ||
| tbl, userSpecifiedColumns, comment, properties, | ||
| originalText, child, allowExisting, replace, viewType) => | ||
|
|
||
| val v1TableName = if (viewType != PersistedView) { | ||
| // temp view doesn't belong to any catalog and we shouldn't resolve catalog in the name. | ||
| tbl | ||
| } else { | ||
| parseV1Table(tbl, "CREATE VIEW") | ||
| } | ||
| CreateViewCommand( | ||
| v1TableName.asTableIdentifier, | ||
| userSpecifiedColumns, | ||
|
|
@@ -496,56 +510,94 @@ class ResolveSessionCatalog( | |
| case ShowTableProperties(r: ResolvedTable, propertyKey) if isSessionCatalog(r.catalog) => | ||
| ShowTablePropertiesCommand(r.identifier.asTableIdentifier, propertyKey) | ||
|
|
||
| case DescribeFunctionStatement(nameParts, extended) => | ||
| val functionIdent = | ||
| parseSessionCatalogFunctionIdentifier(nameParts, "DESCRIBE FUNCTION") | ||
| DescribeFunctionCommand(functionIdent, extended) | ||
|
|
||
| case ShowFunctionsStatement(userScope, systemScope, pattern, fun) => | ||
| val (database, function) = fun match { | ||
| case Some(nameParts) => | ||
| val FunctionIdentifier(fn, db) = | ||
| parseSessionCatalogFunctionIdentifier(nameParts, "SHOW FUNCTIONS") | ||
| (db, Some(fn)) | ||
| case None => (None, pattern) | ||
| } | ||
| ShowFunctionsCommand(database, function, userScope, systemScope) | ||
|
|
||
| case DropFunctionStatement(nameParts, ifExists, isTemp) => | ||
| val FunctionIdentifier(function, database) = | ||
| parseSessionCatalogFunctionIdentifier(nameParts, "DROP FUNCTION") | ||
| DropFunctionCommand(database, function, ifExists, isTemp) | ||
|
|
||
| case CreateFunctionStatement(nameParts, | ||
| className, resources, isTemp, ignoreIfExists, replace) => | ||
| if (isTemp) { | ||
| // temp func doesn't belong to any catalog and we shouldn't resolve catalog in the name. | ||
| val database = if (nameParts.length > 2) { | ||
| throw new AnalysisException(s"Unsupported function name '${nameParts.quoted}'") | ||
|
Contributor
Author
There was a problem hiding this comment. |
||
| } else if (nameParts.length == 2) { | ||
| Some(nameParts.head) | ||
| } else { | ||
| None | ||
| } | ||
| CreateFunctionCommand( | ||
| database, | ||
| nameParts.last, | ||
| className, | ||
| resources, | ||
| isTemp, | ||
| ignoreIfExists, | ||
| replace) | ||
| } else { | ||
| val FunctionIdentifier(function, database) = | ||
| parseSessionCatalogFunctionIdentifier(nameParts, "CREATE FUNCTION") | ||
| CreateFunctionCommand(database, function, className, resources, isTemp, ignoreIfExists, | ||
| replace) | ||
| } | ||
| } | ||
|
|
||
| // TODO: move function related v2 statements to the new framework. | ||
| private def parseSessionCatalogFunctionIdentifier( | ||
| nameParts: Seq[String], | ||
| sql: String): FunctionIdentifier = { | ||
| if (nameParts.length == 1 && isTempFunction(nameParts.head)) { | ||
| return FunctionIdentifier(nameParts.head) | ||
| } | ||
|
|
||
| nameParts match { | ||
| case SessionCatalogAndIdentifier(_, ident) => | ||
| if (nameParts.length == 1) { | ||
| // If there is only one name part, it means the current catalog is the session catalog. | ||
| // Here we don't fill the default database, to keep the error message unchanged for | ||
| // v1 commands. | ||
| FunctionIdentifier(nameParts.head, None) | ||
| } else { | ||
| ident.namespace match { | ||
| // For name parts like `spark_catalog.t`, we need to fill in the default database so | ||
| // that the caller side won't treat it as a temp function. | ||
| case Array() if nameParts.head == CatalogManager.SESSION_CATALOG_NAME => | ||
| FunctionIdentifier( | ||
| ident.name, Some(catalogManager.v1SessionCatalog.getCurrentDatabase)) | ||
| case Array(db) => FunctionIdentifier(ident.name, Some(db)) | ||
| case _ => | ||
| throw new AnalysisException(s"Unsupported function name '$ident'") | ||
| } | ||
| } | ||
|
|
||
| case _ => throw new AnalysisException(s"$sql is only supported in v1 catalog") | ||
| } | ||
| } | ||
|
|
||
| private def parseV1Table(tableName: Seq[String], sql: String): Seq[String] = tableName match { | ||
| case SessionCatalogAndTable(_, tbl) => tbl | ||
| case _ => throw new AnalysisException(s"$sql is only supported with v1 tables.") | ||
| } | ||
|
|
||
| private def parseTempViewOrV1Table( | ||
| nameParts: Seq[String], sql: String): Seq[String] = nameParts match { | ||
| case TempViewOrV1Table(name) => name | ||
| case _ => throw new AnalysisException(s"$sql is only supported with temp views or v1 tables.") | ||
| } | ||
|
|
||
| private def buildCatalogTable( | ||
|
|
@@ -584,7 +636,29 @@ class ResolveSessionCatalog( | |
| object SessionCatalogAndTable { | ||
| def unapply(nameParts: Seq[String]): Option[(CatalogPlugin, Seq[String])] = nameParts match { | ||
| case SessionCatalogAndIdentifier(catalog, ident) => | ||
| if (nameParts.length == 1) { | ||
| // If there is only one name part, it means the current catalog is the session catalog. | ||
| // Here we return the original name part, to keep the error message unchanged for | ||
| // v1 commands. | ||
|
Contributor
Author
There was a problem hiding this comment. I can remove this, but then I need to update many tests slightly to update the error message. We can leave it to 3.1. |
||
| Some(catalog -> nameParts) | ||
| } else { | ||
| Some(catalog -> ident.asMultipartIdentifier) | ||
| } | ||
| case _ => None | ||
| } | ||
| } | ||
|
|
||
| object TempViewOrV1Table { | ||
| def unapply(nameParts: Seq[String]): Option[Seq[String]] = nameParts match { | ||
| case _ if isTempView(nameParts) => Some(nameParts) | ||
| case SessionCatalogAndTable(_, tbl) => | ||
| if (nameParts.head == CatalogManager.SESSION_CATALOG_NAME && tbl.length == 1) { | ||
| // For name parts like `spark_catalog.t`, we need to fill in the default database so | ||
|
Contributor
There was a problem hiding this comment. Unrelated question, what was the reason to allow
Contributor
Author
There was a problem hiding this comment. I noticed this well and think this is not intentional. But this is a different topic and we have many tests using We can open another PR to forbid it if we think we shouldn't support this feature.
Contributor
There was a problem hiding this comment. Got it, thanks for the explanation. It makes more sense to forbid it to keep the behavior consistent. |
||
| // that the caller side won't treat it as a temp view. | ||
| Some(Seq(catalogManager.v1SessionCatalog.getCurrentDatabase, tbl.head)) | ||
| } else { | ||
| Some(tbl) | ||
| } | ||
| case _ => None | ||
| } | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The term
Catalogis being used in many places. For example, our internal SessionCatalog also manages temp objects.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We can move them out and put it in a temp view manager like the
GlobalTempViewManager. I'm talking more about the theory.