Spark: [3/N] Support sending referenced-by to all endpoints

api/src/main/java/org/apache/iceberg/catalog/LoadContext.java

@@ -60,6 +60,25 @@ public static LoadContext empty() {
     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;
 

core/src/main/java/org/apache/iceberg/CachingCatalog.java

@@ -29,6 +29,7 @@
 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 Table loadTable(TableIdentifier ident) {
     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);

core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java

@@ -31,6 +31,7 @@
 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.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 void testCatalogExpirationTtlRefreshesAfterAccessViaCatalog() throws IOEx
     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 static TableIdentifier[] metadataTables(TableIdentifier tableIdent) {
         .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;
+    }
+  }
 }

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)

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")
+}