[SPARK-58389][SQL] Pass all options while loading tables

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.connector.catalog;
 
+import org.apache.spark.SparkIllegalArgumentException;
 import org.apache.spark.annotation.Evolving;
 import org.apache.spark.sql.connector.expressions.Transform;
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
@@ -194,6 +195,50 @@ default Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExce
     throw QueryCompilationErrors.noSuchTableError(name(), ident);
   }
 
+  /**
+   * Load table metadata by {@link Identifier identifier} from the catalog, forwarding all
+   * user-specified options.
+   * <p>
+   * The default implementation ignores {@code options} and delegates to the existing
+   * {@code loadTable} overloads based on {@code context}. Catalogs that want to receive the user
+   * options while reading a table must override this method.
+   * <p>
+   * An override replaces that dispatch and must honor {@code context} itself: apply the time
+   * travel in {@link TableContext#timeTravel()}, and authorize the requested
+   * {@link TableContext#writePrivileges()} as it would in {@link #loadTable(Identifier, Set)}.
+   * Spark does not re-check either afterwards.
+   *
+   * @param ident a table identifier
+   * @param context the parsed load parameters (time travel, write privileges)
+   * @param options all options passed to the read, including any keys that are also parsed into
+   *                {@code context}
+   * @return the table's metadata
+   * @throws NoSuchTableException If the table doesn't exist
+   *
+   * @since 4.3.0
+   */
+  default Table loadTable(
+      Identifier ident,
+      TableContext context,
+      CaseInsensitiveStringMap options) throws NoSuchTableException {
+    if (context.timeTravel().isPresent()) {
+      TimeTravel timeTravel = context.timeTravel().get();
+      if (timeTravel instanceof TimeTravel.AsOfVersion v) {
+        return loadTable(ident, v.version());
+      } else if (timeTravel instanceof TimeTravel.AsOfTimestamp ts) {
+        return loadTable(ident, ts.micros());
+      } else {
+        throw new SparkIllegalArgumentException(
+            "INTERNAL_ERROR",
+            Map.of("message", "Unsupported time travel spec: " + timeTravel));
+      }
+    } else if (!context.writePrivileges().isEmpty()) {
+      return loadTable(ident, context.writePrivileges());
+    } else {
+      return loadTable(ident);
+    }
+  }
+
   /**
    * Load a {@link Changelog} for the given table, representing the row-level changes within the
    * range specified by {@code context}.

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

@@ -0,0 +1,85 @@
+/*
+ * 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.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.spark.SparkIllegalArgumentException;
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+/**
+ * Encapsulates the parsed, Spark-recognized parameters of a table load request, passed from the
+ * analyzer / DataFrame API to the catalog's
+ * {@link TableCatalog#loadTable(Identifier, TableContext, CaseInsensitiveStringMap)} method.
+ * <p>
+ * A load is either a read (optionally with time travel) or a write (carrying write privileges);
+ * time travel and write privileges are mutually exclusive.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public class TableContext {
+
+  // null means no time travel.
+  private final TimeTravel timeTravel;
+  // Never null; an empty set means no write privileges (i.e. a read).
+  private final Set<TableWritePrivilege> writePrivileges;
+
+  public TableContext(TimeTravel timeTravel, Set<TableWritePrivilege> privileges) {
+    this.timeTravel = timeTravel;
+    this.writePrivileges = privileges == null ? Set.of() : Set.copyOf(privileges);
+    if (timeTravel != null && !writePrivileges.isEmpty()) {
+      throw new SparkIllegalArgumentException(
+          "INTERNAL_ERROR",
+          Map.of("message", "Cannot set both time travel and write privileges"));
+    }
+  }
+
+  /** Returns the time-travel spec, or empty for a current-version read. */
+  public Optional<TimeTravel> timeTravel() {
+    return Optional.ofNullable(timeTravel);
+  }
+
+  /** Returns the requested write privileges; empty for a read. */
+  public Set<TableWritePrivilege> writePrivileges() {
+    return writePrivileges;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (!(o instanceof TableContext that)) return false;
+    return Objects.equals(timeTravel, that.timeTravel)
+        && writePrivileges.equals(that.writePrivileges);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(timeTravel, writePrivileges);
+  }
+
+  @Override
+  public String toString() {
+    return "TableContext{timeTravel=" + timeTravel +
+        ", writePrivileges=" + writePrivileges + "}";
+  }
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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 org.apache.spark.annotation.Evolving;
+
+/**
+ * A time-travel specification for reading a table as of a specific version or point in time.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public sealed interface TimeTravel permits TimeTravel.AsOfVersion, TimeTravel.AsOfTimestamp {
+
+  /**
+   * Time travel to a specific version of the table.
+   *
+   * @param version the version identifier (connector-defined)
+   */
+  record AsOfVersion(String version) implements TimeTravel {}
+
+  /**
+   * Time travel to a specific point in time.
+   *
+   * @param micros microseconds since 1970-01-01 00:00:00 UTC
+   */
+  record AsOfTimestamp(long micros) implements TimeTravel {}
+}

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

@@ -132,9 +132,9 @@ object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog with Suppo
  * @param nestedViewDepth The nested depth in the view resolution, this enables us to limit the
  *                        depth of nested views.
  * @param maxNestedViewDepth The maximum allowed depth of nested view resolution.
- * @param relationCache A mapping from qualified table names and time travel spec to resolved
- *                      relations. This can ensure that the table is resolved only once if a table
- *                      is used multiple times in a query.
+ * @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 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,
@@ -154,8 +154,7 @@ case class AnalysisContext(
     resolutionPathEntries: Option[Seq[Seq[String]]] = None,
     nestedViewDepth: Int = 0,
     maxNestedViewDepth: Int = -1,
-    relationCache: mutable.Map[(Seq[String], Option[TimeTravelSpec]), LogicalPlan] =
-      mutable.Map.empty,
+    relationCache: mutable.Map[RelationCacheKey, LogicalPlan] = 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)`.

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

@@ -0,0 +1,32 @@
+/*
+ * 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.util.CaseInsensitiveStringMap
+
+/**
+ * Key for the per-query relation cache in [[AnalysisContext]], shared by [[RelationResolution]].
+ *
+ * Options are part of the key because a catalog's options-aware `loadTable` can return a different
+ * `Table` depending on them, so two references to the same identifier with different options must
+ * not share a cached relation.
+ */
+private[sql] case class RelationCacheKey(
+    nameParts: Seq[String],
+    timeTravelSpec: Option[TimeTravelSpec],
+    options: CaseInsensitiveStringMap)

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

@@ -61,11 +61,10 @@ class RelationResolution(
     with LookupCatalog
     with SQLConfHelper {
 
-  type CacheKey = (Seq[String], Option[TimeTravelSpec])
-
   val v1SessionCatalog = catalogManager.v1SessionCatalog
 
-  private def relationCache: mutable.Map[CacheKey, LogicalPlan] = AnalysisContext.get.relationCache
+  private def relationCache: mutable.Map[RelationCacheKey, LogicalPlan] =
+    AnalysisContext.get.relationCache
 
   /**
    * If we are resolving database objects (relations, functions, etc.) inside views, we may need to
@@ -236,24 +235,26 @@ class RelationResolution(
       finalTimeTravelSpec: Option[TimeTravelSpec]): Option[LogicalPlan] = {
     expandIdentifier(identifier) match {
       case CatalogAndIdentifier(catalog, ident) =>
-        val key = toCacheKey(catalog, ident, finalTimeTravelSpec)
         val planId = u.getTagValue(LogicalPlan.PLAN_ID_TAG)
         val writePrivileges = u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
         val finalOptions = u.clearWritePrivileges.options
+        // Time travel applies to reads only; reject it on a write target (reachable via the option
+        // form, e.g. `INSERT INTO t WITH ('versionAsOf' = ...)`) with a user-facing error.
+        if (finalTimeTravelSpec.nonEmpty && writePrivileges != null) {
+          throw QueryCompilationErrors.timeTravelUnsupportedError(toSQLId(identifier))
+        }
+        val key = toCacheKey(catalog, ident, finalTimeTravelSpec, finalOptions)
         // A reference that requires write privileges is never served from the per-query relation
         // cache. The catalog authorizes the write in `loadTable(ident, writePrivileges)` below, and
         // a cache hit would skip that call entirely. The hit happens whenever the write target is
         // also read in the same statement -- the target is resolved after its query (see
         // `ResolveRelations`), so it finds the relation the query already put in the cache, e.g.
         // for `INSERT INTO t SELECT * FROM t`.
+        //
+        // The cache key includes the options, so a hit means the options already match and each
+        // reference's own bag is honored without re-applying it here.
         val cached = if (writePrivileges == null) relationCache.get(key) else None
         cached
-          // The per-query relation cache is not keyed by options. When the same table is referenced
-          // more than once in a single statement with different dynamic options (e.g. a self-join,
-          // or a second reference sharing the target's cache entry), a cache hit would otherwise
-          // reuse the first reference's options and silently drop this reference's. Re-apply this
-          // reference's options to the cached relation so each reference honors its own bag.
-          .map(applyOptions(_, finalOptions))
           .map(adaptCachedRelation(_, planId))
           .orElse {
             // For a `RelationCatalog` with no time-travel / write privileges, the single-RPC
@@ -280,7 +281,8 @@ class RelationResolution(
                     catalog,
                     ident,
                     finalTimeTravelSpec,
-                    Option(writePrivileges))
+                    Option(writePrivileges),
+                    finalOptions)
                 } else {
                   None
                 }
@@ -308,14 +310,17 @@ class RelationResolution(
             // we don't share-cache views.
             val table: Option[Table] = relation.collect { case t: Table => t }
 
+            // Reuse a cached relation only when this read's options match: the lookup is by name
+            // and `Table.id`, so a differing-options read would otherwise get the cached read's
+            // `Table`.
             val sharedRelationCacheMatch = for {
               t <- table
               if finalTimeTravelSpec.isEmpty && writePrivileges == null && !u.isStreaming
               cached <- lookupSharedRelationCache(catalog, ident, t)
+              if cached.options == finalOptions
             } yield {
-              val updatedRelation = cached.copy(options = finalOptions)
               val nameParts = ident.toQualifiedNameParts(catalog)
-              val aliasedRelation = SubqueryAlias(nameParts, updatedRelation)
+              val aliasedRelation = SubqueryAlias(nameParts, cached)
               relationCache.update(key, aliasedRelation)
               adaptCachedRelation(aliasedRelation, planId)
             }
@@ -370,19 +375,6 @@ class RelationResolution(
     CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident, table, conf)
   }
 
-  /**
-   * Re-applies `options` to the relation in a cached plan. Every `relationCache` entry holds a
-   * single relation for its own identifier (a view's body is still unresolved when it is cached),
-   * so this cannot reach another table's relation.
-   */
-  private def applyOptions(
-      cached: LogicalPlan,
-      options: CaseInsensitiveStringMap): LogicalPlan = cached transform {
-    case r: DataSourceV2Relation => r.copy(options = options)
-    case r: UnresolvedCatalogRelation => r.copy(options = options)
-    case r: StreamingRelationV2 => r.copy(extraOptions = options)
-  }
-
   private def adaptCachedRelation(cached: LogicalPlan, planId: Option[Long]): LogicalPlan = {
     val plan = cached transform {
       case multi: MultiInstanceRelation =>
@@ -496,7 +488,7 @@ class RelationResolution(
   }
 
   private def getOrLoadRelation(ref: V2TableReference): LogicalPlan = {
-    val key = toCacheKey(ref.catalog, ref.identifier)
+    val key = toCacheKey(ref.catalog, ref.identifier, None, ref.options)
     relationCache.get(key) match {
       case Some(cached) =>
         adaptCachedRelation(cached, ref)
@@ -556,8 +548,10 @@ class RelationResolution(
   private def toCacheKey(
       catalog: CatalogPlugin,
       ident: Identifier,
-      timeTravelSpec: Option[TimeTravelSpec] = None): CacheKey = {
-    ((catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq, timeTravelSpec)
+      timeTravelSpec: Option[TimeTravelSpec],
+      options: CaseInsensitiveStringMap): RelationCacheKey = {
+    RelationCacheKey(
+      (catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq, timeTravelSpec, options)
   }
 
   private def cloneWithPlanId(plan: LogicalPlan, planId: Option[Long]): LogicalPlan = {