huaxingao commented on code in PR #13979:
URL: https://github.com/apache/iceberg/pull/13979#discussion_r4058632461
##########
spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala:
##########
@@ -18,29 +18,211 @@
*/
package org.apache.spark.sql.catalyst.analysis
+import org.apache.iceberg.catalog.LoadContext
+import org.apache.iceberg.spark.SparkSQLProperties
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.ViewUtil.IcebergViewHelper
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.expressions.UpCast
+import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias
import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View
+import
org.apache.spark.sql.catalyst.plans.logical.views.UnResolvedRelationFromView
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.CurrentOrigin
+import org.apache.spark.sql.catalyst.trees.Origin
import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.connector.catalog.CatalogPlugin
+import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.LookupCatalog
+import org.apache.spark.sql.connector.catalog.View
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-/**
- * Resolves Iceberg views referenced by commands that use
`UnresolvedTableOrView`.
- *
- * In Spark 4.2, view relations are expanded through
`RelationCatalog.loadRelation`, so Spark owns
- * query parsing, identifier resolution, and schema-mode application. This
rule only converts the
- * remaining command targets to `ResolvedV2View` for Iceberg command planning.
- */
case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with
LookupCatalog {
+
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
protected lazy val catalogManager: CatalogManager =
spark.sessionState.catalogManager
- override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (referencedByEnabled) {
+ resolveWithReferencedBy(plan)
+ } else {
+ resolveViewCommands(plan)
+ }
+ }
+
+ private def resolveWithReferencedBy(plan: LogicalPlan): LogicalPlan = plan
resolveOperators {
+ case u @ UnresolvedRelation(nameParts, _, _)
+ if catalogManager.v1SessionCatalog.isTempView(nameParts) =>
+ u
+
+ case u @ UnresolvedRelation(parts @ CatalogAndIdentifier(catalog, ident),
_, _) =>
+ ViewUtil
+ .loadView(catalog, ident)
+ .map(createViewRelation(parts, catalog, ident, _))
+ .getOrElse(u)
+
+ case u @ UnResolvedRelationFromView(
+ tableParts @ CatalogAndIdentifier(catalog, tableIdent),
+ viewChain,
+ options,
+ isStreaming,
+ timeTravelVersion,
+ timeTravelTimestamp) =>
+ val referencedBy = ViewUtil.buildReferencedByChain(viewChain,
catalog.name())
+ val context = LoadContext.builder().referencedBy(referencedBy).build()
+ try {
+ val table =
+ ViewUtil.loadTable(catalog, tableIdent, context, timeTravelVersion,
timeTravelTimestamp)
+ DataSourceV2Relation.create(table, Some(catalog), Some(tableIdent),
options)
+ } catch {
+ case _: NoSuchTableException =>
+ ViewUtil
+ .loadView(catalog, tableIdent, context)
+ .map(view => createViewRelation(tableParts, catalog, tableIdent,
view, viewChain))
+ .getOrElse(UnresolvedRelation(tableParts, options, isStreaming))
+ }
+
+ case u @ UnresolvedTableOrView(CatalogAndIdentifier(catalog, ident), _, _,
_) =>
+ ViewUtil
+ .loadView(catalog, ident)
+ .map(view => ResolvedV2View(catalog.asViewCatalog, ident, view))
+ .getOrElse(u)
+ }
+
+ private def resolveViewCommands(plan: LogicalPlan): LogicalPlan = plan
resolveOperators {
case u @ UnresolvedTableOrView(CatalogAndIdentifier(catalog, ident), _, _,
_) =>
ViewUtil
.loadView(catalog, ident)
.map(view => ResolvedV2View(catalog.asViewCatalog, ident, view))
.getOrElse(u)
}
+
+ private def createViewRelation(
+ nameParts: Seq[String],
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ view: View,
+ existingChain: Seq[Seq[String]] = Seq.empty): LogicalPlan = {
+ val parsed = parseViewText(nameParts.quoted, view.queryText())
+ val viewCatalogAndNamespace = catalogAndNamespace(catalog, ident, view)
+ val viewChain =
+ ViewUtil.buildViewChain(nameParts, viewCatalogAndNamespace,
existingChain, isCatalog)
Review Comment:
`viewCatalogAndNamespace` is the creation-time catalog and namespace, but
here we need the view's actual catalog and namespace.
@bryanck and @RjLi13 raised this earlier and you replied it was fixed using
the session-resolved identity from `CatalogAndIdentifier`, but it looks like
the 4.2 rework lost it.
--
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]
