-
Notifications
You must be signed in to change notification settings - Fork 29.3k
[SPARK-34152][SQL] Make CreateViewStatement.child to be LogicalPlan's children so that it's resolved in analyze phase #31273
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 37 commits
db4dfaf
71c01e8
8dc1961
7becf96
4b3f184
9085d17
f2b86a9
bfabe9f
1f2e4c7
0b75fdc
ac663aa
30e750c
0fbf1cf
a8581c6
4e77301
60cf521
a0ba508
1949cbc
5e38e2b
388ec16
7e92eeb
fbdcf3b
2a3e5b7
e92cbe1
8422ae8
bace91a
9432fec
737307c
d906e19
4e55a1f
e9cd2f3
9da3b9c
e36fc34
b54df8a
c5651b2
d4e958a
84d817c
82d58ba
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -861,7 +861,20 @@ class Analyzer(override val catalogManager: CatalogManager) | |
| } | ||
|
|
||
| private def isResolvingView: Boolean = AnalysisContext.get.catalogAndNamespace.nonEmpty | ||
| private def referredTempViewNames: Seq[Seq[String]] = AnalysisContext.get.referredTempViewNames | ||
| private def isReferredTempViewName(nameParts: Seq[String]): Boolean = { | ||
| AnalysisContext.get.referredTempViewNames.exists { n => | ||
| (n.length == nameParts.length) && n.zip(nameParts).forall { | ||
| case (a, b) => resolver(a, b) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def getTempViewRawPlan(plan: LogicalPlan): LogicalPlan = { | ||
| EliminateSubqueryAliases(plan) match { | ||
| case v: View if v.isDataFrameTempView => v.child | ||
| case other => other | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Resolve relations to temp views. This is not an actual rule, and is called by | ||
|
|
@@ -887,7 +900,7 @@ class Analyzer(override val catalogManager: CatalogManager) | |
| case write: V2WriteCommand => | ||
| write.table match { | ||
| case UnresolvedRelation(ident, _, false) => | ||
| lookupTempView(ident, performCheck = true).map(EliminateSubqueryAliases(_)).map { | ||
| lookupTempView(ident, performCheck = true).map(getTempViewRawPlan).map { | ||
| case r: DataSourceV2Relation => write.withNewTable(r) | ||
| case _ => throw QueryCompilationErrors.writeIntoTempViewNotAllowedError(ident.quoted) | ||
| }.getOrElse(write) | ||
|
|
@@ -924,7 +937,7 @@ class Analyzer(override val catalogManager: CatalogManager) | |
| isStreaming: Boolean = false, | ||
| performCheck: Boolean = false): Option[LogicalPlan] = { | ||
| // Permanent View can't refer to temp views, no need to lookup at all. | ||
| if (isResolvingView && !referredTempViewNames.contains(identifier)) return None | ||
| if (isResolvingView && !isReferredTempViewName(identifier)) return None | ||
|
|
||
| val tmpView = identifier match { | ||
| case Seq(part1) => v1SessionCatalog.lookupTempView(part1) | ||
|
|
@@ -942,7 +955,7 @@ class Analyzer(override val catalogManager: CatalogManager) | |
| // If we are resolving relations insides views, we need to expand single-part relation names with | ||
| // the current catalog and namespace of when the view was created. | ||
| private def expandRelationName(nameParts: Seq[String]): Seq[String] = { | ||
| if (!isResolvingView || referredTempViewNames.contains(nameParts)) return nameParts | ||
| if (!isResolvingView || isReferredTempViewName(nameParts)) return nameParts | ||
|
|
||
| if (nameParts.length == 1) { | ||
| AnalysisContext.get.catalogAndNamespace :+ nameParts.head | ||
|
|
@@ -1139,7 +1152,10 @@ class Analyzer(override val catalogManager: CatalogManager) | |
| case other => other | ||
| } | ||
|
|
||
| EliminateSubqueryAliases(relation) match { | ||
| // Inserting into a file-based temporary view is allowed. | ||
| // (e.g., spark.read.parquet("path").createOrReplaceTempView("t"). | ||
| // Thus, we need to look at the raw plan of a temporary view. | ||
| getTempViewRawPlan(relation) match { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah sorry I missed this. The relation here can be a table relation, not temp view, so We can fix it in followup if there are no other comments to address. |
||
| case v: View => | ||
| throw QueryCompilationErrors.insertIntoViewNotAllowedError(v.desc.identifier, table) | ||
| case other => i.copy(table = other) | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -622,8 +622,7 @@ class SessionCatalog( | |
| } | ||
|
|
||
| /** | ||
| * Generate a [[View]] operator from the view description if the view stores sql text, | ||
| * otherwise, it is same to `getRawTempView` | ||
| * Generate a [[View]] operator from the temporary view stored. | ||
|
imback82 marked this conversation as resolved.
|
||
| */ | ||
| def getTempView(name: String): Option[LogicalPlan] = synchronized { | ||
| getRawTempView(name).map(getTempViewPlan) | ||
|
|
@@ -641,8 +640,7 @@ class SessionCatalog( | |
| } | ||
|
|
||
| /** | ||
| * Generate a [[View]] operator from the view description if the view stores sql text, | ||
| * otherwise, it is same to `getRawGlobalTempView` | ||
| * Generate a [[View]] operator from the global temporary view stored. | ||
| */ | ||
| def getGlobalTempView(name: String): Option[LogicalPlan] = { | ||
| getRawGlobalTempView(name).map(getTempViewPlan) | ||
|
|
@@ -683,7 +681,7 @@ class SessionCatalog( | |
| val table = formatTableName(name.table) | ||
| if (name.database.isEmpty) { | ||
| tempViews.get(table).map { | ||
| case TemporaryViewRelation(metadata) => metadata | ||
| case TemporaryViewRelation(metadata, _) => metadata | ||
| case plan => | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If https://github.com/apache/spark/pull/31273/files#r580757641 is done, |
||
| CatalogTable( | ||
| identifier = TableIdentifier(table), | ||
|
|
@@ -693,7 +691,7 @@ class SessionCatalog( | |
| }.getOrElse(getTableMetadata(name)) | ||
| } else if (formatDatabaseName(name.database.get) == globalTempViewManager.database) { | ||
| globalTempViewManager.get(table).map { | ||
| case TemporaryViewRelation(metadata) => metadata | ||
| case TemporaryViewRelation(metadata, _) => metadata | ||
| case plan => | ||
| CatalogTable( | ||
| identifier = TableIdentifier(table, Some(globalTempViewManager.database)), | ||
|
|
@@ -838,9 +836,11 @@ class SessionCatalog( | |
|
|
||
| private def getTempViewPlan(plan: LogicalPlan): LogicalPlan = { | ||
| plan match { | ||
| case viewInfo: TemporaryViewRelation => | ||
| fromCatalogTable(viewInfo.tableMeta, isTempView = true) | ||
| case v => v | ||
| case TemporaryViewRelation(tableMeta, None) => | ||
| fromCatalogTable(tableMeta, isTempView = true) | ||
| case TemporaryViewRelation(tableMeta, Some(plan)) => | ||
| View(desc = tableMeta, isTempView = true, child = plan) | ||
| case other => other | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If the current PR's approach is fine, we can remove this
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. +1 |
||
| } | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.plans.logical | |
| import org.apache.spark.sql.catalyst.AliasIdentifier | ||
| import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation | ||
| import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable} | ||
| import org.apache.spark.sql.catalyst.catalog.CatalogTable.VIEW_CREATED_FROM_DATAFRAME | ||
| import org.apache.spark.sql.catalyst.expressions._ | ||
| import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression | ||
| import org.apache.spark.sql.catalyst.plans._ | ||
|
|
@@ -443,21 +444,25 @@ case class InsertIntoDir( | |
| } | ||
|
|
||
| /** | ||
| * A container for holding the view description(CatalogTable), and the output of the view. The | ||
| * child should be a logical plan parsed from the `CatalogTable.viewText`, should throw an error | ||
| * if the `viewText` is not defined. | ||
| * A container for holding the view description(CatalogTable) and info whether the view is temporary | ||
| * or not. If the view description is available, the child should be a logical plan parsed from the | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| * `CatalogTable.viewText`. Otherwise, the view is a temporary one created from a dataframe and the | ||
| * view description should contain a `VIEW_CREATED_FROM_DATAFRAME` property; in this case, the child | ||
| * must be already resolved. | ||
| * | ||
| * This operator will be removed at the end of analysis stage. | ||
| * | ||
| * @param desc A view description(CatalogTable) that provides necessary information to resolve the | ||
| * view. | ||
| * we are able to decouple the output from the underlying structure. | ||
| * @param child The logical plan of a view operator, it should be a logical plan parsed from the | ||
| * `CatalogTable.viewText`, should throw an error if the `viewText` is not defined. | ||
| * @param isTempView A flag to indicate whether the view is temporary or not. | ||
| * @param child The logical plan of a view operator. If the view description is available, it should | ||
| * be a logical plan parsed from the `CatalogTable.viewText`. | ||
| */ | ||
| case class View( | ||
| desc: CatalogTable, | ||
| isTempView: Boolean, | ||
| child: LogicalPlan) extends UnaryNode { | ||
| require(!isDataFrameTempView || child.resolved) | ||
|
|
||
| override def output: Seq[Attribute] = child.output | ||
|
|
||
|
|
@@ -470,6 +475,9 @@ case class View( | |
| case _ => child.canonicalized | ||
| } | ||
|
|
||
| def isDataFrameTempView: Boolean = | ||
| isTempView && desc.properties.contains(VIEW_CREATED_FROM_DATAFRAME) | ||
|
|
||
| // When resolving a SQL view, we use an extra Project to add cast and alias to make sure the view | ||
| // output schema doesn't change even if the table referenced by the view is changed after view | ||
| // creation. We should remove this extra Project during canonicalize if it does nothing. | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -59,6 +59,28 @@ trait AnalysisTest extends PlanTest { | |
| } | ||
| } | ||
|
|
||
| protected def checkAnalysisWithTransform( | ||
| inputPlan: LogicalPlan, | ||
| expectedPlan: LogicalPlan, | ||
| caseSensitive: Boolean = true)(transform: LogicalPlan => LogicalPlan): Unit = { | ||
| withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) { | ||
| val analyzer = getAnalyzer | ||
| val actualPlan = analyzer.executeAndCheck(inputPlan, new QueryPlanningTracker) | ||
| comparePlans(transform(actualPlan), expectedPlan) | ||
| } | ||
| } | ||
|
|
||
| protected def checkAnalysisWithoutViewWrapper( | ||
| inputPlan: LogicalPlan, | ||
| expectedPlan: LogicalPlan, | ||
| caseSensitive: Boolean = true): Unit = { | ||
| checkAnalysisWithTransform(inputPlan, expectedPlan, caseSensitive) { plan => | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. shall we inline |
||
| plan transformUp { | ||
| case v: View if v.isDataFrameTempView => v.child | ||
| } | ||
| } | ||
| } | ||
|
|
||
| protected override def comparePlans( | ||
| plan1: LogicalPlan, | ||
| plan2: LogicalPlan, | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.