This is an automated email from the ASF dual-hosted git repository.
singhpk234 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new b71a64ddc2 Spark: [3/N] Support sending referenced-by to all endpoints
(#13979)
b71a64ddc2 is described below
commit b71a64ddc28e4b39e07cec71cd12bc5ed293f8bb
Author: Prashant Singh <[email protected]>
AuthorDate: Tue Nov 3 14:47:21 2026 -0800
Spark: [3/N] Support sending referenced-by to all endpoints (#13979)
* Spark: Send the referenced-by view chain through SparkCatalog
Resolve the chain of views a relation was reached through and pass it to the
catalog as a LoadContext, so a REST catalog can forward it to the server.
SparkSupportsLoadContext extends the loading contract with context-carrying
overloads of loadTable and loadView, including the time-travel forms, and
SparkCatalog and SparkSessionCatalog implement it. ResolveViews records the
chain while resolving nested views, with UnResolvedRelationFromView
carrying it
through analysis. Cross-catalog view references fail rather than silently
dropping the chain.
Gated behind spark.sql.iceberg.referenced-by-enabled, off by default.
* Spark: Move referenced-by context to Spark 4.2
Generated-by: Codex
* Spark: Fix ViewUtil Scala import grouping
Generated-by: Codex
* Spark 4.2: Implement contextual loadTable in TestSparkCatalog
TestSparkCatalog only overrode loadTable(Identifier), so once the read
path started resolving through loadRelation -> loadTable(Identifier,
LoadContext), the double was no longer consulted and lookups fell
through to the real catalog.
Override the contextual variant to delegate to the existing one. The
double ignores the context; TestReferencedByViewChain covers the
referenced-by behaviour against real catalogs.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
* Spark 4.2: Restore explicit referenced-by view resolution
Restore the Spark 4.1-style analyzer path for referenced-by propagation in
Spark 4.2. ResolveViews now expands Iceberg views, carries the view chain
through UnResolvedRelationFromView, and loads nested relations with an explicit
LoadContext instead of using SQLConf-backed view metadata.
Generated-by: Codex
* Core, Spark: Address referenced-by review feedback
Preserve existing cross-catalog view behavior by dropping catalog-relative
referenced-by entries that cannot be represented for the target catalog. Bypass
CachingCatalog for non-empty LoadContext loads so referenced-by reaches
catalogs such as REST, and add regression coverage for cache-enabled Spark
reads, time-travel views, and cross-catalog view references.
Generated-by: Codex
* Spark 4.2: Gate referenced-by view resolution
Add a default-off Spark SQL flag for referenced-by view-chain propagation.
When disabled, Spark keeps its native view resolution behavior, including
cross-catalog views. When enabled, Iceberg uses the explicit analyzer path and
rejects cross-catalog chains because LoadContext carries catalog-relative
identifiers.
Generated-by: Codex
* Spark 4.2: Use resolved view identity for referenced-by
Build the referenced-by view chain from the catalog and identifier Spark
resolved when loading the view. Keep view metadata current catalog and
namespace only for resolving unqualified references inside the view SQL.
Generated-by: Codex
* Spark 4.2: Address referenced-by review feedback
Keep the view chain on Spark's pre-execution table refresh by binding
it to the catalog of relations resolved through a view, so a table read
through a view is never reloaded as direct access. Convert view time
travel with Spark's TimeTravelSpec so string and integer timestamps work
and pinned snapshots are not refreshed. Leave V1 and delegating tables
read through a view to Spark, and alias view-resolved tables so
table-qualified columns resolve. Remove loadRelation overrides that
duplicate BaseCatalog.
* Spark 4.2: Resolve referenced-by views in a dedicated rule
Move referenced-by view expansion out of ResolveViews into
ResolveReferencedByViews, which does nothing when referenced-by is
disabled. ResolveViews is back to its main behavior.
With referenced-by enabled, BaseCatalog.loadRelation returns tables
only, so Spark resolves tables itself and leaves Iceberg views
unresolved for the new rule. This removes the hint-batch registration,
which Spark's Hive session state ignores, and keeps a table read to a
single load. Expanded views are wrapped in Spark's View node so cycle
detection and cache invalidation keep working, and parse, streaming and
not-found errors match Spark's.
LoadContext gains equals and hashCode so cached plans over a view are
reused by later queries.
SparkSessionCatalog fails a load with a referenced-by chain when its
Iceberg catalog cannot pass the context, instead of dropping it.
Generated-by: Claude Code
* Core, Spark 4.2: Bound referenced-by view depth and canonicalize
contextual loads
Fail referenced-by view expansion past spark.sql.view.maxNestedViewDepth
with the same error Spark raises when referenced-by is disabled, so a
view cycle no longer overflows the stack during analysis.
CachingCatalog now sends contextual loads the same canonical identifier
as cached loads, so a table read through a view and read directly reach
the delegate under one name when the cache is case-insensitive.
Drop two Spark tests covered by more specific ones.
Generated-by: Claude Code
* Spark 4.2: Expand referenced-by views through SessionCatalog.getRelation
Referenced-by view expansion now uses the same Spark path as when the
flag is off, so view columns bind by name, every view schema mode is
honored, and the view text is parsed with the SQL configs stored at
CREATE VIEW. Previously columns were bound by position, which returned
wrong data after the underlying columns were reordered.
Tests that should behave the same with the flag on or off now run with
both settings and assert the expected outcome directly.
Generated-by: Claude Code
---------
Co-authored-by: Prashant Singh <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../org/apache/iceberg/catalog/LoadContext.java | 19 +
.../java/org/apache/iceberg/CachingCatalog.java | 10 +
.../apache/iceberg/hadoop/TestCachingCatalog.java | 66 ++
.../extensions/IcebergSparkSessionExtensions.scala | 2 +
.../analysis/LoadContextTableCatalog.scala | 59 ++
.../analysis/ResolveReferencedByViews.scala | 256 +++++++
.../spark/sql/catalyst/analysis/ResolveViews.scala | 3 +-
.../logical/views/UnResolvedRelationFromView.scala | 50 ++
.../extensions/TestReferencedByViewChain.java | 788 +++++++++++++++++++++
.../java/org/apache/iceberg/spark/BaseCatalog.java | 7 +
.../org/apache/iceberg/spark/SparkCatalog.java | 41 +-
.../apache/iceberg/spark/SparkSQLProperties.java | 5 +
.../apache/iceberg/spark/SparkSessionCatalog.java | 68 +-
.../iceberg/spark/SparkSupportsLoadContext.java | 78 ++
.../spark/sql/catalyst/analysis/ViewUtil.scala | 82 +++
.../iceberg/spark/TestSparkSessionCatalog.java | 45 ++
.../iceberg/spark/source/TestSparkCatalog.java | 6 +
17 files changed, 1573 insertions(+), 12 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/catalog/LoadContext.java
b/api/src/main/java/org/apache/iceberg/catalog/LoadContext.java
index af31a37282..dce80095b8 100644
--- a/api/src/main/java/org/apache/iceberg/catalog/LoadContext.java
+++ b/api/src/main/java/org/apache/iceberg/catalog/LoadContext.java
@@ -60,6 +60,25 @@ public final class LoadContext {
return EMPTY;
}
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+
+ if (other == null || getClass() != other.getClass()) {
+ return false;
+ }
+
+ LoadContext that = (LoadContext) other;
+ return referencedBy.equals(that.referencedBy);
+ }
+
+ @Override
+ public int hashCode() {
+ return referencedBy.hashCode();
+ }
+
public static final class Builder {
private List<TableIdentifier> referencedBy;
diff --git a/core/src/main/java/org/apache/iceberg/CachingCatalog.java
b/core/src/main/java/org/apache/iceberg/CachingCatalog.java
index 4ee6d61c09..cf0c0836d8 100644
--- a/core/src/main/java/org/apache/iceberg/CachingCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/CachingCatalog.java
@@ -29,6 +29,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.LoadContext;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
@@ -168,6 +169,15 @@ public class CachingCatalog implements Catalog {
return table;
}
+ @Override
+ public Table loadTable(TableIdentifier ident, LoadContext context) {
+ if (context == null || context.referencedBy().isEmpty()) {
+ return loadTable(ident);
+ }
+
+ return catalog.loadTable(canonicalizeIdentifier(ident), context);
+ }
+
@Override
public boolean dropTable(TableIdentifier ident, boolean purge) {
boolean dropped = catalog.dropTable(ident, purge);
diff --git
a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
index 1c36c1acea..4a8953b5f9 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
@@ -31,6 +31,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.BaseMetadataTable;
import org.apache.iceberg.CachingCatalog;
import org.apache.iceberg.CatalogProperties;
@@ -40,6 +41,7 @@ import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TestableCachingCatalog;
import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.LoadContext;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -275,6 +277,42 @@ public class TestCachingCatalog extends
HadoopTableTestBase {
assertThat(catalog.remainingAgeFor(tableIdent)).get().isEqualTo(HALF_OF_EXPIRATION);
}
+ @Test
+ void contextualLoadBypassesTableCache() throws IOException {
+ ContextTrackingHadoopCatalog underlyingCatalog =
contextTrackingHadoopCatalog();
+ TestableCachingCatalog catalog =
+ TestableCachingCatalog.wrap(underlyingCatalog, EXPIRATION_TTL, ticker);
+ Namespace namespace = Namespace.of("db", "ns1", "ns2");
+ TableIdentifier tableIdent = TableIdentifier.of(namespace, "tbl");
+ TableIdentifier viewIdent = TableIdentifier.of(namespace, "view");
+ Table cachedTable = catalog.createTable(tableIdent, SCHEMA, SPEC,
ImmutableMap.of());
+
+ LoadContext context =
LoadContext.builder().referencedBy(Lists.newArrayList(viewIdent)).build();
+ Table contextualTable = catalog.loadTable(tableIdent, context);
+
+ assertThat(contextualTable).isNotSameAs(cachedTable);
+ assertThat(catalog.cache().asMap()).containsEntry(tableIdent, cachedTable);
+ assertThat(underlyingCatalog.loadContexts()).hasSize(1);
+
assertThat(underlyingCatalog.loadContexts().get(0).referencedBy()).containsExactly(viewIdent);
+ }
+
+ @Test
+ void contextualLoadUsesCanonicalIdentifierWhenCaseInsensitive() throws
IOException {
+ ContextTrackingHadoopCatalog underlyingCatalog =
contextTrackingHadoopCatalog();
+ Catalog catalog = CachingCatalog.wrap(underlyingCatalog, false,
EXPIRATION_TTL.toMillis());
+ Namespace namespace = Namespace.of("db");
+ TableIdentifier tableIdent = TableIdentifier.of(namespace, "tbl");
+ catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of());
+
+ LoadContext context =
+ LoadContext.builder()
+ .referencedBy(Lists.newArrayList(TableIdentifier.of(namespace,
"view")))
+ .build();
+ catalog.loadTable(TableIdentifier.of(Namespace.of("DB"), "TBL"), context);
+
+
assertThat(underlyingCatalog.loadIdentifiers()).containsExactly(tableIdent);
+ }
+
@Test
public void testCacheExpirationEagerlyRemovesMetadataTables() throws
IOException {
TestableCachingCatalog catalog =
@@ -441,4 +479,32 @@ public class TestCachingCatalog extends
HadoopTableTestBase {
.map(type -> TableIdentifier.parse(tableIdent + "." +
type.name().toLowerCase(Locale.ROOT)))
.toArray(TableIdentifier[]::new);
}
+
+ private ContextTrackingHadoopCatalog contextTrackingHadoopCatalog() {
+ ContextTrackingHadoopCatalog catalog = new ContextTrackingHadoopCatalog();
+ catalog.setConf(new Configuration());
+ catalog.initialize(
+ "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
tempDir.getAbsolutePath()));
+ return catalog;
+ }
+
+ private static class ContextTrackingHadoopCatalog extends HadoopCatalog {
+ private final List<TableIdentifier> loadIdentifiers = Lists.newArrayList();
+ private final List<LoadContext> loadContexts = Lists.newArrayList();
+
+ @Override
+ public Table loadTable(TableIdentifier identifier, LoadContext context) {
+ loadIdentifiers.add(identifier);
+ loadContexts.add(context);
+ return super.loadTable(identifier);
+ }
+
+ private List<TableIdentifier> loadIdentifiers() {
+ return loadIdentifiers;
+ }
+
+ private List<LoadContext> loadContexts() {
+ return loadContexts;
+ }
+ }
}
diff --git
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
index 81824e05e9..f8532b1296 100644
---
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
+++
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
@@ -21,6 +21,7 @@ package org.apache.iceberg.spark.extensions
import org.apache.spark.sql.SparkSessionExtensions
import org.apache.spark.sql.catalyst.analysis.CheckViews
import org.apache.spark.sql.catalyst.analysis.ResolveBranch
+import org.apache.spark.sql.catalyst.analysis.ResolveReferencedByViews
import org.apache.spark.sql.catalyst.analysis.ResolveViews
import org.apache.spark.sql.catalyst.optimizer.ReplaceStaticInvoke
import
org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser
@@ -33,6 +34,7 @@ class IcebergSparkSessionExtensions extends
(SparkSessionExtensions => Unit) {
extensions.injectParser { case (_, parser) => new
IcebergSparkSqlExtensionsParser(parser) }
// analyzer extensions
+ extensions.injectResolutionRule { spark => ResolveReferencedByViews(spark)
}
extensions.injectResolutionRule { spark => ResolveViews(spark) }
extensions.injectPostHocResolutionRule { spark => ResolveBranch(spark) }
extensions.injectCheckRule(_ => CheckViews)
diff --git
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/LoadContextTableCatalog.scala
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/LoadContextTableCatalog.scala
new file mode 100644
index 0000000000..4c5d4bcaf7
--- /dev/null
+++
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/LoadContextTableCatalog.scala
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.iceberg.catalog.LoadContext
+import org.apache.iceberg.spark.SparkSupportsLoadContext
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.connector.catalog.TableChange
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Catalog for a table read through a view, so that reloads keep the view
chain. Spark's table
+ * version refresh before execution reloads tables by identifier only, using
the relation's
+ * catalog. Only read operations are supported.
+ */
+private[analysis] case class LoadContextTableCatalog(
+ delegate: TableCatalog with SparkSupportsLoadContext,
+ context: LoadContext)
+ extends TableCatalog {
+
+ override def name(): String = delegate.name()
+
+ override def initialize(name: String, options: CaseInsensitiveStringMap):
Unit = {}
+
+ override def loadTable(ident: Identifier): Table = delegate.loadTable(ident,
context)
+
+ override def invalidateTable(ident: Identifier): Unit =
delegate.invalidateTable(ident)
+
+ override def listTables(namespace: Array[String]): Array[Identifier] =
+ delegate.listTables(namespace)
+
+ override def alterTable(ident: Identifier, changes: TableChange*): Table =
+ throw new UnsupportedOperationException(s"Cannot alter table $ident
through a view relation")
+
+ override def dropTable(ident: Identifier): Boolean =
+ throw new UnsupportedOperationException(s"Cannot drop table $ident through
a view relation")
+
+ override def renameTable(oldIdent: Identifier, newIdent: Identifier): Unit =
+ throw new UnsupportedOperationException(
+ s"Cannot rename table $oldIdent through a view relation")
+}
diff --git
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencedByViews.scala
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencedByViews.scala
new file mode 100644
index 0000000000..d096437564
--- /dev/null
+++
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencedByViews.scala
@@ -0,0 +1,256 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.iceberg.catalog.LoadContext
+import org.apache.iceberg.spark.SparkSupportsLoadContext
+import org.apache.spark.sql.SparkSession
+import
org.apache.spark.sql.catalyst.analysis.ResolveReferencedByViews.CheckedForView
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.{View => ViewNode}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias
+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.TreeNodeTag
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.connector.catalog.CatalogPlugin
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.connector.catalog.DelegatingTable
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.catalog.LookupCatalog
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.connector.catalog.V1Table
+import org.apache.spark.sql.connector.catalog.View
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Expands Iceberg views when referenced-by is enabled, so that the tables and
views they read are
+ * loaded with the referencing view chain.
+ *
+ * Spark 4.2 expands views itself without passing the view identity to the
catalog, so with
+ * referenced-by enabled Iceberg catalogs leave views unresolved for this rule.
+ */
+case class ResolveReferencedByViews(spark: SparkSession)
+ extends Rule[LogicalPlan]
+ with LookupCatalog {
+
+ protected lazy val catalogManager: CatalogManager =
spark.sessionState.catalogManager
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (ViewUtil.referencedByEnabled) {
+ resolveWithReferencedBy(plan)
+ } else {
+ plan
+ }
+ }
+
+ private def resolveWithReferencedBy(plan: LogicalPlan): LogicalPlan = plan
resolveOperators {
+ case u @ UnresolvedRelation(nameParts, _, _) if isTempView(nameParts) =>
+ u
+
+ case u @ UnresolvedRelation(CatalogAndIdentifier(catalog, ident), _, true)
+ if u.getTagValue(CheckedForView).isEmpty =>
+ ViewUtil.loadView(catalog, ident).foreach { _ =>
+ throw
QueryCompilationErrors.permanentViewNotSupportedByStreamingReadingAPIError(
+ ident.quoted)
+ }
+ markCheckedForView(u)
+
+ case u @ UnresolvedRelation(CatalogAndIdentifier(catalog, ident), options,
_)
+ if !options.containsKey(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
&&
+ u.getTagValue(CheckedForView).isEmpty =>
+ ViewUtil
+ .loadView(catalog, ident)
+ .map(createViewRelation(catalog, ident, _))
+ .getOrElse(markCheckedForView(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)
+ table match {
+ case _: V1Table | _: DelegatingTable =>
+ val relation = markCheckedForView(UnresolvedRelation(tableParts,
options, isStreaming))
+ if (timeTravelSpec.isEmpty) {
+ relation
+ } else {
+ RelationTimeTravel(relation, timeTravelTimestamp,
timeTravelVersion)
+ }
+
+ case _ =>
+ val relationCatalog = catalog match {
+ case contextual: TableCatalog with SparkSupportsLoadContext
+ if timeTravelSpec.isEmpty =>
+ LoadContextTableCatalog(contextual, context)
+ case _ => catalog
+ }
+ SubqueryAlias(
+ tableIdent.toQualifiedNameParts(catalog),
+ DataSourceV2Relation.create(
+ table,
+ Some(relationCatalog),
+ Some(tableIdent),
+ options,
+ timeTravelSpec))
+ }
+ } catch {
+ case _: NoSuchTableException =>
+ ViewUtil
+ .loadView(catalog, tableIdent, context)
+ .map { view =>
+ if (viewChain.size >= conf.maxNestedViewDepth) {
+ throw
QueryCompilationErrors.viewDepthExceedsMaxResolutionDepthError(
+ tableIdent.toQualifiedNameParts(catalog),
+ conf.maxNestedViewDepth,
+ u)
+ }
+ createViewRelation(catalog, tableIdent, view, viewChain)
+ }
+ .getOrElse(throw tableOrViewNotFound(u))
+ }
+ }
+
+ private def createViewRelation(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ view: View,
+ existingChain: Seq[Seq[String]] = Seq.empty): LogicalPlan = {
+ val desc = V1Table.toCatalogTable(catalog, ident, view)
+ val viewQueryCatalogAndNamespace = queryCatalogAndNamespace(catalog,
ident, view)
+ val viewChain = existingChain :+ ident.toQualifiedNameParts(catalog)
+ catalogManager.v1SessionCatalog.getRelation(desc) transformDown { case v:
ViewNode =>
+ v.copy(child = rewriteIdentifiers(v.child, viewQueryCatalogAndNamespace,
viewChain))
+ }
+ }
+
+ private def queryCatalogAndNamespace(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ view: View): Seq[String] = {
+ val currentCatalog =
Option(view.currentCatalog()).filter(_.nonEmpty).getOrElse(catalog.name())
+ val currentNamespace =
Option(view.currentNamespace()).map(_.toIndexedSeq).getOrElse {
+ ident.namespace().toIndexedSeq
+ }
+ currentCatalog +: currentNamespace
+ }
+
+ private def rewriteIdentifiers(
+ plan: LogicalPlan,
+ catalogAndNamespace: Seq[String],
+ viewChain: Seq[Seq[String]]): LogicalPlan = {
+ qualifyTableIdentifiers(
+ qualifyFunctionIdentifiers(CTESubstitution.apply(plan),
catalogAndNamespace),
+ catalogAndNamespace,
+ viewChain)
+ }
+
+ private def qualifyFunctionIdentifiers(
+ plan: LogicalPlan,
+ catalogAndNamespace: Seq[String]): LogicalPlan = plan
transformExpressions {
+ case u @ UnresolvedFunction(Seq(name), _, _, _, _, _, _) =>
+ if (!isBuiltinFunction(name)) {
+ u.copy(nameParts = catalogAndNamespace :+ name)
+ } else {
+ u
+ }
+ case u @ UnresolvedFunction(parts, _, _, _, _, _, _) if
!isCatalog(parts.head) =>
+ u.copy(nameParts = catalogAndNamespace.head +: parts)
+ }
+
+ private def qualifyTableIdentifiers(
+ child: LogicalPlan,
+ catalogAndNamespace: Seq[String],
+ viewChain: Seq[Seq[String]]): LogicalPlan = {
+ child transform {
+ case UnresolvedRelation(parts, options, isStreaming) =>
+ val qualifiedTableId = ViewUtil.qualifyParts(parts,
catalogAndNamespace, isCatalog)
+ UnResolvedRelationFromView(
+ qualifiedTableId,
+ parts,
+ catalogAndNamespace,
+ viewChain,
+ options,
+ isStreaming)
+ case RelationTimeTravel(
+ UnresolvedRelation(parts, options, isStreaming),
+ timestampOpt,
+ versionOpt) =>
+ val qualifiedTableId = ViewUtil.qualifyParts(parts,
catalogAndNamespace, isCatalog)
+ UnResolvedRelationFromView(
+ qualifiedTableId,
+ parts,
+ catalogAndNamespace,
+ viewChain,
+ options,
+ isStreaming,
+ timeTravelVersion = versionOpt,
+ timeTravelTimestamp = timestampOpt)
+ case other =>
+ other.transformExpressions { case subquery: SubqueryExpression =>
+ subquery.withNewPlan(
+ qualifyTableIdentifiers(subquery.plan, catalogAndNamespace,
viewChain))
+ }
+ }
+ }
+
+ private def tableOrViewNotFound(relation: UnResolvedRelationFromView):
Throwable = {
+ val catalogAndNamespace = relation.viewCatalogAndNamespace
+ val searchPath = catalogManager
+ .sqlResolutionPathEntries(catalogAndNamespace.head,
catalogAndNamespace.tail)
+ .map(QueryCompilationErrors.toSQLId)
+ QueryCompilationErrors.tableOrViewNotFoundWithSearchPath(
+ relation.nameParts,
+ searchPath,
+ relation.origin)
+ }
+
+ private def markCheckedForView(relation: UnresolvedRelation):
UnresolvedRelation = {
+ relation.setTagValue(CheckedForView, ())
+ relation
+ }
+
+ private def isTempView(nameParts: Seq[String]): Boolean = {
+ catalogManager.v1SessionCatalog.isTempView(nameParts)
+ }
+
+ private def isCatalog(name: String): Boolean = {
+ catalogManager.isCatalogRegistered(name)
+ }
+
+ private def isBuiltinFunction(name: String): Boolean = {
+ catalogManager.v1SessionCatalog.isBuiltinFunction(name)
+ }
+}
+
+object ResolveReferencedByViews {
+ private val CheckedForView =
TreeNodeTag[Unit]("iceberg.referenced-by.checked-for-view")
+}
diff --git
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala
index 4945c7ea56..fac9cb6ec6 100644
---
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala
+++
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala
@@ -31,7 +31,8 @@ import org.apache.spark.sql.connector.catalog.LookupCatalog
*
* 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.
+ * remaining command targets to `ResolvedV2View` for Iceberg command planning.
With referenced-by
+ * enabled, Iceberg views in queries are expanded by
`ResolveReferencedByViews` instead.
*/
case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with
LookupCatalog {
protected lazy val catalogManager: CatalogManager =
spark.sessionState.catalogManager
diff --git
a/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/views/UnResolvedRelationFromView.scala
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/views/UnResolvedRelationFromView.scala
new file mode 100644
index 0000000000..28820a8168
--- /dev/null
+++
b/spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/views/UnResolvedRelationFromView.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.spark.sql.catalyst.plans.logical.views
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.plans.logical.LeafNode
+import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+case class UnResolvedRelationFromView(
+ tableMultipartIdentifier: Seq[String],
+ nameParts: Seq[String],
+ viewCatalogAndNamespace: Seq[String],
+ viewChain: Seq[Seq[String]],
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty(),
+ override val isStreaming: Boolean = false,
+ timeTravelVersion: Option[String] = None,
+ timeTravelTimestamp: Option[Expression] = None)
+ extends LeafNode {
+
+ override def output: Seq[Attribute] = Nil
+
+ override lazy val resolved: Boolean = false
+
+ def tableName: String = tableMultipartIdentifier.quoted
+
+ def viewName: String = viewChain.map(_.quoted).mkString(" -> ")
+
+ override def simpleString(maxFields: Int): String = {
+ s"'UnResolvedRelationFromView [table=$tableName, views=$viewName, " +
+ s"${if (isStreaming) "streaming=true, " else ""}options=$options]"
+ }
+}
diff --git
a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestReferencedByViewChain.java
b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestReferencedByViewChain.java
new file mode 100644
index 0000000000..cc0d7b8189
--- /dev/null
+++
b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestReferencedByViewChain.java
@@ -0,0 +1,788 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import static org.apache.iceberg.TestHelpers.waitUntilAfter;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.net.InetAddress;
+import java.nio.file.Path;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.LoadContext;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.SparkSQLProperties;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.spark.SparkSessionCatalog;
+import org.apache.iceberg.spark.SparkTestHelperBase;
+import org.apache.iceberg.spark.TestBase;
+import org.apache.iceberg.spark.source.HasIcebergCatalog;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.view.View;
+import org.apache.spark.SparkThrowable;
+import org.apache.spark.sql.AnalysisException;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+public class TestReferencedByViewChain extends SparkTestHelperBase {
+
+ private static final String CATALOG_NAME = "ref_test_catalog";
+ private static final String OTHER_CATALOG_NAME = "other_ref_test_catalog";
+ private static final Namespace NAMESPACE = Namespace.of("default");
+ private static final Namespace VIEW_DEFAULT_NAMESPACE =
Namespace.of("view_defaults");
+ private static final String TABLE_NAME = "test_table";
+ private static final String OTHER_TABLE_NAME = "other_table";
+ private static final String VIEW_DEFAULT_TABLE_NAME = "view_default_table";
+ private static final String VIEW_DEFAULT_VIEW_NAME = "view_default_view";
+ private static final String SESSION_CATALOG_NAME = "spark_catalog";
+ private static final String PARQUET_TABLE_NAME = "parquet_table";
+ private static final String PARQUET_VIEW_NAME = "parquet_view";
+ private static final String SCHEMA_TABLE_NAME = "schema_table";
+
+ @TempDir private static Path warehouse;
+
+ private static SparkSession spark;
+
+ public static class ContextTrackingCatalog extends InMemoryCatalog {
+ public static final List<CapturedContext> CAPTURED = new
CopyOnWriteArrayList<>();
+ public static final List<CapturedContext> CAPTURED_VIEWS = new
CopyOnWriteArrayList<>();
+
+ public static class CapturedContext {
+ public final TableIdentifier tableIdentifier;
+ public final List<TableIdentifier> referencedBy;
+
+ CapturedContext(TableIdentifier tableIdentifier, List<TableIdentifier>
referencedBy) {
+ this.tableIdentifier = tableIdentifier;
+ this.referencedBy = referencedBy;
+ }
+ }
+
+ public static void clearCaptured() {
+ CAPTURED.clear();
+ CAPTURED_VIEWS.clear();
+ }
+
+ @Override
+ public Table loadTable(TableIdentifier identifier) {
+ CAPTURED.add(new CapturedContext(identifier, Collections.emptyList()));
+ return super.loadTable(identifier);
+ }
+
+ @Override
+ public Table loadTable(TableIdentifier identifier, LoadContext context) {
+ CAPTURED.add(new CapturedContext(identifier, referencedBy(context)));
+ return super.loadTable(identifier);
+ }
+
+ @Override
+ public View loadView(TableIdentifier identifier) {
+ CAPTURED_VIEWS.add(new CapturedContext(identifier,
Collections.emptyList()));
+ return super.loadView(identifier);
+ }
+
+ @Override
+ public View loadView(TableIdentifier identifier, LoadContext context) {
+ CAPTURED_VIEWS.add(new CapturedContext(identifier,
referencedBy(context)));
+ return super.loadView(identifier);
+ }
+
+ private static List<TableIdentifier> referencedBy(LoadContext context) {
+ return context != null ? context.referencedBy() :
Collections.emptyList();
+ }
+ }
+
+ @BeforeAll
+ static void startSpark() {
+ spark =
+ SparkSession.builder()
+ .master("local[2]")
+ .config("spark.driver.host",
InetAddress.getLoopbackAddress().getHostAddress())
+ .config(TestBase.DISABLE_UI)
+ .config("spark.sql.extensions",
IcebergSparkSessionExtensions.class.getName())
+ .config("spark.sql.catalog." + CATALOG_NAME,
SparkCatalog.class.getName())
+ .config(
+ "spark.sql.catalog." + CATALOG_NAME + "." +
CatalogProperties.CATALOG_IMPL,
+ ContextTrackingCatalog.class.getName())
+ .config("spark.sql.catalog." + CATALOG_NAME +
".default-namespace", "default")
+ .config("spark.sql.catalog." + CATALOG_NAME + ".cache-enabled",
"true")
+ .config("spark.sql.catalog." + OTHER_CATALOG_NAME,
SparkCatalog.class.getName())
+ .config(
+ "spark.sql.catalog." + OTHER_CATALOG_NAME + "." +
CatalogProperties.CATALOG_IMPL,
+ ContextTrackingCatalog.class.getName())
+ .config("spark.sql.catalog." + OTHER_CATALOG_NAME +
".default-namespace", "default")
+ .config("spark.sql.catalog." + OTHER_CATALOG_NAME +
".cache-enabled", "false")
+ .config(
+ "spark.sql.catalog." + SESSION_CATALOG_NAME,
SparkSessionCatalog.class.getName())
+ .config(
+ "spark.sql.catalog." + SESSION_CATALOG_NAME + "." +
CatalogProperties.CATALOG_IMPL,
+ InMemoryCatalog.class.getName())
+ .config("spark.sql.catalog." + SESSION_CATALOG_NAME +
".cache-enabled", "false")
+ .config("spark.sql.warehouse.dir", warehouse.toString())
+ .config("spark.sql.defaultCatalog", CATALOG_NAME)
+ // Cover Hive-enabled sessions, whose analyzer is built separately
from the default one
+ .config("spark.hadoop.hive.metastore.uris", "")
+ .config(
+ "spark.hadoop.javax.jdo.option.ConnectionURL",
+ "jdbc:derby:memory:" + UUID.randomUUID() + ";create=true")
+ .enableHiveSupport()
+ .getOrCreate();
+
+ spark.sql(String.format("CREATE NAMESPACE IF NOT EXISTS %s.%s",
CATALOG_NAME, NAMESPACE));
+ spark.sql(
+ String.format(
+ "CREATE NAMESPACE IF NOT EXISTS %s.%s", CATALOG_NAME,
VIEW_DEFAULT_NAMESPACE));
+ spark.sql(String.format("CREATE NAMESPACE IF NOT EXISTS %s.%s",
OTHER_CATALOG_NAME, NAMESPACE));
+ ((SupportsNamespaces) Spark3Util.loadIcebergCatalog(spark,
SESSION_CATALOG_NAME))
+ .createNamespace(NAMESPACE);
+ }
+
+ @AfterAll
+ static void stopSpark() {
+ if (spark != null) {
+ spark.stop();
+ spark = null;
+ }
+ }
+
+ @BeforeEach
+ void before() {
+ spark.sql(String.format("USE %s.%s", CATALOG_NAME, NAMESPACE));
+ spark.sql(String.format("CREATE TABLE IF NOT EXISTS %s (id INT, data
STRING)", TABLE_NAME));
+ appendRecords(TABLE_NAME);
+ spark.sql(String.format("REFRESH TABLE %s", TABLE_NAME));
+
+ ContextTrackingCatalog.clearCaptured();
+ }
+
+ @AfterEach
+ void after() {
+ spark.conf().unset(SparkSQLProperties.REFERENCED_BY_ENABLED);
+ spark.sql(String.format("USE %s.%s", CATALOG_NAME, NAMESPACE));
+ spark.sql("DROP VIEW IF EXISTS simple_view");
+ spark.sql("DROP VIEW IF EXISTS view_a");
+ spark.sql("DROP VIEW IF EXISTS view_b");
+ spark.sql("DROP VIEW IF EXISTS view_c");
+ spark.sql("DROP VIEW IF EXISTS time_travel_view");
+ spark.sql("DROP VIEW IF EXISTS cross_view");
+ spark.sql("DROP VIEW IF EXISTS shadow_v");
+ spark.sql(String.format("DROP VIEW IF EXISTS %s", VIEW_DEFAULT_VIEW_NAME));
+ spark.sql(String.format("DROP TABLE IF EXISTS %s", TABLE_NAME));
+ spark.sql(String.format("DROP TABLE IF EXISTS %s", SCHEMA_TABLE_NAME));
+ spark.sql(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s", VIEW_DEFAULT_NAMESPACE,
VIEW_DEFAULT_TABLE_NAME));
+ spark.sql(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s.%s", OTHER_CATALOG_NAME, NAMESPACE,
OTHER_TABLE_NAME));
+ viewCatalog(SESSION_CATALOG_NAME).dropView(TableIdentifier.of(NAMESPACE,
PARQUET_VIEW_NAME));
+ spark.sql(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s.%s", SESSION_CATALOG_NAME, NAMESPACE,
PARQUET_TABLE_NAME));
+ ContextTrackingCatalog.clearCaptured();
+ }
+
+ @Test
+ void referencedByIsDisabledByDefault() {
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM simple_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertThat(ContextTrackingCatalog.CAPTURED)
+ .allSatisfy(captured -> assertThat(captured.referencedBy).isEmpty());
+ assertThat(ContextTrackingCatalog.CAPTURED_VIEWS)
+ .allSatisfy(captured -> assertThat(captured.referencedBy).isEmpty());
+ }
+
+ @Test
+ void singleViewPassesViewIdentifierInContext() {
+ setReferencedByEnabled(true);
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+ spark.sql(String.format("REFRESH TABLE %s", TABLE_NAME));
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM simple_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertThat(capturedLoadsOf(TableIdentifier.of(NAMESPACE, TABLE_NAME)))
+ .isNotEmpty()
+ .allSatisfy(
+ captured ->
+ assertThat(captured.referencedBy)
+ .containsExactly(TableIdentifier.of(NAMESPACE,
"simple_view")));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void viewBodyResolvesTableQualifiedColumns(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("view_a", String.format("SELECT %s.id FROM %s", TABLE_NAME,
TABLE_NAME));
+ createView(
+ "view_b",
+ String.format(
+ "SELECT %s.%s.%s.id FROM %s.%s.%s",
+ CATALOG_NAME, NAMESPACE, TABLE_NAME, CATALOG_NAME, NAMESPACE,
TABLE_NAME));
+
+ assertThat(spark.sql("SELECT * FROM view_a").collectAsList()).hasSize(5);
+ assertThat(spark.sql("SELECT * FROM view_b").collectAsList()).hasSize(5);
+ }
+
+ @Test
+ void referencedByUsesResolvedViewIdentifier() {
+ setReferencedByEnabled(true);
+ spark.sql(
+ String.format(
+ "CREATE TABLE %s.%s (id INT, data STRING)",
+ VIEW_DEFAULT_NAMESPACE, VIEW_DEFAULT_TABLE_NAME));
+ appendRecords(String.format("%s.%s", VIEW_DEFAULT_NAMESPACE,
VIEW_DEFAULT_TABLE_NAME));
+ spark.sql(
+ String.format("REFRESH TABLE %s.%s", VIEW_DEFAULT_NAMESPACE,
VIEW_DEFAULT_TABLE_NAME));
+
+ String viewSql = String.format("SELECT id FROM %s",
VIEW_DEFAULT_TABLE_NAME);
+ String schemaSql =
+ String.format("SELECT id FROM %s.%s", VIEW_DEFAULT_NAMESPACE,
VIEW_DEFAULT_TABLE_NAME);
+ viewCatalog(CATALOG_NAME)
+ .buildView(TableIdentifier.of(NAMESPACE, VIEW_DEFAULT_VIEW_NAME))
+ .withQuery("spark", viewSql)
+ .withDefaultNamespace(VIEW_DEFAULT_NAMESPACE)
+ .withDefaultCatalog(CATALOG_NAME)
+ .withSchema(SparkSchemaUtil.convert(spark.sql(schemaSql).schema()))
+ .create();
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result =
+ spark.sql(String.format("SELECT * FROM %s",
VIEW_DEFAULT_VIEW_NAME)).collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertCapturedTableChain(
+ ContextTrackingCatalog.CAPTURED,
+ VIEW_DEFAULT_NAMESPACE,
+ VIEW_DEFAULT_TABLE_NAME,
+ VIEW_DEFAULT_VIEW_NAME);
+ }
+
+ @Test
+ void nestedViewChainAccumulatesContext() {
+ setReferencedByEnabled(true);
+ createView("view_a", String.format("SELECT id, data FROM %s", TABLE_NAME));
+ createView("view_b", "SELECT id FROM view_a WHERE id <= 3");
+ createView("view_c", "SELECT id FROM view_b WHERE id > 1");
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Object[]> result = rowsToJava(spark.sql("SELECT * FROM
view_c").collectAsList());
+ assertThat(result).hasSize(2).containsExactlyInAnyOrder(new Object[] {2},
new Object[] {3});
+
+ assertCapturedTableChain(
+ ContextTrackingCatalog.CAPTURED, TABLE_NAME, "view_c", "view_b",
"view_a");
+ assertCapturedTableChain(ContextTrackingCatalog.CAPTURED_VIEWS, "view_a",
"view_c", "view_b");
+ }
+
+ @Test
+ void timeTravelViewPassesViewIdentifierInContext() throws Exception {
+ setReferencedByEnabled(true);
+ long snapshotId = Spark3Util.loadIcebergTable(spark,
TABLE_NAME).currentSnapshot().snapshotId();
+ createView(
+ "time_travel_view",
+ String.format("SELECT id FROM %s VERSION AS OF %d", TABLE_NAME,
snapshotId));
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM
time_travel_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertCapturedTableChain(ContextTrackingCatalog.CAPTURED, TABLE_NAME,
"time_travel_view");
+ }
+
+ @Test
+ void crossCatalogViewAccessDoesNotFailWhenReferencedByIsDisabled() {
+ createCrossCatalogView();
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM cross_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertThat(capturedLoadsOf(TableIdentifier.of(NAMESPACE,
OTHER_TABLE_NAME)))
+ .isNotEmpty()
+ .allSatisfy(captured -> assertThat(captured.referencedBy).isEmpty());
+ }
+
+ @Test
+ void crossCatalogViewAccessFailsWhenReferencedByIsEnabled() {
+ setReferencedByEnabled(true);
+ createCrossCatalogView();
+
+ assertThatThrownBy(() -> spark.sql("SELECT * FROM
cross_view").collectAsList())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cross-catalog view references are not supported with
referenced-by enabled. "
+ + "Views from catalogs [%s] cannot reference entities in
catalog [%s]",
+ CATALOG_NAME, OTHER_CATALOG_NAME);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void versionTimeTravelViewReadsPinnedSnapshot(boolean referencedByEnabled)
throws Exception {
+ setReferencedByEnabled(referencedByEnabled);
+ long snapshotId = Spark3Util.loadIcebergTable(spark,
TABLE_NAME).currentSnapshot().snapshotId();
+ createView(
+ "time_travel_view",
+ String.format("SELECT id FROM %s VERSION AS OF %d", TABLE_NAME,
snapshotId));
+ appendRecords(TABLE_NAME);
+
+ List<Row> result = spark.sql("SELECT * FROM
time_travel_view").collectAsList();
+ assertThat(result).hasSize(5);
+ }
+
+ @Test
+ void timestampStringTimeTravelView() throws Exception {
+ setReferencedByEnabled(true);
+ long asOfMillis = waitUntilAfter(currentSnapshotMillis());
+ String asOf = new Timestamp(asOfMillis).toString();
+ createView(
+ "time_travel_view",
+ String.format("SELECT id FROM %s TIMESTAMP AS OF '%s'", TABLE_NAME,
asOf));
+ waitUntilAfter(asOfMillis);
+ appendRecords(TABLE_NAME);
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM
time_travel_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertCapturedTableChain(ContextTrackingCatalog.CAPTURED, TABLE_NAME,
"time_travel_view");
+ }
+
+ @Test
+ void timestampSecondsTimeTravelView() throws Exception {
+ setReferencedByEnabled(true);
+ long asOfSeconds =
TimeUnit.MILLISECONDS.toSeconds(currentSnapshotMillis()) + 1;
+ createView(
+ "time_travel_view",
+ String.format("SELECT id FROM %s TIMESTAMP AS OF %d", TABLE_NAME,
asOfSeconds));
+ waitUntilAfter(TimeUnit.SECONDS.toMillis(asOfSeconds));
+ appendRecords(TABLE_NAME);
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM
time_travel_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertCapturedTableChain(ContextTrackingCatalog.CAPTURED, TABLE_NAME,
"time_travel_view");
+ }
+
+ @Test
+ void timestampExpressionTimeTravelView() {
+ setReferencedByEnabled(true);
+ createView(
+ "time_travel_view",
+ String.format("SELECT id FROM %s TIMESTAMP AS OF current_timestamp()",
TABLE_NAME));
+ ContextTrackingCatalog.clearCaptured();
+
+ List<Row> result = spark.sql("SELECT * FROM
time_travel_view").collectAsList();
+ assertThat(result).hasSize(5);
+
+ assertCapturedTableChain(ContextTrackingCatalog.CAPTURED, TABLE_NAME,
"time_travel_view");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void directReadOfNonIcebergTableInSessionCatalog(boolean
referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ String parquetTable = createSessionCatalogParquetTable();
+
+ assertThat(spark.sql(String.format("SELECT * FROM %s",
parquetTable)).count()).isEqualTo(3);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void viewReadsNonIcebergTableInSessionCatalog(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ String parquetTable = createSessionCatalogParquetTable();
+ createView(
+ SESSION_CATALOG_NAME, PARQUET_VIEW_NAME, String.format("SELECT id FROM
%s", parquetTable));
+
+ assertThat(
+ spark
+ .sql(
+ String.format(
+ "SELECT * FROM %s.%s.%s",
+ SESSION_CATALOG_NAME, NAMESPACE, PARQUET_VIEW_NAME))
+ .count())
+ .isEqualTo(3);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void directTableReadDoesNotLoadView(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ spark.sql(String.format("SELECT * FROM %s", TABLE_NAME)).collectAsList();
+
+ assertThat(ContextTrackingCatalog.CAPTURED_VIEWS).isEmpty();
+ }
+
+ @Test
+ void directTableReadLoadsTableAsOftenAsWhenReferencedByIsDisabled() {
+ String otherTable = String.format("%s.%s.%s", OTHER_CATALOG_NAME,
NAMESPACE, OTHER_TABLE_NAME);
+ spark.sql(String.format("CREATE TABLE %s (id INT, data STRING)",
otherTable));
+ appendRecords(otherTable);
+ TableIdentifier ident = TableIdentifier.of(NAMESPACE, OTHER_TABLE_NAME);
+
+ ContextTrackingCatalog.clearCaptured();
+ spark.sql(String.format("SELECT * FROM %s", otherTable)).collectAsList();
+ int expectedLoads = capturedLoadsOf(ident).size();
+
+ setReferencedByEnabled(true);
+ ContextTrackingCatalog.clearCaptured();
+ spark.sql(String.format("SELECT * FROM %s", otherTable)).collectAsList();
+
+ assertThat(expectedLoads).isPositive();
+ assertThat(capturedLoadsOf(ident)).hasSize(expectedLoads);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void tempViewShadowsIcebergView(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("shadow_v", String.format("SELECT id FROM %s", TABLE_NAME));
+ spark.sql("CREATE TEMPORARY VIEW shadow_v AS SELECT 'x' AS temp_col");
+ try {
+ List<String> columns =
+ spark.sql("DESCRIBE TABLE shadow_v").collectAsList().stream()
+ .map(row -> row.getString(0))
+ .collect(Collectors.toList());
+
+ assertThat(columns).contains("temp_col").doesNotContain("id");
+ } finally {
+ spark.catalog().dropTempView("shadow_v");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void cachedViewIsReusedByLaterQuery(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+ Dataset<Row> cached = spark.sql("SELECT * FROM simple_view");
+ cached.cache();
+ try {
+ cached.count();
+ Dataset<Row> query = spark.sql("SELECT * FROM simple_view");
+
+
assertThat(query.queryExecution().withCachedData().toString()).contains("InMemoryRelation");
+ } finally {
+ cached.unpersist();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void streamingReadOfViewFails(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+
+ assertThatThrownBy(() -> spark.readStream().table("simple_view"))
+ .isInstanceOf(AnalysisException.class)
+ .hasMessageContaining(
+ "is a permanent view, which is not supported by streaming reading
API");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void deleteFromViewFails(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+
+ assertFailsWithCondition(
+ () -> spark.sql("DELETE FROM simple_view WHERE id = 1"),
"TABLE_OR_VIEW_NOT_FOUND");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void createOrReplaceViewWithCycleFails(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("view_a", String.format("SELECT id FROM %s", TABLE_NAME));
+ createView("view_b", "SELECT id FROM view_a");
+
+ assertFailsWithCondition(
+ () -> spark.sql("CREATE OR REPLACE VIEW view_a AS SELECT id FROM
view_b"),
+ "RECURSIVE_VIEW");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void viewCycleFails(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createViewWithoutAnalysis("view_a", "SELECT id FROM view_b");
+ createViewWithoutAnalysis("view_b", "SELECT id FROM view_a");
+
+ assertFailsWithCondition(
+ () -> spark.sql("SELECT * FROM view_a").collectAsList(),
"VIEW_EXCEED_MAX_NESTED_DEPTH");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void alterViewPropertiesUncachesView(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+ spark.catalog().cacheTable("simple_view");
+
+ spark.sql("ALTER VIEW simple_view SET TBLPROPERTIES ('key' = 'value')");
+
+ assertThat(spark.catalog().isCached("simple_view")).isFalse();
+ }
+
+ @Test
+ void viewCommandsMatchReferencedByDisabled() {
+ createView("simple_view", String.format("SELECT id FROM %s", TABLE_NAME));
+
+ for (String command : List.of("DESCRIBE TABLE simple_view", "SHOW CREATE
TABLE simple_view")) {
+ setReferencedByEnabled(false);
+ List<Row> expected = spark.sql(command).collectAsList();
+
+ setReferencedByEnabled(true);
+
assertThat(spark.sql(command).collectAsList()).as(command).isEqualTo(expected);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void invalidViewSqlFails(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createViewWithoutAnalysis("simple_view", "invalid SQL");
+
+ assertFailsWithCondition(
+ () -> spark.sql("SELECT * FROM simple_view").collectAsList(),
"PARSE_SYNTAX_ERROR");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void viewReadingMissingTableFails(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createViewWithoutAnalysis("simple_view", "SELECT id FROM non_existing");
+
+ assertFailsWithCondition(
+ () -> spark.sql("SELECT * FROM simple_view").collectAsList(),
"TABLE_OR_VIEW_NOT_FOUND");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void viewColumnsBindByName(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createSchemaTable();
+ spark.sql(String.format("CREATE VIEW simple_view AS SELECT * FROM %s",
SCHEMA_TABLE_NAME));
+ spark.sql(String.format("ALTER TABLE %s ALTER COLUMN b FIRST",
SCHEMA_TABLE_NAME));
+
+ assertThat(rowsToJava(spark.sql("SELECT * FROM
simple_view").collectAsList()))
+ .containsExactlyInAnyOrder(new Object[] {1, 10}, new Object[] {2, 20});
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void schemaEvolutionViewFailsAfterSchemaChange(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createSchemaTable();
+ spark.sql(
+ String.format(
+ "CREATE VIEW simple_view WITH SCHEMA EVOLUTION AS SELECT * FROM
%s",
+ SCHEMA_TABLE_NAME));
+ spark.sql(String.format("ALTER TABLE %s ADD COLUMN c INT",
SCHEMA_TABLE_NAME));
+
+ assertFailsWithCondition(
+ () -> spark.sql("SELECT * FROM simple_view").collectAsList(),
"FAILED_UPDATE_VIEW_SCHEMA");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void typeEvolutionViewFailsAfterTypeChange(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createSchemaTable();
+ spark.sql(
+ String.format(
+ "CREATE VIEW simple_view WITH SCHEMA TYPE EVOLUTION AS SELECT *
FROM %s",
+ SCHEMA_TABLE_NAME));
+ spark.sql(String.format("ALTER TABLE %s ALTER COLUMN a TYPE BIGINT",
SCHEMA_TABLE_NAME));
+
+ assertFailsWithCondition(
+ () -> spark.sql("SELECT * FROM simple_view").collectAsList(),
"FAILED_UPDATE_VIEW_SCHEMA");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void viewTextParsesWithStoredSqlConfigs(boolean referencedByEnabled) {
+ setReferencedByEnabled(referencedByEnabled);
+ createSchemaTable();
+ spark.conf().set("spark.sql.ansi.doubleQuotedIdentifiers", "true");
+ try {
+ spark.sql(
+ String.format("CREATE VIEW simple_view AS SELECT \"a\" FROM %s",
SCHEMA_TABLE_NAME));
+ } finally {
+ spark.conf().unset("spark.sql.ansi.doubleQuotedIdentifiers");
+ }
+
+ assertThat(rowsToJava(spark.sql("SELECT * FROM
simple_view").collectAsList()))
+ .containsExactlyInAnyOrder(new Object[] {1}, new Object[] {2});
+ }
+
+ private void createSchemaTable() {
+ spark.sql(String.format("CREATE TABLE %s (a INT, b INT)",
SCHEMA_TABLE_NAME));
+ spark.sql(String.format("INSERT INTO %s VALUES (1, 10), (2, 20)",
SCHEMA_TABLE_NAME));
+ }
+
+ private void setReferencedByEnabled(boolean enabled) {
+ spark.conf().set(SparkSQLProperties.REFERENCED_BY_ENABLED,
String.valueOf(enabled));
+ }
+
+ private void assertFailsWithCondition(ThrowingCallable action, String
condition) {
+ assertThatThrownBy(action)
+ .isInstanceOf(SparkThrowable.class)
+ .hasMessageContaining("[" + condition + "]");
+ }
+
+ private String createSessionCatalogParquetTable() {
+ String parquetTable =
+ String.format("%s.%s.%s", SESSION_CATALOG_NAME, NAMESPACE,
PARQUET_TABLE_NAME);
+ spark.sql(String.format("CREATE TABLE %s (id INT) USING parquet",
parquetTable));
+ spark.sql(String.format("INSERT INTO %s VALUES (1), (2), (3)",
parquetTable));
+ return parquetTable;
+ }
+
+ private long currentSnapshotMillis() throws Exception {
+ return Spark3Util.loadIcebergTable(spark,
TABLE_NAME).currentSnapshot().timestampMillis();
+ }
+
+ private List<ContextTrackingCatalog.CapturedContext>
capturedLoadsOf(TableIdentifier ident) {
+ return capturedLoadsOf(ContextTrackingCatalog.CAPTURED, ident);
+ }
+
+ private List<ContextTrackingCatalog.CapturedContext> capturedLoadsOf(
+ List<ContextTrackingCatalog.CapturedContext> captures, TableIdentifier
ident) {
+ return captures.stream()
+ .filter(captured -> captured.tableIdentifier.equals(ident))
+ .collect(Collectors.toList());
+ }
+
+ private void createView(String viewName, String sql) {
+ createView(CATALOG_NAME, viewName, sql);
+ }
+
+ private void createView(String catalogName, String viewName, String sql) {
+ createView(catalogName, viewName, sql,
SparkSchemaUtil.convert(spark.sql(sql).schema()));
+ }
+
+ private void createViewWithoutAnalysis(String viewName, String sql) {
+ createView(
+ CATALOG_NAME,
+ viewName,
+ sql,
+ new Schema(Types.NestedField.optional(1, "id",
Types.IntegerType.get())));
+ }
+
+ private void createView(String catalogName, String viewName, String sql,
Schema schema) {
+ viewCatalog(catalogName)
+ .buildView(TableIdentifier.of(NAMESPACE, viewName))
+ .withQuery("spark", sql)
+ .withDefaultNamespace(NAMESPACE)
+ .withDefaultCatalog(catalogName)
+ .withSchema(schema)
+ .create();
+ }
+
+ private void createCrossCatalogView() {
+ String otherTable = String.format("%s.%s.%s", OTHER_CATALOG_NAME,
NAMESPACE, OTHER_TABLE_NAME);
+ spark.sql(String.format("CREATE TABLE %s (id INT, data STRING)",
otherTable));
+ appendRecords(otherTable);
+ spark.sql(String.format("REFRESH TABLE %s", otherTable));
+ createView("cross_view", String.format("SELECT id FROM %s", otherTable));
+ spark.sql(String.format("REFRESH TABLE %s", otherTable));
+ }
+
+ private void appendRecords(String tableName) {
+ try {
+ List<SimpleRecord> records =
+ IntStream.rangeClosed(1, 5)
+ .mapToObj(i -> new SimpleRecord(i, String.valueOf(i)))
+ .collect(Collectors.toList());
+ Dataset<Row> df = spark.createDataFrame(records, SimpleRecord.class);
+ df.writeTo(tableName).append();
+ } catch (NoSuchTableException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void assertCapturedTableChain(
+ List<ContextTrackingCatalog.CapturedContext> captures,
+ String targetName,
+ String... expectedViewNames) {
+ assertCapturedTableChain(captures, NAMESPACE, targetName,
expectedViewNames);
+ }
+
+ private void assertCapturedTableChain(
+ List<ContextTrackingCatalog.CapturedContext> captures,
+ Namespace targetNamespace,
+ String targetName,
+ String... expectedViewNames) {
+ List<ContextTrackingCatalog.CapturedContext> matching =
+ capturedLoadsOf(captures, TableIdentifier.of(targetNamespace,
targetName)).stream()
+ .filter(captured -> !captured.referencedBy.isEmpty())
+ .collect(Collectors.toList());
+
+ assertThat(matching).isNotEmpty();
+ assertThat(matching.get(0).referencedBy)
+ .containsExactly(
+ Arrays.stream(expectedViewNames)
+ .map(name -> TableIdentifier.of(NAMESPACE, name))
+ .toArray(TableIdentifier[]::new));
+ }
+
+ private ViewCatalog viewCatalog(String catalogName) {
+ CatalogPlugin catalogPlugin =
spark.sessionState().catalogManager().catalog(catalogName);
+ assertThat(catalogPlugin).isInstanceOf(HasIcebergCatalog.class);
+ ViewCatalog viewCatalog = ((HasIcebergCatalog)
catalogPlugin).icebergViewCatalog();
+ assertThat(viewCatalog).isNotNull();
+ return viewCatalog;
+ }
+}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
index c2a43306ce..df9a62e609 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
@@ -25,6 +25,7 @@ import org.apache.iceberg.spark.source.HasIcebergCatalog;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.catalyst.analysis.NoSuchViewException;
+import org.apache.spark.sql.catalyst.analysis.ViewUtil;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.ProcedureCatalog;
import org.apache.spark.sql.connector.catalog.Relation;
@@ -108,6 +109,12 @@ abstract class BaseCatalog
try {
return loadTable(ident);
} catch (NoSuchTableException e) {
+ if (ViewUtil.referencedByEnabled()) {
+ // Don't return views: Spark's own view expansion can't pass the
referenced-by chain to the
+ // catalog. ResolveReferencedByViews expands Iceberg views that are
left unresolved instead.
+ throw e;
+ }
+
try {
return loadView(ident);
} catch (NoSuchViewException viewException) {
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
index a0bdfea817..45f44fa8e9 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
@@ -44,6 +44,7 @@ import org.apache.iceberg.TableUtil;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.UnboundPartitionSpec;
import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.LoadContext;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -126,7 +127,7 @@ import org.slf4j.LoggerFactory;
*
* <p>
*/
-public class SparkCatalog extends BaseCatalog {
+public class SparkCatalog extends BaseCatalog implements
SparkSupportsLoadContext {
private static final Logger LOG =
LoggerFactory.getLogger(SparkCatalog.class);
private static final Set<String> DEFAULT_NS_KEYS =
ImmutableSet.of(TableCatalog.PROP_OWNER);
@@ -194,17 +195,34 @@ public class SparkCatalog extends BaseCatalog {
@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
@@ -691,9 +709,15 @@ public class SparkCatalog extends BaseCatalog {
@Override
public View loadView(Identifier ident) throws NoSuchViewException {
+ return loadView(ident, LoadContext.empty());
+ }
+
+ @Override
+ public View loadView(Identifier ident, LoadContext context) throws
NoSuchViewException {
if (null != asViewCatalog) {
try {
- org.apache.iceberg.view.View icebergView =
asViewCatalog.loadView(buildIdentifier(ident));
+ org.apache.iceberg.view.View icebergView =
+ asViewCatalog.loadView(buildIdentifier(ident), context);
return SparkView.toView(catalogName, icebergView);
} catch (org.apache.iceberg.exceptions.NoSuchViewException e) {
throw new NoSuchViewException(ident);
@@ -1055,13 +1079,14 @@ public class SparkCatalog extends BaseCatalog {
}
}
- private Table load(Identifier ident, TimeTravel timeTravel) throws
NoSuchTableException {
+ private Table load(Identifier ident, TimeTravel timeTravel, LoadContext
context)
+ throws NoSuchTableException {
if (isPathIdentifier(ident)) {
return loadPath((PathIdentifier) ident, timeTravel);
}
try {
- org.apache.iceberg.Table table =
icebergCatalog.loadTable(buildIdentifier(ident));
+ org.apache.iceberg.Table table =
icebergCatalog.loadTable(buildIdentifier(ident), context);
return SparkTable.create(table, timeTravel);
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
@@ -1074,7 +1099,7 @@ public class SparkCatalog extends BaseCatalog {
TableIdentifier namespaceAsIdent =
buildIdentifier(namespaceToIdentifier(ident.namespace()));
org.apache.iceberg.Table table;
try {
- table = icebergCatalog.loadTable(namespaceAsIdent);
+ table = icebergCatalog.loadTable(namespaceAsIdent, context);
} catch (Exception ignored) {
// the namespace does not identify a table, so it cannot be a table
with a snapshot selector
// throw an exception for the original identifier
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
index 428644ecfc..84ef44dca5 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
@@ -107,6 +107,11 @@ public class SparkSQLProperties {
// Rely on identifier fields as a PRIMARY KEY constraint for query
optimization (not enforced)
public static final String IDENTIFIER_FIELDS_RELY =
"spark.sql.iceberg.identifier-fields-rely";
+ // Controls whether to send referenced-by view chains while loading
relations; requires the
+ // Iceberg SQL extensions, which expand Iceberg views when this is enabled
+ public static final String REFERENCED_BY_ENABLED =
"spark.sql.iceberg.referenced-by.enabled";
+ public static final boolean REFERENCED_BY_ENABLED_DEFAULT = false;
+
// Prefix for custom snapshot properties
public static final String SNAPSHOT_PROPERTY_PREFIX =
"spark.sql.iceberg.snapshot-property.";
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
index f86e897bcf..37c3fd2a2f 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.LoadContext;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -68,12 +69,13 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
*/
public class SparkSessionCatalog<
T extends TableCatalog & FunctionCatalog & SupportsNamespaces &
ViewCatalog>
- extends BaseCatalog implements CatalogExtension {
+ extends BaseCatalog implements CatalogExtension, SparkSupportsLoadContext {
private String catalogName = null;
private TableCatalog icebergCatalog = null;
private StagingTableCatalog asStagingCatalog = null;
private ViewCatalog asViewCatalog = null;
+ private SparkSupportsLoadContext asLoadContextCatalog = null;
private T sessionCatalog = null;
private boolean createParquetAsIceberg = false;
private boolean createAvroAsIceberg = false;
@@ -198,7 +200,17 @@ public class SparkSessionCatalog<
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ return loadTable(ident, LoadContext.empty());
+ }
+
+ @Override
+ public Table loadTable(Identifier ident, LoadContext context) throws
NoSuchTableException {
try {
+ if (asLoadContextCatalog != null) {
+ return asLoadContextCatalog.loadTable(ident, context);
+ }
+
+ checkLoadContextSupported("table", ident, context);
return icebergCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
return getSessionCatalog().loadTable(ident);
@@ -207,7 +219,18 @@ public class SparkSessionCatalog<
@Override
public Table loadTable(Identifier ident, String version) throws
NoSuchTableException {
+ return loadTable(ident, version, LoadContext.empty());
+ }
+
+ @Override
+ public Table loadTable(Identifier ident, String version, LoadContext context)
+ throws NoSuchTableException {
try {
+ if (asLoadContextCatalog != null) {
+ return asLoadContextCatalog.loadTable(ident, version, context);
+ }
+
+ checkLoadContextSupported("table", ident, context);
return icebergCatalog.loadTable(ident, version);
} catch (NoSuchTableException e) {
return getSessionCatalog().loadTable(ident, version);
@@ -216,7 +239,18 @@ public class SparkSessionCatalog<
@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 (asLoadContextCatalog != null) {
+ return asLoadContextCatalog.loadTable(ident, timestamp, context);
+ }
+
+ checkLoadContextSupported("table", ident, context);
return icebergCatalog.loadTable(ident, timestamp);
} catch (NoSuchTableException e) {
return getSessionCatalog().loadTable(ident, timestamp);
@@ -426,6 +460,10 @@ public class SparkSessionCatalog<
this.asViewCatalog = (ViewCatalog) icebergCatalog;
}
+ if (icebergCatalog instanceof SparkSupportsLoadContext) {
+ this.asLoadContextCatalog = (SparkSupportsLoadContext) icebergCatalog;
+ }
+
this.createParquetAsIceberg = options.getBoolean("parquet-enabled",
createParquetAsIceberg);
this.createAvroAsIceberg = options.getBoolean("avro-enabled",
createAvroAsIceberg);
this.createOrcAsIceberg = options.getBoolean("orc-enabled",
createOrcAsIceberg);
@@ -594,15 +632,39 @@ public class SparkSessionCatalog<
@Override
public View loadView(Identifier ident) throws NoSuchViewException {
- if (null != asViewCatalog && asViewCatalog.viewExists(ident)) {
+ return loadView(ident, LoadContext.empty());
+ }
+
+ @Override
+ public View loadView(Identifier ident, LoadContext context) throws
NoSuchViewException {
+ if (null != asViewCatalog && null != asLoadContextCatalog) {
+ try {
+ return asLoadContextCatalog.loadView(ident, context);
+ } catch (NoSuchViewException ignored) {
+ // not an Iceberg view, so check the session catalog below
+ }
+ } else if (null != asViewCatalog && asViewCatalog.viewExists(ident)) {
+ checkLoadContextSupported("view", ident, context);
return asViewCatalog.loadView(ident);
- } else if (isViewCatalog() && getSessionCatalog().viewExists(ident)) {
+ }
+
+ if (isViewCatalog() && getSessionCatalog().viewExists(ident)) {
return getSessionCatalog().loadView(ident);
}
throw new NoSuchViewException(ident);
}
+ private void checkLoadContextSupported(
+ String relationType, Identifier ident, LoadContext context) {
+ if (!context.referencedBy().isEmpty()) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Cannot load %s %s with referenced-by: %s does not support load
context",
+ relationType, ident, icebergCatalog.getClass().getName()));
+ }
+ }
+
@Override
public View createView(Identifier ident, View view)
throws ViewAlreadyExistsException, NoSuchNamespaceException {
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSupportsLoadContext.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSupportsLoadContext.java
new file mode 100644
index 0000000000..76f2a12487
--- /dev/null
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSupportsLoadContext.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import org.apache.iceberg.catalog.LoadContext;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.analysis.NoSuchViewException;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.View;
+
+/**
+ * A Spark Catalog API extension for loading tables and views with additional
load context. Extends
+ * the loading contract to accept a {@link LoadContext} that carries the view
reference chain and
+ * other contextual information.
+ */
+public interface SparkSupportsLoadContext {
+
+ /**
+ * Load a table with additional load context.
+ *
+ * @param identifier a table identifier
+ * @param context the load context
+ * @return instance of {@link Table} referred by the identifier
+ * @throws NoSuchTableException if the table does not exist
+ */
+ Table loadTable(Identifier identifier, LoadContext context) throws
NoSuchTableException;
+
+ /**
+ * Load a table at a specific version with additional load context.
+ *
+ * @param identifier a table identifier
+ * @param version the version string
+ * @param context the load context
+ * @return instance of {@link Table} referred by the identifier at the
specified version
+ * @throws NoSuchTableException if the table does not exist
+ */
+ Table loadTable(Identifier identifier, String version, LoadContext context)
+ throws NoSuchTableException;
+
+ /**
+ * Load a table at a specific timestamp with additional load context.
+ *
+ * @param identifier a table identifier
+ * @param timestamp the timestamp in microseconds
+ * @param context the load context
+ * @return instance of {@link Table} referred by the identifier as of the
specified timestamp
+ * @throws NoSuchTableException if the table does not exist
+ */
+ Table loadTable(Identifier identifier, long timestamp, LoadContext context)
+ throws NoSuchTableException;
+
+ /**
+ * Load a view with additional load context.
+ *
+ * @param identifier a view identifier
+ * @param context the load context
+ * @return instance of {@link View} referred by the identifier
+ * @throws NoSuchViewException if the view does not exist
+ */
+ View loadView(Identifier identifier, LoadContext context) throws
NoSuchViewException;
+}
diff --git
a/spark/v4.2/spark/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewUtil.scala
b/spark/v4.2/spark/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewUtil.scala
index e7a7060097..cff270eb46 100644
---
a/spark/v4.2/spark/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewUtil.scala
+++
b/spark/v4.2/spark/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewUtil.scala
@@ -18,20 +18,28 @@
*/
package org.apache.spark.sql.catalyst.analysis
+import org.apache.iceberg.catalog.LoadContext
+import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.spark.Spark3Util
+import org.apache.iceberg.spark.SparkSQLProperties
+import org.apache.iceberg.spark.SparkSupportsLoadContext
import org.apache.iceberg.spark.source.HasIcebergCatalog
import org.apache.iceberg.spark.source.SparkView
import org.apache.iceberg.view.{View => IcebergView}
import org.apache.spark.sql.connector.catalog.CatalogPlugin
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.catalog.Table
import org.apache.spark.sql.connector.catalog.TableCatalog
import org.apache.spark.sql.connector.catalog.V1View
import org.apache.spark.sql.connector.catalog.View
import org.apache.spark.sql.connector.catalog.ViewCatalog
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
import scala.jdk.CollectionConverters._
object ViewUtil {
+
val RESERVED_PROPERTIES: Seq[String] =
Seq(
TableCatalog.PROP_COMMENT,
@@ -62,6 +70,17 @@ object ViewUtil {
case _ => None
}
+ def loadView(catalog: CatalogPlugin, ident: Identifier, context:
LoadContext): Option[View] =
+ catalog match {
+ case supportsLoadContext: SparkSupportsLoadContext =>
+ try {
+ Option(supportsLoadContext.loadView(ident,
context)).filter(isIcebergView)
+ } catch {
+ case _: NoSuchViewException => None
+ }
+ case _ => loadView(catalog, ident)
+ }
+
def loadIcebergView(catalog: CatalogPlugin, ident: Identifier):
Option[IcebergView] =
catalog match {
case catalogWithIceberg: HasIcebergCatalog =>
@@ -73,6 +92,69 @@ 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(parts.tail: _*)
+ }
+
+ 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 loadTable(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ context: LoadContext,
+ timeTravelSpec: Option[TimeTravelSpec]): Table = {
+ catalog match {
+ case supportsLoadContext: SparkSupportsLoadContext =>
+ timeTravelSpec match {
+ case Some(AsOfVersion(version)) =>
supportsLoadContext.loadTable(ident, version, context)
+ case Some(AsOfTimestamp(micros)) =>
supportsLoadContext.loadTable(ident, micros, context)
+ case None => supportsLoadContext.loadTable(ident, context)
+ }
+ case _ => CatalogV2Util.getTable(catalog, ident, timeTravelSpec)
+ }
+ }
+
+ def referencedByEnabled: Boolean = {
+ java.lang.Boolean.parseBoolean(
+ SQLConf.get.getConfString(
+ SparkSQLProperties.REFERENCED_BY_ENABLED,
+ SparkSQLProperties.REFERENCED_BY_ENABLED_DEFAULT.toString))
+ }
+
def isIcebergViewCatalog(catalog: CatalogPlugin): Boolean = {
catalog.isInstanceOf[ViewCatalog] &&
catalog.isInstanceOf[HasIcebergCatalog]
}
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
index 61b0eaa15d..a7373fcdae 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
@@ -32,6 +32,8 @@ import static org.mockito.Mockito.withSettings;
import java.nio.file.Path;
import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.catalog.LoadContext;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
@@ -631,6 +633,49 @@ public class TestSparkSessionCatalog extends TestBase {
verify(icebergViewCatalog, never()).createOrReplaceView(viewIdent,
replacement);
}
+ @Test
+ void loadTableWithReferencedByFailsWhenCatalogCannotPassContext() {
+ Identifier ident = Identifier.of(new String[] {"default"}, "table");
+ TableCatalog icebergCatalog = icebergCatalogWithViews();
+ SparkSessionCatalog<?> catalog = catalogWithViews(icebergCatalog,
sessionCatalogWithViews());
+ LoadContext context =
+
LoadContext.builder().referencedBy(List.of(TableIdentifier.of("default",
"view"))).build();
+
+ assertThatThrownBy(() -> catalog.loadTable(ident, context))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage(
+ "Cannot load table default.table with referenced-by: %s does not
support load context",
+ icebergCatalog.getClass().getName());
+ }
+
+ @Test
+ void loadTableWithoutReferencedByUsesCatalogWithoutContext() throws
NoSuchTableException {
+ Identifier ident = Identifier.of(new String[] {"default"}, "table");
+ TableCatalog icebergCatalog = icebergCatalogWithViews();
+ SparkSessionCatalog<?> catalog = catalogWithViews(icebergCatalog,
sessionCatalogWithViews());
+
+ catalog.loadTable(ident, LoadContext.empty());
+
+ verify(icebergCatalog).loadTable(ident);
+ }
+
+ @Test
+ void loadViewWithReferencedByDoesNotCheckExistenceWithoutContext() throws
NoSuchViewException {
+ Identifier ident = Identifier.of(new String[] {"default"}, "view");
+ TableCatalog icebergCatalog =
+ mock(
+ TableCatalog.class,
+ withSettings().extraInterfaces(ViewCatalog.class,
SparkSupportsLoadContext.class));
+ SparkSessionCatalog<?> catalog = catalogWithViews(icebergCatalog,
sessionCatalogWithViews());
+ LoadContext context =
+
LoadContext.builder().referencedBy(List.of(TableIdentifier.of("default",
"outer"))).build();
+
+ catalog.loadView(ident, context);
+
+ verify((SparkSupportsLoadContext) icebergCatalog).loadView(ident, context);
+ verify((ViewCatalog) icebergCatalog, never()).viewExists(any());
+ }
+
private static TableCatalog sessionCatalogWithViews() {
return mock(
TableCatalog.class,
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCatalog.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCatalog.java
index ce24e173c6..9416535b02 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCatalog.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkCatalog.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.util.Map;
+import org.apache.iceberg.catalog.LoadContext;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -66,6 +67,11 @@ public class TestSparkCatalog<
return new SparkTable(table);
}
+ @Override
+ public Table loadTable(Identifier ident, LoadContext context) throws
NoSuchTableException {
+ return loadTable(ident);
+ }
+
public static void clearTables() {
TABLE_MAP.clear();
}
