huaxingao commented on code in PR #13979:
URL: https://github.com/apache/iceberg/pull/13979#discussion_r4213528356
##########
spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala:
##########
@@ -18,29 +18,263 @@
*/
package org.apache.spark.sql.catalyst.analysis
+import org.apache.iceberg.catalog.LoadContext
+import org.apache.iceberg.spark.SparkSQLProperties
+import org.apache.iceberg.spark.SparkSupportsLoadContext
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.RelationCatalog
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.catalog.TableCatalog
+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 = {
+ val timeTravelRelations = plan.collect { case RelationTimeTravel(u:
UnresolvedRelation, _, _) =>
+ u
+ }
+
+ plan resolveOperators {
+ case u @ UnresolvedRelation(nameParts, _, _)
+ if catalogManager.v1SessionCatalog.isTempView(nameParts) =>
+ u
+
+ case u @ UnresolvedRelation(
+ parts @ CatalogAndIdentifier(catalog: RelationCatalog, ident),
+ options,
+ false)
+ if ViewUtil.isIcebergViewCatalog(catalog) &&
+ !options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
&&
+ !timeTravelRelations.exists(_ eq u) =>
+ try {
+ catalog.loadRelation(ident) match {
+ case view: View if ViewUtil.isIcebergView(view) =>
+ createViewRelation(parts, catalog, ident, view)
+ case table: Table =>
+ SubqueryAlias(
+ ident.toQualifiedNameParts(catalog),
+ DataSourceV2Relation.create(table, Some(catalog), Some(ident),
options))
+ case _ => u
+ }
+ } catch {
+ case _: NoSuchTableException => 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) if timeTravelTimestamp.forall(_.resolved) =>
+ val referencedBy = ViewUtil.buildReferencedByChain(viewChain,
catalog.name())
+ val context = LoadContext.builder().referencedBy(referencedBy).build()
+ val timeTravelSpec =
+ TimeTravelSpec.create(timeTravelTimestamp, timeTravelVersion,
conf.sessionLocalTimeZone)
+ try {
+ val table = ViewUtil.loadTable(catalog, tableIdent, context,
timeTravelSpec)
+ val relationCatalog = catalog match {
+ case contextual: TableCatalog with SparkSupportsLoadContext =>
+ LoadContextTableCatalog(contextual, context)
+ case _ => catalog
+ }
+ DataSourceV2Relation.create(
Review Comment:
Line 85 wraps the relation in `SubqueryAlias`, but this one doesn't. Without
the alias, I think a view like `SELECT t.id FROM db.t` can't resolve `t.id`.
Could we add `SubqueryAlias` here too, and a test with a table-qualified column
in the view body?
##########
spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala:
##########
@@ -18,29 +18,263 @@
*/
package org.apache.spark.sql.catalyst.analysis
+import org.apache.iceberg.catalog.LoadContext
+import org.apache.iceberg.spark.SparkSQLProperties
+import org.apache.iceberg.spark.SparkSupportsLoadContext
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.RelationCatalog
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.catalog.TableCatalog
+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 = {
+ val timeTravelRelations = plan.collect { case RelationTimeTravel(u:
UnresolvedRelation, _, _) =>
+ u
+ }
+
+ plan resolveOperators {
+ case u @ UnresolvedRelation(nameParts, _, _)
+ if catalogManager.v1SessionCatalog.isTempView(nameParts) =>
+ u
+
+ case u @ UnresolvedRelation(
+ parts @ CatalogAndIdentifier(catalog: RelationCatalog, ident),
+ options,
+ false)
+ if ViewUtil.isIcebergViewCatalog(catalog) &&
+ !options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
&&
+ !timeTravelRelations.exists(_ eq u) =>
+ try {
+ catalog.loadRelation(ident) match {
+ case view: View if ViewUtil.isIcebergView(view) =>
+ createViewRelation(parts, catalog, ident, view)
+ case table: Table =>
Review Comment:
With `spark.sql.iceberg.referenced-by.enabled=true` and
`SparkSessionCatalog`, reading a non-Iceberg table fails:
```sql
CREATE TABLE default.t1 (id INT) USING parquet;
SELECT * FROM default.t1;
-- [UNSUPPORTED_FEATURE.TABLE_OPERATION] Table `default`.`t1` does not
support batch scan
```
For a non-Iceberg table, `SparkSessionCatalog` returns a `V1Table`, and
wrapping it in `DataSourceV2Relation` doesn't work. Could we return `u` for a
`V1Table` and let Spark handle it? The same issue happens when an Iceberg view
reads a parquet table (the `ViewUtil.loadTable` call below). Could we also add
a test with `SparkSessionCatalog`?
--
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]
