[SPARK-58389][SQL][FOLLOWUP] Pin DSv2 table instances by state options during analysis

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsTableStateOptions.java

@@ -0,0 +1,52 @@
+/*
+ * 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.connector.catalog;
+
+import java.util.Set;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A catalog capability for identifying options that select a table's state.
+ * <p>
+ * Spark may resolve the same table more than once while analyzing or refreshing one query. A
+ * catalog can implement this interface to declare which raw read options may cause
+ * {@link TableCatalog#loadTable(Identifier, TableContext,
+ * org.apache.spark.sql.util.CaseInsensitiveStringMap)} to select a different table state, such as
+ * a branch, tag, snapshot, or version. Spark can then reuse one concrete {@link Table} instance
+ * for references whose table-state options match while preserving every reference's complete
+ * option map for scan planning.
+ * <p>
+ * Option key matching is case-insensitive. Option values remain case-sensitive. Parsed Spark time
+ * travel is handled independently and must not be included in the returned set.
+ * <p>
+ * Catalogs that do not implement this capability are handled conservatively: Spark treats every
+ * raw option as table-state-affecting.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface SupportsTableStateOptions extends CatalogPlugin {
+
+  /**
+   * Returns the raw option keys that may affect the table state selected by {@code loadTable}.
+   *
+   * @return a non-null set of case-insensitive option keys
+   */
+  Set<String> tableStateOptionKeys();
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -135,6 +135,9 @@ object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog with Suppo
  * @param relationCache A mapping from (qualified table name, time travel spec, options) to
  *                      resolved relations. This can ensure that the table is resolved only once if
  *                      a table is used multiple times in a query with the same options.
+ * @param tableCache A mapping from (catalog, identifier, time travel spec, table-state options) to
+ *                   concrete tables. This pins one table state while allowing references to keep
+ *                   different read-specific options.
  * @param referredTempViewNames All the temp view names referred by the current view we are
  *                              resolving. It's used to make sure the relation resolution is
  *                              consistent between view creation and view resolution. For example,
@@ -155,6 +158,7 @@ case class AnalysisContext(
     nestedViewDepth: Int = 0,
     maxNestedViewDepth: Int = -1,
     relationCache: mutable.Map[RelationCacheKey, LogicalPlan] = mutable.Map.empty,
+    tableCache: mutable.Map[TableCacheKey, Table] = mutable.Map.empty,
     referredTempViewNames: Seq[Seq[String]] = Seq.empty,
     // 1. If we are resolving a view, this field will be restored from the view metadata,
     //    by calling `AnalysisContext.withAnalysisContext(viewDesc)`.
@@ -249,6 +253,7 @@ object AnalysisContext {
       nestedViewDepth = originContext.nestedViewDepth + 1,
       maxNestedViewDepth = maxNestedViewDepth,
       relationCache = originContext.relationCache,
+      tableCache = originContext.tableCache,
       referredTempViewNames = viewDesc.viewReferredTempViewNames,
       referredTempFunctionNames = mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
       referredTempVariableNames = viewDesc.viewReferredTempVariableNames,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala

@@ -62,6 +62,7 @@ class RelationResolution(
   val v1SessionCatalog = catalogManager.v1SessionCatalog
 
   private def relationCache = AnalysisContext.get.relationCache
+  private def tableCache = AnalysisContext.get.tableCache
 
   /**
    * If we are resolving database objects (relations, functions, etc.) inside views, we may need to
@@ -254,6 +255,10 @@ class RelationResolution(
         cached
           .map(adaptCachedRelation(_, planId))
           .orElse {
+            lazy val tableKey =
+              toTableCacheKey(catalog, ident, finalTimeTravelSpec, finalOptions)
+            val pinnedTable = if (writePrivileges == null) tableCache.get(tableKey) else None
+
             // For a `RelationCatalog` with no time-travel / write privileges, the single-RPC
             // `loadRelation` answers both "is there a table?" and "is there a view?" in one
             // call. Time-travel and write privileges apply to tables only, so for those the
@@ -263,45 +268,48 @@ class RelationResolution(
             // Skip the table-side lookup entirely for view-only catalogs (no `TableCatalog`
             // mixin): `CatalogV2Util.loadTable` would call `asTableCatalog` and throw
             // MISSING_CATALOG_ABILITY.TABLES, masking the legitimate view-resolution path.
-            val relation: Option[Relation] = catalog match {
-              case mc: RelationCatalog if finalTimeTravelSpec.isEmpty && writePrivileges == null =>
-                try {
-                  Some(mc.loadRelation(ident))
-                } catch {
-                  case _: NoSuchTableException => None
-                }
-              case _ =>
-                val tableSide: Option[Table] = if (
-                  CatalogV2Util.isSessionCatalog(catalog) || catalog.isInstanceOf[TableCatalog]
-                ) {
-                  CatalogV2Util.loadTable(
-                    catalog,
-                    ident,
-                    finalTimeTravelSpec,
-                    Option(writePrivileges),
-                    finalOptions)
-                } else {
-                  None
-                }
-                // Fallback to ViewCatalog for catalogs that host views but where loadTable
-                // returned None (or was skipped because there's no TableCatalog mixin).
-                // Time-travel / write privileges only apply to tables, not views, so the
-                // fallback only fires when both are absent.
-                tableSide.orElse {
-                  if (finalTimeTravelSpec.isEmpty && writePrivileges == null) {
-                    catalog match {
-                      case vc: ViewCatalog =>
-                        try {
-                          Some(vc.loadView(ident))
-                        } catch {
-                          case _: NoSuchViewException => None
-                        }
-                      case _ => None
-                    }
+            val relation: Option[Relation] = pinnedTable.orElse {
+              catalog match {
+                case mc: RelationCatalog
+                    if finalTimeTravelSpec.isEmpty && writePrivileges == null =>
+                  try {
+                    Some(mc.loadRelation(ident))
+                  } catch {
+                    case _: NoSuchTableException => None
+                  }
+                case _ =>
+                  val tableSide: Option[Table] = if (
+                    CatalogV2Util.isSessionCatalog(catalog) || catalog.isInstanceOf[TableCatalog]
+                  ) {
+                    CatalogV2Util.loadTable(
+                      catalog,
+                      ident,
+                      finalTimeTravelSpec,
+                      Option(writePrivileges),
+                      finalOptions)
                   } else {
                     None
                   }
-                }
+                  // Fallback to ViewCatalog for catalogs that host views but where loadTable
+                  // returned None (or was skipped because there's no TableCatalog mixin).
+                  // Time-travel / write privileges only apply to tables, not views, so the
+                  // fallback only fires when both are absent.
+                  tableSide.orElse {
+                    if (finalTimeTravelSpec.isEmpty && writePrivileges == null) {
+                      catalog match {
+                        case vc: ViewCatalog =>
+                          try {
+                            Some(vc.loadView(ident))
+                          } catch {
+                            case _: NoSuchViewException => None
+                          }
+                        case _ => None
+                      }
+                    } else {
+                      None
+                    }
+                  }
+              }
             }
             // `table` is `relation` filtered to tables only -- used for cache lookup since
             // we don't share-cache views.
@@ -312,12 +320,14 @@ class RelationResolution(
             // `Table`.
             val sharedRelationCacheMatch = for {
               t <- table
-              if finalTimeTravelSpec.isEmpty && writePrivileges == null && !u.isStreaming
+              if pinnedTable.isEmpty && finalTimeTravelSpec.isEmpty &&
+                writePrivileges == null && !u.isStreaming
               cached <- lookupSharedRelationCache(catalog, ident, t)
               if cached.options == finalOptions
             } yield {
               val nameParts = ident.toQualifiedNameParts(catalog)
               val aliasedRelation = SubqueryAlias(nameParts, cached)
+              tableCache.update(tableKey, cached.table)
               relationCache.update(key, aliasedRelation)
               adaptCachedRelation(aliasedRelation, planId)
             }
@@ -330,6 +340,9 @@ class RelationResolution(
                 finalOptions,
                 u.isStreaming,
                 finalTimeTravelSpec)
+              if (writePrivileges == null && pinnedTable.isEmpty) {
+                table.foreach(tableCache.update(tableKey, _))
+              }
               loaded.foreach(relationCache.update(key, _))
               loaded.map(cloneWithPlanId(_, planId))
             }
@@ -476,23 +489,60 @@ class RelationResolution(
 
   def resolveReference(ref: V2TableReference): LogicalPlan = {
     val relation = if (ref.context.cacheable) {
-      getOrLoadRelation(ref)
+      // A temporary view may contain a relation pinned by CacheManager, so its re-resolution
+      // consults sharedRelationCache to preserve that Table. Transaction references use the
+      // Table loaded through the transaction catalog instead.
+      val useSharedRelationCache =
+        ref.context.isInstanceOf[V2TableReference.TemporaryViewContext]
+      getOrLoadRelation(ref, useSharedRelationCache)
     } else {
       loadRelation(ref)
     }
     val planId = ref.getTagValue(LogicalPlan.PLAN_ID_TAG)
     cloneWithPlanId(relation, planId)
   }
 
-  private def getOrLoadRelation(ref: V2TableReference): LogicalPlan = {
+  private def getOrLoadRelation(
+      ref: V2TableReference,
+      useSharedRelationCache: Boolean): LogicalPlan = {
     val key = toCacheKey(ref.catalog, ref.identifier, None, ref.options)
     relationCache.get(key) match {
       case Some(cached) =>
         adaptCachedRelation(cached, ref)
       case None =>
-        val relation = loadRelation(ref)
-        relationCache.update(key, relation)
-        relation
+        val resolvedCatalog = catalogManager.catalog(ref.catalog.name).asTableCatalog
+        val tableKey = toTableCacheKey(resolvedCatalog, ref.identifier, None, ref.options)
+        tableCache.get(tableKey) match {
+          case Some(pinnedTable) =>
+            val relation = createRelation(ref, resolvedCatalog, pinnedTable)
+            relationCache.update(key, relation)
+            relation
+          case None =>
+            val loadedTable = CatalogV2Util.getTable(
+              resolvedCatalog,
+              ref.identifier,
+              options = ref.options)
+            val sharedRelationCacheMatch = if (useSharedRelationCache) {
+              lookupSharedRelationCache(
+                resolvedCatalog,
+                ref.identifier,
+                loadedTable).filter(_.options == ref.options)
+            } else {
+              None
+            }
+            sharedRelationCacheMatch match {
+              case Some(cached) =>
+                val relation = adaptCachedRelation(cached, ref)
+                tableCache.update(tableKey, cached.table)
+                relationCache.update(key, relation)
+                relation
+              case None =>
+                val relation = createRelation(ref, resolvedCatalog, loadedTable)
+                tableCache.update(tableKey, loadedTable)
+                relationCache.update(key, relation)
+                relation
+            }
+        }
     }
   }
 
@@ -508,6 +558,13 @@ class RelationResolution(
   private def loadRelation(ref: V2TableReference): LogicalPlan = {
     val resolvedCatalog = catalogManager.catalog(ref.catalog.name).asTableCatalog
     val table = resolvedCatalog.loadTable(ref.identifier)
+    createRelation(ref, resolvedCatalog, table)
+  }
+
+  private def createRelation(
+      ref: V2TableReference,
+      resolvedCatalog: TableCatalog,
+      table: Table): DataSourceV2Relation = {
     V2TableReferenceUtils.validateLoadedTable(table, ref)
     DataSourceV2Relation(
       table = table,
@@ -551,6 +608,18 @@ class RelationResolution(
     RelationCacheKey(nameParts, timeTravelSpec, options)
   }
 
+  private def toTableCacheKey(
+      catalog: CatalogPlugin,
+      ident: Identifier,
+      timeTravelSpec: Option[TimeTravelSpec],
+      options: CaseInsensitiveStringMap): TableCacheKey = {
+    TableCacheKey(
+      catalog,
+      ident,
+      timeTravelSpec,
+      CatalogV2Util.tableStateOptions(catalog, options))
+  }
+
   private def cloneWithPlanId(plan: LogicalPlan, planId: Option[Long]): LogicalPlan = {
     planId match {
       case Some(id) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableCacheKey.scala

@@ -0,0 +1,33 @@
+/*
+ * 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.spark.sql.connector.catalog.{CatalogPlugin, Identifier}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Key for the per-query table-state cache in [[AnalysisContext]].
+ *
+ * Unlike [[RelationCacheKey]], this key contains only options declared to affect table state. This
+ * lets references retain different scan options while sharing one concrete table state.
+ */
+private[sql] case class TableCacheKey(
+    catalog: CatalogPlugin,
+    identifier: Identifier,
+    timeTravelSpec: Option[TimeTravelSpec],
+    stateOptions: CaseInsensitiveStringMap)

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala

@@ -481,6 +481,29 @@ private[sql] object CatalogV2Util {
       case _: NoSuchDatabaseException => None
     }
 
+  /**
+   * Projects a complete read option map to the options that may select table state.
+   *
+   * Catalogs must explicitly opt in to projection. For all other catalogs, every option is kept
+   * so that reusing a concrete table cannot silently combine states the catalog considers
+   * different.
+   */
+  def tableStateOptions(
+      catalog: CatalogPlugin,
+      options: CaseInsensitiveStringMap): CaseInsensitiveStringMap = catalog match {
+    case supports: SupportsTableStateOptions =>
+      val stateKeys = supports.tableStateOptionKeys().asScala
+        .map(_.toLowerCase(Locale.ROOT))
+        .toSet
+      val projected = options.entrySet().asScala.collect {
+        case entry if stateKeys.contains(entry.getKey.toLowerCase(Locale.ROOT)) =>
+          entry.getKey -> entry.getValue
+      }.toMap
+      new CaseInsensitiveStringMap(projected.asJava)
+    case _ =>
+      options
+  }
+
   def getTable(
       catalog: CatalogPlugin,
       ident: Identifier,

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TableLookupCacheSuite.scala

@@ -119,4 +119,21 @@ class TableLookupCacheSuite extends AnalysisTest with Matchers {
       verify(catalog, times(1)).getTable("default", "t1")
     }
   }
+
+  test("nested view analysis shares both query-scoped caches") {
+    AnalysisContext.withNewAnalysisContext {
+      val outer = AnalysisContext.get
+      val viewDesc = CatalogTable(
+        TableIdentifier("view", Some("default")),
+        CatalogTableType.VIEW,
+        CatalogStorageFormat.empty,
+        StructType(Seq(StructField("a", IntegerType))),
+        viewText = Some("select * from t1"))
+
+      AnalysisContext.withAnalysisContext(viewDesc) {
+        assert(AnalysisContext.get.relationCache eq outer.relationCache)
+        assert(AnalysisContext.get.tableCache eq outer.tableCache)
+      }
+    }
+  }
 }