huaxingao commented on code in PR #13979:
URL: https://github.com/apache/iceberg/pull/13979#discussion_r4170241355
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java:
##########
@@ -179,17 +181,48 @@ protected TableIdentifier buildIdentifier(Identifier
identifier) {
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
- return load(ident, null /* no time travel */);
+ return loadTable(ident, LoadContext.empty());
+ }
+
+ @Override
+ public Table loadTable(Identifier ident, LoadContext context) throws
NoSuchTableException {
+ return load(ident, null /* no time travel */, context);
}
@Override
public Table loadTable(Identifier ident, String version) throws
NoSuchTableException {
- return load(ident, TimeTravel.version(version));
+ return loadTable(ident, version, LoadContext.empty());
+ }
+
+ @Override
+ public Table loadTable(Identifier ident, String version, LoadContext context)
+ throws NoSuchTableException {
+ return load(ident, TimeTravel.version(version), context);
}
@Override
public Table loadTable(Identifier ident, long timestampMicros) throws
NoSuchTableException {
- return load(ident, TimeTravel.timestampMicros(timestampMicros));
+ return loadTable(ident, timestampMicros, LoadContext.empty());
+ }
+
+ @Override
+ public Table loadTable(Identifier ident, long timestampMicros, LoadContext
context)
+ throws NoSuchTableException {
+ return load(ident, TimeTravel.timestampMicros(timestampMicros), context);
+ }
+
+ @Override
+ public Relation loadRelation(Identifier ident) throws NoSuchTableException {
Review Comment:
Delete this method because this is already implemented in
`BaseCatalog.loadRelation`
##########
spark/v4.2/spark/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewUtil.scala:
##########
@@ -73,10 +100,107 @@ object ViewUtil {
None
}
+ /**
+ * Build the referenced-by view chain from fully qualified view identifier
parts.
+ * Entries must belong to the same catalog as the loaded target.
+ */
+ def buildReferencedByChain(
+ viewChain: Seq[Seq[String]],
+ targetCatalogName: String): java.util.List[TableIdentifier] = {
+ viewChain.foreach { parts =>
+ require(
+ parts.size >= 3,
+ s"View chain entry must be fully qualified [catalog, namespace...,
name], got: " +
+ parts.mkString("."))
+ }
+
+ val crossCatalogViews = viewChain.filter(parts =>
!parts.headOption.contains(targetCatalogName))
+ if (crossCatalogViews.nonEmpty) {
+ throw new IllegalStateException(
+ s"Cross-catalog view references are not supported with referenced-by
enabled. " +
+ s"Views from catalogs
[${crossCatalogViews.map(_.head).distinct.mkString(", ")}] " +
+ s"cannot reference entities in catalog [$targetCatalogName]")
+ }
+
+ val viewIdentifiers = viewChain.map { parts =>
+ TableIdentifier.of(Namespace.of(parts.drop(1).init: _*), parts.last)
+ }
+
+ new util.ArrayList[TableIdentifier](viewIdentifiers.asJava)
+ }
+
+ def qualifyParts(
+ parts: Seq[String],
+ catalogAndNamespace: Seq[String],
+ isCatalog: String => Boolean): Seq[String] = {
+ parts match {
+ case Seq(name) => catalogAndNamespace :+ name
+ case _ if !isCatalog(parts.head) => catalogAndNamespace.head +: parts
+ case _ => parts
+ }
+ }
+
+ def buildViewChain(
+ nameParts: Seq[String],
+ viewCatalogAndNamespace: Seq[String],
+ existingChain: Seq[Seq[String]],
+ isCatalog: String => Boolean): Seq[Seq[String]] = {
+ existingChain :+ qualifyParts(nameParts, viewCatalogAndNamespace,
isCatalog)
+ }
+
+ def loadTable(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ context: LoadContext,
+ timeTravelVersion: Option[String] = None,
+ timeTravelTimestamp: Option[Expression] = None): Table = {
+ catalog match {
+ case supportsLoadContext: SparkSupportsLoadContext =>
+ loadTableWithTimeTravel(
+ supportsLoadContext,
+ ident,
+ context,
+ timeTravelVersion,
+ timeTravelTimestamp)
+ case c if c.asTableCatalog.isInstanceOf[SparkSupportsLoadContext] =>
+ loadTableWithTimeTravel(
+ c.asTableCatalog.asInstanceOf[SparkSupportsLoadContext],
+ ident,
+ context,
+ timeTravelVersion,
+ timeTravelTimestamp)
+ case _ =>
+ (timeTravelVersion, timeTravelTimestamp) match {
+ case (Some(version), _) =>
+ catalog.asTableCatalog.loadTable(ident, version)
+ case (_, Some(timestamp)) =>
+ catalog.asTableCatalog.loadTable(ident,
timestamp.eval().asInstanceOf[Long])
+ case _ =>
+ catalog.asTableCatalog.loadTable(ident)
+ }
+ }
+ }
+
def isIcebergViewCatalog(catalog: CatalogPlugin): Boolean = {
catalog.isInstanceOf[ViewCatalog] &&
catalog.isInstanceOf[HasIcebergCatalog]
}
+ private def loadTableWithTimeTravel(
Review Comment:
With `spark.sql.iceberg.referenced-by.enabled=true`, a view defined with
`TIMESTAMP AS OF` fails at query time:
- `TIMESTAMP AS OF '2026-10-01 12:00:00'` -> `ClassCastException: UTF8String
cannot be cast to java.lang.Long`
- `TIMESTAMP AS OF 1790000000` -> `ClassCastException: Integer cannot be
cast to java.lang.Long`
Both work with the flag off. `timestamp.eval().asInstanceOf[Long]` skips the
cast and conversion Spark applies (`TimeTravelSpec.create`), so the expression
can be a string or an int. Can we reuse Spark's handling here and add tests for
both forms?
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java:
##########
@@ -217,13 +243,38 @@ public Table loadTable(Identifier ident, String version)
throws NoSuchTableExcep
@Override
public Table loadTable(Identifier ident, long timestamp) throws
NoSuchTableException {
+ return loadTable(ident, timestamp, LoadContext.empty());
+ }
+
+ @Override
+ public Table loadTable(Identifier ident, long timestamp, LoadContext context)
+ throws NoSuchTableException {
try {
+ if (icebergCatalog instanceof SparkSupportsLoadContext) {
+ return ((SparkSupportsLoadContext) icebergCatalog).loadTable(ident,
timestamp, context);
+ }
+
+ logUnsupportedContext("table", ident, context);
return icebergCatalog.loadTable(ident, timestamp);
} catch (NoSuchTableException e) {
return getSessionCatalog().loadTable(ident, timestamp);
}
}
+ @Override
+ public Relation loadRelation(Identifier ident) throws NoSuchTableException {
Review Comment:
Delete this method because this is already implemented in
`BaseCatalog.loadRelation`
##########
spark/v4.2/spark/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewUtil.scala:
##########
@@ -73,10 +100,107 @@ object ViewUtil {
None
}
+ /**
+ * Build the referenced-by view chain from fully qualified view identifier
parts.
+ * Entries must belong to the same catalog as the loaded target.
+ */
+ def buildReferencedByChain(
+ viewChain: Seq[Seq[String]],
+ targetCatalogName: String): java.util.List[TableIdentifier] = {
+ viewChain.foreach { parts =>
+ require(
+ parts.size >= 3,
+ s"View chain entry must be fully qualified [catalog, namespace...,
name], got: " +
+ parts.mkString("."))
+ }
+
+ val crossCatalogViews = viewChain.filter(parts =>
!parts.headOption.contains(targetCatalogName))
+ if (crossCatalogViews.nonEmpty) {
+ throw new IllegalStateException(
+ s"Cross-catalog view references are not supported with referenced-by
enabled. " +
+ s"Views from catalogs
[${crossCatalogViews.map(_.head).distinct.mkString(", ")}] " +
+ s"cannot reference entities in catalog [$targetCatalogName]")
+ }
+
+ val viewIdentifiers = viewChain.map { parts =>
+ TableIdentifier.of(Namespace.of(parts.drop(1).init: _*), parts.last)
+ }
+
+ new util.ArrayList[TableIdentifier](viewIdentifiers.asJava)
+ }
+
+ def qualifyParts(
+ parts: Seq[String],
+ catalogAndNamespace: Seq[String],
+ isCatalog: String => Boolean): Seq[String] = {
+ parts match {
+ case Seq(name) => catalogAndNamespace :+ name
+ case _ if !isCatalog(parts.head) => catalogAndNamespace.head +: parts
+ case _ => parts
+ }
+ }
+
+ def buildViewChain(
+ nameParts: Seq[String],
+ viewCatalogAndNamespace: Seq[String],
+ existingChain: Seq[Seq[String]],
+ isCatalog: String => Boolean): Seq[Seq[String]] = {
+ existingChain :+ qualifyParts(nameParts, viewCatalogAndNamespace,
isCatalog)
+ }
+
+ def loadTable(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ context: LoadContext,
+ timeTravelVersion: Option[String] = None,
+ timeTravelTimestamp: Option[Expression] = None): Table = {
+ catalog match {
+ case supportsLoadContext: SparkSupportsLoadContext =>
+ loadTableWithTimeTravel(
+ supportsLoadContext,
+ ident,
+ context,
+ timeTravelVersion,
+ timeTravelTimestamp)
+ case c if c.asTableCatalog.isInstanceOf[SparkSupportsLoadContext] =>
Review Comment:
nit: this case is effectively the same check as the one above.
`asTableCatalog` just casts the same object to `TableCatalog`, so if the
catalog implements `SparkSupportsLoadContext`, the first case already matched.
Can we drop this case?
##########
spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala:
##########
@@ -18,29 +18,221 @@
*/
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)
Review Comment:
I think the context loaded here gets lost before the query runs. Example:
Alice can't read `sales.orders`, but she can read the view
`sales.order_summary`, which selects from it.
1. `SELECT * FROM sales.order_summary`: this rule expands the view and calls
`loadTable(sales.orders)` with `referencedBy=[sales.order_summary]`. The
catalog allows it, and the resulting table goes into the plan here.
2. Before execution, Spark 4.2's `QueryExecution` calls
`V2TableRefreshUtil.refresh`, which reloads every versioned table with a plain
`catalog.loadTable(ident)` and puts that table into the plan instead.
3. That second load has no `referencedBy`, so the catalog sees Alice asking
for `sales.orders` directly and denies it or returns the wrong credentials. The
scan runs against that second table.
##########
spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala:
##########
@@ -18,29 +18,221 @@
*/
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),
_, _) =>
Review Comment:
With the flag on, this rule checks whether every relation name is a view
before the table is loaded. For a plain table, that means one extra `loadView`
call that fails before Spark loads the table. For a REST catalog, that's an
extra round trip that returns 404 for every table in the query. Could we load
the table first and fall back to the view only when the table isn't found?
--
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]
