[SPARK-34546][SQL] AlterViewAs.query should be analyzed during the analysis phase, and AlterViewAs should invalidate the cache #31652
There are no files selected for viewing
This file contains 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 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 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 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 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 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 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 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 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,8 +22,9 @@ import scala.collection.mutable | |
| import org.json4s.JsonAST.{JArray, JString} | ||
| import org.json4s.jackson.JsonMethods._ | ||
|
|
||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.sql.{AnalysisException, Row, SparkSession} | ||
| import org.apache.spark.sql.catalyst.{FunctionIdentifier, SQLConfHelper, TableIdentifier} | ||
| import org.apache.spark.sql.catalyst.analysis.{GlobalTempView, LocalTempView, PersistedView, ViewType} | ||
| import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, SessionCatalog, TemporaryViewRelation} | ||
| import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, SubqueryExpression, UserDefinedExpression} | ||
|
|
@@ -115,48 +116,27 @@ case class CreateViewCommand( | |
|
|
||
| if (viewType == LocalTempView) { | ||
| val aliasedPlan = aliasPlan(sparkSession, analyzedPlan) | ||
| val tableDefinition = createTemporaryViewRelation( | ||
| name, | ||
| sparkSession, | ||
| replace, | ||
| catalog.getRawTempView, | ||
| originalText, | ||
| analyzedPlan, | ||
| aliasedPlan) | ||
| catalog.createTempView(name.table, tableDefinition, overrideIfExists = replace) | ||
| } else if (viewType == GlobalTempView) { | ||
| val db = sparkSession.sessionState.conf.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE) | ||
| val viewIdent = TableIdentifier(name.table, Option(db)) | ||
| val aliasedPlan = aliasPlan(sparkSession, analyzedPlan) | ||
| val tableDefinition = createTemporaryViewRelation( | ||
| viewIdent, | ||
| sparkSession, | ||
| replace, | ||
| catalog.getRawGlobalTempView, | ||
| originalText, | ||
| analyzedPlan, | ||
| aliasedPlan) | ||
| catalog.createGlobalTempView(name.table, tableDefinition, overrideIfExists = replace) | ||
| } else if (catalog.tableExists(name)) { | ||
| val tableMetadata = catalog.getTableMetadata(name) | ||
|
|
@@ -192,20 +172,6 @@ case class CreateViewCommand( | |
| Seq.empty[Row] | ||
| } | ||
|
|
||
| /** | ||
| * If `userSpecifiedColumns` is defined, alias the analyzed plan to the user specified columns, | ||
| * else return the analyzed plan directly. | ||
|
|
@@ -274,28 +240,29 @@ case class AlterViewAsCommand( | |
| override def innerChildren: Seq[QueryPlan[_]] = Seq(query) | ||
|
|
||
| override def run(session: SparkSession): Seq[Row] = { | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment. shall we remove the similar code in There was a problem hiding this comment. We cannot yet because of the following: Here, |
||
| if (session.sessionState.catalog.isTempView(name)) { | ||
| alterTemporaryView(session, query) | ||
| } else { | ||
| alterPermanentView(session, query) | ||
| } | ||
| Seq.empty[Row] | ||
| } | ||
|
|
||
| private def alterTemporaryView(session: SparkSession, analyzedPlan: LogicalPlan): Unit = { | ||
| val catalog = session.sessionState.catalog | ||
| val getRawTempView: String => Option[LogicalPlan] = if (name.database.isEmpty) { | ||
| catalog.getRawTempView | ||
| } else { | ||
| catalog.getRawGlobalTempView | ||
| } | ||
| val tableDefinition = createTemporaryViewRelation( | ||
| name, | ||
| session, | ||
| replace = true, | ||
| getRawTempView, | ||
| Some(originalText), | ||
| analyzedPlan, | ||
| aliasedPlan = analyzedPlan) | ||
| session.sessionState.catalog.alterTempViewDefinition(name, tableDefinition) | ||
| } | ||
|
|
||
|
|
@@ -306,6 +273,9 @@ case class AlterViewAsCommand( | |
| val viewIdent = viewMeta.identifier | ||
| checkCyclicViewReference(analyzedPlan, Seq(viewIdent), viewIdent) | ||
|
|
||
| logDebug(s"Try to uncache ${viewIdent.quotedString} before replacing.") | ||
| CommandUtils.uncacheTableOrView(session, viewIdent.quotedString) | ||
|
|
||
| val newProperties = generateViewProperties( | ||
| viewMeta.properties, session, analyzedPlan, analyzedPlan.schema.fieldNames) | ||
|
|
||
|
|
@@ -349,7 +319,7 @@ case class ShowViewsCommand( | |
| } | ||
| } | ||
|
|
||
| object ViewHelper extends SQLConfHelper with Logging { | ||
|
|
||
| private val configPrefixDenyList = Seq( | ||
| SQLConf.MAX_NESTED_VIEW_DEPTH.key, | ||
|
|
@@ -596,19 +566,80 @@ object ViewHelper { | |
| (collectTempViews(child), collectTempFunctions(child)) | ||
| } | ||
|
|
||
| /** | ||
| * Returns a [[TemporaryViewRelation]] that contains information about a temporary view | ||
| * to create, given an analyzed plan of the view. If a temp view is to be replaced and it is | ||
| * cached, it will be uncached before being replaced. | ||
| * | ||
| * @param name the name of the temporary view to create/replace. | ||
| * @param session the spark session. | ||
| * @param replace if true and the existing view is cached, it will be uncached. | ||
| * @param getRawTempView the function that returns an optional raw plan of the local or | ||
| * global temporary view. | ||
| * @param originalText the original SQL text of this view, can be None if this view is created via | ||
| * Dataset API or spark.sql.legacy.storeAnalyzedPlanForView is set to true. | ||
| * @param analyzedPlan the logical plan that represents the view; this is used to generate the | ||
| * logical plan for temporary view and the view schema. | ||
| * @param aliasedPlan the aliased logical plan based on the user specified columns. If there are | ||
| * no user specified plans, this should be same as `analyzedPlan`. | ||
| */ | ||
| def createTemporaryViewRelation( | ||
| name: TableIdentifier, | ||
| session: SparkSession, | ||
| replace: Boolean, | ||
| getRawTempView: String => Option[LogicalPlan], | ||
| originalText: Option[String], | ||
| analyzedPlan: LogicalPlan, | ||
| aliasedPlan: LogicalPlan): TemporaryViewRelation = { | ||
| val uncache = getRawTempView(name.table).map { r => | ||
| needsToUncache(r, aliasedPlan) | ||
| }.getOrElse(false) | ||
| if (replace && uncache) { | ||
| logDebug(s"Try to uncache ${name.quotedString} before replacing.") | ||
| checkCyclicViewReference(analyzedPlan, Seq(name), name) | ||
| CommandUtils.uncacheTableOrView(session, name.quotedString) | ||
| } | ||
| if (!conf.storeAnalyzedPlanForView && originalText.nonEmpty) { | ||
| TemporaryViewRelation( | ||
| prepareTemporaryView( | ||
| name, | ||
| session, | ||
| analyzedPlan, | ||
| aliasedPlan.schema, | ||
| originalText.get)) | ||
| } else { | ||
| TemporaryViewRelation( | ||
| prepareTemporaryViewStoringAnalyzedPlan(name, aliasedPlan), | ||
| Some(aliasedPlan)) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Checks if need to uncache the temp view being replaced. | ||
| */ | ||
| private def needsToUncache( | ||
| rawTempView: LogicalPlan, | ||
| aliasedPlan: LogicalPlan): Boolean = rawTempView match { | ||
| // If TemporaryViewRelation doesn't store the analyzed view, always uncache. | ||
| case TemporaryViewRelation(_, None) => true | ||
|
There was a problem hiding this comment. This is unnecessary because |
||
| // Do not need to uncache if the to-be-replaced temp view plan and the new plan are the | ||
| // same-result plans. | ||
| case TemporaryViewRelation(_, Some(p)) => !p.sameResult(aliasedPlan) | ||
| case p => !p.sameResult(aliasedPlan) | ||
|
There was a problem hiding this comment. Do we still need it? There was a problem hiding this comment. We need to first update , which I will get to right after this PR. Once the PR is done, we can update Let me actually create a JIRA to capture these as subtasks. There was a problem hiding this comment. I created https://issues.apache.org/jira/browse/SPARK-34698 to track these. |
||
| } | ||
|
|
||
| /** | ||
| * Returns a [[CatalogTable]] that contains information for temporary view. | ||
| * Generate the view-specific properties(e.g. view default database, view query output | ||
| * column names) and store them as properties in the CatalogTable, and also creates | ||
| * the proper schema for the view. | ||
| */ | ||
| private def prepareTemporaryView( | ||
| viewName: TableIdentifier, | ||
| session: SparkSession, | ||
| analyzedPlan: LogicalPlan, | ||
| viewSchema: StructType, | ||
| originalText: String): CatalogTable = { | ||
|
|
||
| val catalog = session.sessionState.catalog | ||
| val (tempViews, tempFunctions) = collectTemporaryObjects(catalog, analyzedPlan) | ||
|
|
@@ -622,22 +653,22 @@ object ViewHelper { | |
| tableType = CatalogTableType.VIEW, | ||
| storage = CatalogStorageFormat.empty, | ||
| schema = viewSchema, | ||
| viewText = Some(originalText), | ||
| properties = newProperties) | ||
| } | ||
|
|
||
| /** | ||
| * Returns a [[CatalogTable]] that contains information for the temporary view storing | ||
| * an analyzed plan. | ||
| */ | ||
| private def prepareTemporaryViewStoringAnalyzedPlan( | ||
| viewName: TableIdentifier, | ||
| analyzedPlan: LogicalPlan): CatalogTable = { | ||
| CatalogTable( | ||
| identifier = viewName, | ||
| tableType = CatalogTableType.VIEW, | ||
| storage = CatalogStorageFormat.empty, | ||
| schema = analyzedPlan.schema, | ||
| properties = Map((VIEW_STORING_ANALYZED_PLAN, "true"))) | ||
| } | ||
| } | ||
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.
I renamed because this type of view can also be created not from a dataframe - e.g.,
ALTER VIEW ... ASwithspark.sql.legacy.storeAnalyzedPlanForViewset to true.