You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ao...@apache.org on 2026/08/04 22:55:21 UTC
(spark) branch branch-4.x updated: [SPARK-58389][SQL] Pass all options while loading tables for reads
This is an automated email from the ASF dual-hosted git repository.
aokolnychyi pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5fbd37d9bf61 [SPARK-58389][SQL] Pass all options while loading tables for reads
5fbd37d9bf61 is described below
commit 5fbd37d9bf610f2c7aa23af2c7389b18042c9d08
Author: yyanyy <yy...@gmail.com>
AuthorDate: Tue Aug 4 15:53:59 2026 -0700
[SPARK-58389][SQL] Pass all options while loading tables for reads
### What changes were proposed in this pull request?
This is a follow-up to #56044 (which passed all options while loading changelogs). It does the same for table reads by adding `TableCatalog.loadTable(Identifier, TableContext, CaseInsensitiveStringMap)`, where `TableContext` carries the parsed, Spark-recognized parameters (time travel, write privileges) and the `CaseInsensitiveStringMap` carries all raw user options.
- New public connector types `TableContext` and `TimeTravel` (a clean sealed interface with `AsOfVersion`/`AsOfTimestamp` records), mirroring how #56044 introduced `ChangelogContext`/`ChangelogRange` rather than leaking the catalyst-internal `TimeTravelSpec`.
- The new `loadTable` overload has a default implementation that delegates to the existing `loadTable` overloads based on `TableContext`, so existing connectors keep working unchanged.
- `CatalogV2Util.getTable`/`loadTable` now build a `TableContext` from the catalyst `TimeTravelSpec` + write-privileges string and forward the user options, making the Java default the single dispatch site.
- Options are threaded through the read paths in `RelationResolution` and `DataSourceV2Utils`.
- The analyzer per-query relation cache is keyed on the user options (extracted into `RelationCacheKey`), so two references to the same table with different options each get their own `Table` instead of sharing the first one's. The shared relation cache (`CACHE TABLE`) is reused only when the options match.
- `V2TableRefreshUtil` now forwards options during refresh, keys refreshes by options, and only reuses cached relations with matching options.
- `CacheManager` preserves and forwards options when recaching V2 relations.
Potential follow-ups (intentionally out of scope here):
- The `RelationCatalog` single-RPC `loadRelation(Identifier)` path is not updated to forward options; closing it cleanly would add a `loadRelation(Identifier, CaseInsensitiveStringMap)` overload.
- The `V2TableReference` re-resolution path (`getOrLoadRelation`/`loadRelation`, used by DataFrame temp views, transaction re-resolution, and write targets) still loads via the single-arg `loadTable`. It is transaction-aware (the transaction catalog intercepts `loadTable` to track reads), so forwarding options there is better done as a separate change.
- Writer entry points such as `DataFrameWriterV2`, `DataFrameWriter.insertInto` and `DataStreamWriter.toTable` are not updated to forward all options.
### Why are the changes needed?
To make the API usable in connectors like Iceberg and Delta, which need the user options while reading a table. The functionality hasn't been released yet.
### Does this PR introduce _any_ user-facing change?
Yes, minor: `INSERT INTO t WITH ('versionAsOf' = ...)` (time travel on a write target, reachable via the option form) now raises a user-facing `UNSUPPORTED_FEATURE.TIME_TRAVEL` analysis error instead of an internal `AssertionError`. Connectors that override the new `loadTable` overload also see one `loadTable` call per distinct option bag when the same table is referenced multiple times. Reads with dynamic options no longer reuse a `CACHE TABLE` snapshot when the options differ; they [...]
### How was this patch tested?
New tests in `CatalogV2UtilSuite` (the default-dispatch mapping for each `TableContext` shape, and the time-travel/write-privileges mutual-exclusion invariant), `DataSourceV2OptionSuite` (end-to-end option forwarding via the DataFrame API, SQL, streaming, time travel; per-option-bag loads on a self-join; a temp view's options not leaking to a later reference; `CACHE TABLE` results not reused for a differing-options read), `PlanResolutionSuite` (shared relation cache reused only when o [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude code Opus 4.8
Closes #57582 from yyanyy/spark-dsv2-loadtable-options-20260727.
Authored-by: yyanyy <yy...@gmail.com>
Signed-off-by: Anton Okolnychyi <ao...@apache.org>
(cherry picked from commit c6a2bf819d0c42828ce07a2fe86aefa60000e4a5)
Signed-off-by: Anton Okolnychyi <ao...@apache.org>
---
.../spark/sql/connector/catalog/TableCatalog.java | 45 +++
.../spark/sql/connector/catalog/TableContext.java | 85 +++++
.../spark/sql/connector/catalog/TimeTravel.java | 43 +++
.../spark/sql/catalyst/analysis/Analyzer.scala | 9 +-
.../sql/catalyst/analysis/RelationCacheKey.scala | 32 ++
.../sql/catalyst/analysis/RelationResolution.scala | 53 ++-
.../sql/connector/catalog/CatalogV2Util.scala | 45 +--
.../sql/connector/catalog/CatalogV2UtilSuite.scala | 82 ++++-
.../connector/catalog/InMemoryTableCatalog.scala | 24 ++
.../apache/spark/sql/execution/CacheManager.scala | 6 +-
.../datasources/v2/DataSourceV2Utils.scala | 2 +-
.../datasources/v2/V2TableRefreshUtil.scala | 13 +-
.../sql/connector/DataSourceV2OptionSuite.scala | 361 ++++++++++++++++++++-
.../connector/SupportsCatalogOptionsSuite.scala | 15 +
.../command/AlignAssignmentsSuiteBase.scala | 7 +-
.../execution/command/PlanResolutionSuite.scala | 63 +++-
16 files changed, 813 insertions(+), 72 deletions(-)
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
index 23e9499932c1..58a40c237332 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
+++ b/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 @@ public interface TableCatalog extends CatalogPlugin {
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}.
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java
new file mode 100644
index 000000000000..9437149895d2
--- /dev/null
+++ b/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 + "}";
+ }
+}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TimeTravel.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TimeTravel.java
new file mode 100644
index 000000000000..c9ecc80d65d5
--- /dev/null
+++ b/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 {}
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index a8d687c9ad02..453392616562 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/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)`.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCacheKey.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCacheKey.scala
new file mode 100644
index 000000000000..c89a9b0b9581
--- /dev/null
+++ b/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)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
index ef6648486310..d217219cd8f9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
@@ -17,8 +17,6 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable
-
import org.apache.spark.internal.Logging
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.SQLConfHelper
@@ -61,11 +59,9 @@ 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 = AnalysisContext.get.relationCache
/**
* If we are resolving database objects (relations, functions, etc.) inside views, we may need to
@@ -236,24 +232,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 +278,8 @@ class RelationResolution(
catalog,
ident,
finalTimeTravelSpec,
- Option(writePrivileges))
+ Option(writePrivileges),
+ finalOptions)
} else {
None
}
@@ -308,14 +307,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 +372,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 +485,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 +545,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 = {
+ val nameParts = (catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq
+ RelationCacheKey(nameParts, timeTravelSpec, options)
}
private def cloneWithPlanId(plan: LogicalPlan, planId: Option[Long]): LogicalPlan = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
index c0905ede4f0d..28aaae4a81ff 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
@@ -472,9 +472,10 @@ private[sql] object CatalogV2Util {
catalog: CatalogPlugin,
ident: Identifier,
timeTravelSpec: Option[TimeTravelSpec] = None,
- writePrivilegesString: Option[String] = None): Option[Table] =
+ writePrivilegesString: Option[String] = None,
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Option[Table] =
try {
- Option(getTable(catalog, ident, timeTravelSpec, writePrivilegesString))
+ Option(getTable(catalog, ident, timeTravelSpec, writePrivilegesString, options))
} catch {
case _: NoSuchTableException => None
case _: NoSuchDatabaseException => None
@@ -484,23 +485,29 @@ private[sql] object CatalogV2Util {
catalog: CatalogPlugin,
ident: Identifier,
timeTravelSpec: Option[TimeTravelSpec] = None,
- writePrivilegesString: Option[String] = None): Table = {
- if (timeTravelSpec.nonEmpty) {
- assert(writePrivilegesString.isEmpty, "Should not write to a table with time travel")
- timeTravelSpec.get match {
- case v: AsOfVersion =>
- catalog.asTableCatalog.loadTable(ident, v.version)
- case ts: AsOfTimestamp =>
- catalog.asTableCatalog.loadTable(ident, ts.timestamp)
- }
- } else {
- if (writePrivilegesString.isDefined) {
- val writePrivileges = writePrivilegesString.get.split(",").map(_.trim)
- .map(TableWritePrivilege.valueOf).toSet.asJava
- catalog.asTableCatalog.loadTable(ident, writePrivileges)
- } else {
- catalog.asTableCatalog.loadTable(ident)
- }
+ writePrivilegesString: Option[String] = None,
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Table = {
+ val timeTravel: TimeTravel = timeTravelSpec match {
+ case Some(v: AsOfVersion) => new TimeTravel.AsOfVersion(v.version)
+ case Some(ts: AsOfTimestamp) => new TimeTravel.AsOfTimestamp(ts.timestamp)
+ case None => null
+ }
+ val context = new TableContext(timeTravel, parseWritePrivileges(writePrivilegesString))
+ catalog.asTableCatalog.loadTable(ident, context, options)
+ }
+
+ /**
+ * Parses the comma-separated write-privileges string (as carried in the internal
+ * [[org.apache.spark.sql.catalyst.analysis.UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES]]
+ * option) into a set of [[TableWritePrivilege]]. Returns an empty set when absent (a read).
+ */
+ private def parseWritePrivileges(
+ writePrivilegesString: Option[String]): util.Set[TableWritePrivilege] = {
+ writePrivilegesString match {
+ case Some(str) =>
+ str.split(",").map(_.trim).map(TableWritePrivilege.valueOf).toSet.asJava
+ case None =>
+ util.Set.of()
}
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala
index eda401ceb6bd..4b9d55be07e3 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala
@@ -17,15 +17,30 @@
package org.apache.spark.sql.connector.catalog
-import org.mockito.Mockito.{mock, when}
+import org.mockito.ArgumentMatchers.{any, eq => mockEq}
+import org.mockito.Mockito.{mock, verify, when}
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkFunSuite, SparkIllegalArgumentException}
+import org.apache.spark.sql.catalyst.analysis.{AsOfTimestamp, AsOfVersion, TimeTravelSpec}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
class CatalogV2UtilSuite extends SparkFunSuite {
- test("Load relation should encode the identifiers for V2Relations") {
+
+ // CatalogV2Util.getTable routes through the options-aware TableCatalog.loadTable, whose default
+ // implementation dispatches to the existing overloads. Stub only that method to run the real
+ // default so the dispatch is exercised; the leaf overloads stay as plain mock methods (returning
+ // null) that we then `verify`.
+ private def mockCatalogWithRealDispatch(): TableCatalog = {
val testCatalog = mock(classOf[TableCatalog])
+ when(testCatalog.loadTable(
+ any[Identifier], any[TableContext], any[CaseInsensitiveStringMap])).thenCallRealMethod()
+ testCatalog
+ }
+
+ test("Load relation should encode the identifiers for V2Relations") {
+ val testCatalog = mockCatalogWithRealDispatch()
val ident = mock(classOf[Identifier])
val table = mock(classOf[Table])
when(table.columns()).thenReturn(Array(Column.create("i", IntegerType)))
@@ -37,4 +52,65 @@ class CatalogV2UtilSuite extends SparkFunSuite {
assert(v2Relation.catalog.exists(_ == testCatalog))
assert(v2Relation.identifier.exists(_ == ident))
}
+
+ private def getTableAndVerifyDispatch(
+ timeTravelSpec: Option[TimeTravelSpec],
+ writePrivilegesString: Option[String])(
+ verifyOverload: TableCatalog => Unit): Unit = {
+ val testCatalog = mockCatalogWithRealDispatch()
+ val ident = mock(classOf[Identifier])
+ CatalogV2Util.getTable(testCatalog, ident, timeTravelSpec, writePrivilegesString)
+ verifyOverload(testCatalog)
+ }
+
+ test("getTable dispatches to loadTable(ident) with no time travel and no write privileges") {
+ getTableAndVerifyDispatch(None, None) { c => verify(c).loadTable(any[Identifier]) }
+ }
+
+ test("getTable dispatches to loadTable(ident, writePrivileges) with write privileges") {
+ getTableAndVerifyDispatch(None, Some("INSERT,DELETE")) { c =>
+ verify(c).loadTable(
+ any[Identifier],
+ mockEq(java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)))
+ }
+ }
+
+ test("getTable dispatches to loadTable(ident, version) for version time travel") {
+ getTableAndVerifyDispatch(Some(AsOfVersion("v1")), None) { c =>
+ verify(c).loadTable(any[Identifier], mockEq("v1"))
+ }
+ }
+
+ test("getTable dispatches to loadTable(ident, timestamp) for timestamp time travel") {
+ getTableAndVerifyDispatch(Some(AsOfTimestamp(123L)), None) { c =>
+ verify(c).loadTable(any[Identifier], mockEq(123L))
+ }
+ }
+
+ test("getTable rejects combining time travel and write privileges") {
+ val testCatalog = mockCatalogWithRealDispatch()
+ val ident = mock(classOf[Identifier])
+ val e = intercept[SparkIllegalArgumentException] {
+ CatalogV2Util.getTable(testCatalog, ident, Some(AsOfVersion("v1")), Some("INSERT"))
+ }
+ assert(e.getMessage.contains("Cannot set both time travel and write privileges"))
+ }
+
+ test("TableContext normalizes null time travel and null write privileges to empty") {
+ val context = new TableContext(null, null)
+ assert(context.timeTravel().isEmpty)
+ assert(context.writePrivileges().isEmpty)
+ }
+
+ test("TableContext equals / hashCode / toString") {
+ val emptyPrivileges = java.util.Set.of[TableWritePrivilege]()
+ val a = new TableContext(new TimeTravel.AsOfVersion("v1"), emptyPrivileges)
+ val b = new TableContext(new TimeTravel.AsOfVersion("v1"), emptyPrivileges)
+ val c = new TableContext(new TimeTravel.AsOfTimestamp(1L), emptyPrivileges)
+ assert(a == b)
+ assert(a.hashCode() == b.hashCode())
+ assert(a != c)
+ assert(a.toString.contains("timeTravel"))
+ assert(a.toString.contains("writePrivileges"))
+ }
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
index bb137ba4830d..53a7fa05b40e 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
@@ -22,6 +22,7 @@ import java.util.Collections
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger
+import scala.collection.mutable
import scala.jdk.CollectionConverters._
import org.apache.spark.sql.catalyst.InternalRow
@@ -48,6 +49,19 @@ class BasicInMemoryTableCatalog extends TableCatalog {
private var _name: Option[String] = None
private var copyOnLoad: Boolean = false
+ // Records every (TableContext, options) pair passed to the options-aware loadTable(), in call
+ // order, so tests can verify that the analyzer / DataFrame API correctly constructed and
+ // forwarded them -- including how many times loadTable was called when the same table is
+ // referenced more than once in a statement with different options.
+ // "loadTable" is in the name because the subclass InMemoryChangelogCatalog has an analogous
+ // `lastOptions` recording the options passed to loadChangelog(); the two must not collide.
+ private val _loadTableCalls = mutable.ArrayBuffer.empty[(TableContext, CaseInsensitiveStringMap)]
+ def loadTableCalls: Seq[(TableContext, CaseInsensitiveStringMap)] = _loadTableCalls.toSeq
+ def resetLoadTableCalls(): Unit = _loadTableCalls.clear()
+
+ def lastTableContext: Option[TableContext] = _loadTableCalls.lastOption.map(_._1)
+ def lastLoadTableOptions: Option[CaseInsensitiveStringMap] = _loadTableCalls.lastOption.map(_._2)
+
override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {
_name = Some(name)
copyOnLoad = options.getBoolean("copyOnLoad", false)
@@ -124,6 +138,16 @@ class BasicInMemoryTableCatalog extends TableCatalog {
}
}
+ // Records the forwarded context/options so tests can verify they reached the catalog, then
+ // defers to the default dispatch in TableCatalog (rather than reimplementing it here).
+ override def loadTable(
+ ident: Identifier,
+ context: TableContext,
+ options: CaseInsensitiveStringMap): Table = {
+ _loadTableCalls += ((context, options))
+ super.loadTable(ident, context, options)
+ }
+
override def invalidateTable(ident: Identifier): Unit = {
invalidatedTables.add(ident)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
index 345c1d5d635f..3541c939909f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan, Resolv
import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
import org.apache.spark.sql.catalyst.util.sideBySide
import org.apache.spark.sql.classic.{Dataset, SparkSession}
-import org.apache.spark.sql.connector.catalog.CatalogPlugin
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, CatalogV2Util}
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.{IdentifierHelper, MultipartIdentifierHelper}
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.transactions.Transaction
@@ -419,9 +419,9 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
try {
EliminateSubqueryAliases(plan) match {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if r.timeTravelSpec.isEmpty =>
- val table = catalog.loadTable(ident)
+ val table = CatalogV2Util.getTable(catalog, ident, options = r.options)
if (r.table.id == table.id) {
- Some(DataSourceV2Relation.create(table, Some(catalog), Some(ident)))
+ Some(DataSourceV2Relation.create(table, Some(catalog), Some(ident), r.options))
} else {
None
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala
index a3b5c5aeb799..6f7447faa621 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala
@@ -141,7 +141,7 @@ private[sql] object DataSourceV2Utils extends Logging {
}
val timeTravel = TimeTravelSpec.create(
timeTravelTimestamp, timeTravelVersion, conf.sessionLocalTimeZone)
- val tbl = CatalogV2Util.getTable(catalog, ident, timeTravel)
+ val tbl = CatalogV2Util.getTable(catalog, ident, timeTravel, options = dsOptions)
(tbl, Some(catalog), Some(ident), timeTravel)
case _ =>
// TODO: Non-catalog paths for DSV2 are currently not well defined.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
index f1ff11b1a4a6..10408aa51363 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
@@ -26,11 +26,14 @@ import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog, V2TableUtil}
import org.apache.spark.sql.connector.catalog.CatalogV2Util
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.sql.util.SchemaValidationMode
import org.apache.spark.sql.util.SchemaValidationMode.ALLOW_NEW_FIELDS
import org.apache.spark.sql.util.SchemaValidationMode.PROHIBIT_CHANGES
private[sql] object V2TableRefreshUtil extends SQLConfHelper with Logging {
+ private type CurrentTableKey = (TableCatalog, Identifier, CaseInsensitiveStringMap)
+
/**
* Refreshes table metadata for tables in the plan.
*
@@ -81,19 +84,19 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper with Logging {
plan: LogicalPlan,
versionedOnly: Boolean,
schemaValidationMode: SchemaValidationMode): LogicalPlan = {
- val currentTables = mutable.HashMap.empty[(TableCatalog, Identifier), Table]
+ val currentTables = mutable.HashMap.empty[CurrentTableKey, Table]
plan transformWithSubqueries {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
if (r.isVersioned || !versionedOnly) && r.timeTravelSpec.isEmpty =>
- val currentTable = currentTables.getOrElseUpdate((catalog, ident), {
+ val currentTable = currentTables.getOrElseUpdate((catalog, ident, r.options), {
val tableName = V2TableUtil.toQualifiedName(catalog, ident)
lookupCachedRelation(spark, catalog, ident, r.table) match {
- case Some(cached) =>
+ case Some(cached) if cached.options == r.options =>
logDebug(s"Refreshing table metadata for $tableName using shared relation cache")
cached.table
- case None =>
+ case _ =>
logDebug(s"Refreshing table metadata for $tableName using catalog")
- catalog.loadTable(ident)
+ CatalogV2Util.getTable(catalog, ident, options = r.options)
}
})
validateTableIdentity(currentTable, r)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
index f785cac9c012..8cdf12b18f44 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
@@ -17,20 +17,35 @@
package org.apache.spark.sql.connector
+import java.util.concurrent.atomic.AtomicInteger
+
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
-import org.apache.spark.sql.connector.catalog.{InMemoryBaseTable, InMemoryRowLevelOperationTableCatalog}
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryBaseTable, InMemoryCatalog, InMemoryRowLevelOperationTableCatalog, Table, TimeTravel}
import org.apache.spark.sql.execution.CommandResultExec
import org.apache.spark.sql.execution.datasources.v2._
import org.apache.spark.sql.functions.lit
+class LoadCountingInMemoryCatalog extends InMemoryCatalog {
+ val singleArgLoads = new AtomicInteger(0)
+
+ override def loadTable(ident: Identifier): Table = {
+ singleArgLoads.incrementAndGet()
+ super.loadTable(ident)
+ }
+}
+
class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
import testImplicits._
private val catalogAndNamespace = "testcat.ns1.ns2."
+ private def inMemoryCatalog: InMemoryCatalog =
+ catalog("testcat").asInstanceOf[InMemoryCatalog]
+
test("SPARK-36680: Supports Dynamic Table Options for SQL Select") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
@@ -426,4 +441,348 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
assert (collected.size == 1)
}
}
+
+ test("options are forwarded to loadTable - DataFrame API") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ spark.read.option("customOption", "customValue").table(t1).collect()
+
+ val opts = inMemoryCatalog.lastLoadTableOptions
+ assert(opts.isDefined)
+ assert(opts.get.get("customOption") === "customValue")
+ }
+ }
+
+ test("options are forwarded to loadTable - SQL") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"SELECT * FROM $t1 WITH ('customOption' = 'customValue')").collect()
+
+ val opts = inMemoryCatalog.lastLoadTableOptions
+ assert(opts.isDefined)
+ assert(opts.get.get("customOption") === "customValue")
+ }
+ }
+
+ test("options are forwarded to loadTable - DataStreamReader") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ // Trigger analysis of the streaming relation.
+ spark.readStream.option("customOption", "customValue").table(t1).queryExecution.analyzed
+
+ val opts = inMemoryCatalog.lastLoadTableOptions
+ assert(opts.isDefined)
+ assert(opts.get.get("customOption") === "customValue")
+ }
+ }
+
+ test("options are forwarded to loadTable alongside time travel") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ // versionAsOf is the default time-travel version option key
+ // (SQLConf.TIME_TRAVEL_VERSION_KEY). Pin a versioned copy so the versioned load succeeds.
+ inMemoryCatalog.pinTable(Identifier.of(Array("ns1", "ns2"), "table"), "v1")
+
+ spark.read
+ .option("versionAsOf", "v1")
+ .option("customOption", "customValue")
+ .table(t1)
+ .collect()
+
+ val ctx = inMemoryCatalog.lastTableContext
+ assert(ctx.isDefined)
+ assert(ctx.get.timeTravel().isPresent)
+ assert(ctx.get.timeTravel().get() === new TimeTravel.AsOfVersion("v1"))
+
+ val opts = inMemoryCatalog.lastLoadTableOptions
+ assert(opts.isDefined)
+ assert(opts.get.get("customOption") === "customValue")
+ assert(opts.get.get("versionAsOf") === "v1")
+ }
+ }
+
+ test("write privileges are carried in TableContext, internal key stripped") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a')")
+
+ val ctx = inMemoryCatalog.lastTableContext
+ assert(ctx.isDefined)
+ assert(!ctx.get.writePrivileges().isEmpty)
+
+ val opts = inMemoryCatalog.lastLoadTableOptions
+ assert(opts.isDefined)
+ // The internal write-privileges marker must not leak to the connector as a user option.
+ assert(opts.get.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES) === null)
+ }
+ }
+
+ test("SPARK-58389: execution refresh forwards options on a plain table read") {
+ registerCatalog("loadcounting", classOf[LoadCountingInMemoryCatalog])
+ val loadCountingCatalog =
+ catalog("loadcounting").asInstanceOf[LoadCountingInMemoryCatalog]
+ val t1 = "loadcounting.ns1.ns2.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ loadCountingCatalog.resetLoadTableCalls()
+ loadCountingCatalog.singleArgLoads.set(0)
+
+ spark.read.option("split-size", "5").table(t1).collect()
+
+ // Both analysis and the execution-time refresh must enter through the options-aware
+ // overload. Each then delegates to the single-argument overload in this test catalog.
+ val optionAwareLoads = loadCountingCatalog.loadTableCalls
+ .map(_._2.get("split-size"))
+ .filter(_ != null)
+ assert(optionAwareLoads === Seq("5", "5"),
+ s"expected analysis and refresh to forward split-size=5, got: $optionAwareLoads")
+ assert(loadCountingCatalog.singleArgLoads.get() === 2,
+ s"expected two delegated single-argument loads, got: " +
+ loadCountingCatalog.singleArgLoads.get())
+ }
+ }
+
+ test("SPARK-58389: a self-join with different options loads the table once per option bag") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
+
+ // The two references share a name but carry different options. Because a catalog's
+ // options-aware loadTable can return a different Table depending on the options, the analyzer
+ // relation cache is keyed on the options, so each reference triggers its own loadTable rather
+ // than reusing the first reference's Table.
+ val df = sql(s"SELECT a.id FROM $t1 WITH (`split-size` = 5) a " +
+ s"JOIN $t1 WITH (`split-size` = 9) b ON a.id = b.id")
+ df.queryExecution.analyzed
+
+ // Each distinct option bag reaches the catalog as its own loadTable call during analysis.
+ val loadedOptions = inMemoryCatalog.loadTableCalls
+ .map(_._2.get("split-size"))
+ .filter(_ != null)
+ .sorted
+ assert(loadedOptions === Seq("5", "9"),
+ s"expected one loadTable per distinct option bag, got: $loadedOptions")
+
+ // The execution-time refresh also reloads once per option bag instead of sharing one Table
+ // across the two references.
+ inMemoryCatalog.resetLoadTableCalls()
+ df.collect()
+ val refreshedOptions = inMemoryCatalog.loadTableCalls
+ .map(_._2.get("split-size"))
+ .filter(_ != null)
+ .sorted
+ assert(refreshedOptions === Seq("5", "9"),
+ s"expected refresh to load each distinct option bag, got: $refreshedOptions")
+
+ // Each scan also keeps its own option end-to-end (neither reference inherits the other's).
+ val splitSizes = df.queryExecution.optimizedPlan.collect {
+ case s: DataSourceV2ScanRelation => s.relation.options.get("split-size")
+ }.sorted
+ assert(splitSizes === Seq("5", "9"))
+ }
+ }
+
+ test("SPARK-58389: repeated references with the same options load the table once") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+ inMemoryCatalog.resetLoadTableCalls()
+
+ // Both references carry the same options, so they share one relation-cache entry: the table
+ // is loaded once (resolve-once-per-query is preserved for identical option bags).
+ val df = sql(s"SELECT a.id FROM $t1 WITH (`split-size` = 5) a " +
+ s"JOIN $t1 WITH (`split-size` = 5) b ON a.id = b.id")
+ df.queryExecution.analyzed
+
+ val splitSizeLoads = inMemoryCatalog.loadTableCalls
+ .count(_._2.get("split-size") === "5")
+ assert(splitSizeLoads === 1,
+ s"expected a single loadTable for identical option bags, got: $splitSizeLoads")
+
+ // The refresh phase also reuses one load for repeated references with identical options.
+ inMemoryCatalog.resetLoadTableCalls()
+ df.collect()
+ val refreshedSplitSizeLoads = inMemoryCatalog.loadTableCalls
+ .count(_._2.get("split-size") === "5")
+ assert(refreshedSplitSizeLoads === 1,
+ s"expected refresh to load identical option bags once, got: $refreshedSplitSizeLoads")
+ }
+ }
+
+ test("SPARK-58389: time travel option on a write target is rejected with a user-facing error") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+
+ // A time-travel option on a write target is reachable via the option form (the `AS OF`
+ // syntax is blocked earlier by the parser). It must surface as a user-facing analysis error,
+ // not the internal TableContext mutual-exclusion guard (which would report INTERNAL_ERROR).
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"INSERT INTO $t1 WITH ('versionAsOf' = 'v1') VALUES (1, 'a')")
+ },
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> "`testcat`.`ns1`.`ns2`.`table`"))
+ }
+ }
+
+ test("SPARK-58389: CACHE TABLE result is not reused for a read carrying different options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ // Cache the option-free read. The CacheManager keys entries on the query plan, and a DSv2
+ // relation's plan carries its `options`, so the cached entry's fingerprint is "no options".
+ val cached = spark.table(t1)
+ cached.cache()
+ try {
+ val cacheManager = spark.sharedState.cacheManager
+
+ // An option-free read has the same plan fingerprint, so it reuses the cached result.
+ assert(cacheManager.lookupCachedData(spark.table(t1)).isDefined,
+ "an option-free read should hit the cached result")
+
+ // A read carrying options has a different fingerprint, so it must NOT reuse the cached
+ // result -- otherwise the connector's options would be silently ignored on a cache hit.
+ assert(
+ cacheManager.lookupCachedData(spark.read.option("split-size", "5").table(t1)).isEmpty,
+ "a read carrying options must not reuse the option-free cached result")
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+
+ test("SPARK-58389: execution refresh does not reuse cached relation with different options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ val cached = spark.table(t1)
+ cached.cache()
+ try {
+ assert(cached.count() === 2)
+
+ // Analysis correctly rejects the option-free shared relation cache entry. Reset after
+ // analysis to isolate the execution-time refresh, which must apply the same options check.
+ val df = spark.read.option("split-size", "5").table(t1).filter("id > 0")
+ df.queryExecution.analyzed
+ inMemoryCatalog.resetLoadTableCalls()
+ df.collect()
+
+ val refreshLoads = inMemoryCatalog.loadTableCalls
+ .map(_._2.get("split-size"))
+ .filter(_ != null)
+ assert(refreshLoads === Seq("5"),
+ s"expected refresh to reject the option-free cache entry, got: $refreshLoads")
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+
+ test("SPARK-58389: recaching preserves and forwards table options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ val cached = spark.read.option("split-size", "5").table(t1)
+ cached.cache()
+ try {
+ assert(cached.count() === 2)
+
+ // Refreshing a cached table rebuilds its CacheManager entry. The rebuilt relation must
+ // retain the original options, and the catalog must use them when it reloads the Table.
+ inMemoryCatalog.resetLoadTableCalls()
+ spark.catalog.refreshTable(t1)
+
+ val recacheLoads = inMemoryCatalog.loadTableCalls.map(_._2.get("split-size"))
+ assert(recacheLoads.contains("5"),
+ s"expected recache to forward split-size=5, got: $recacheLoads")
+
+ val cacheManager = spark.sharedState.cacheManager
+ val sameOptions = spark.read.option("split-size", "5").table(t1)
+ val recached = cacheManager.lookupCachedData(sameOptions)
+ assert(recached.isDefined, "a read with the original options should hit after recache")
+
+ val recachedOptions = recached.get.plan.collect {
+ case r: DataSourceV2Relation => r.options.get("split-size")
+ }
+ assert(recachedOptions === Seq("5"),
+ s"expected the recached relation to retain split-size=5, got: $recachedOptions")
+
+ assert(cacheManager.lookupCachedData(spark.table(t1)).isEmpty,
+ "an option-free read must not reuse the recached option-carrying result")
+ } finally {
+ spark.catalog.clearCache()
+ }
+ }
+ }
+
+ test("SPARK-58389: recaching a non-relation plan forwards table options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ // The filter forces CacheManager.tryRefreshPlan through V2TableRefreshUtil instead of the
+ // bare-relation fast path covered by the preceding test.
+ val cached = spark.read.option("split-size", "5").table(t1).filter("id > 0")
+ cached.cache()
+ try {
+ assert(cached.count() === 2)
+ inMemoryCatalog.resetLoadTableCalls()
+
+ spark.catalog.refreshTable(t1)
+
+ val recacheLoads = inMemoryCatalog.loadTableCalls
+ .map(_._2.get("split-size"))
+ .filter(_ != null)
+ assert(recacheLoads.contains("5"),
+ s"expected non-relation recache to forward split-size=5, got: $recacheLoads")
+
+ val samePlan = spark.read.option("split-size", "5").table(t1).filter("id > 0")
+ assert(spark.sharedState.cacheManager.lookupCachedData(samePlan).isDefined,
+ "the filtered plan should remain cached after refresh")
+ } finally {
+ spark.catalog.clearCache()
+ }
+ }
+ }
+
+ test("SPARK-58389: a DataFrame temp view's options do not leak to a later reference") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ // The temp view resolves via the V2TableReference path, whose relation carries the view's
+ // options. A later option-free reference to the same table must not inherit them.
+ withTempView("v") {
+ spark.read.option("split-size", "5").table(t1).createOrReplaceTempView("v")
+ val df = sql(s"SELECT v.id FROM v JOIN $t1 b ON v.id = b.id")
+
+ val splitSizes = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation => Option(r.options.get("split-size"))
+ }
+ // Exactly one reference (`v`) keeps its option; `b` (option-free) must not inherit it.
+ assert(splitSizes.flatten === Seq("5"),
+ s"option leaked to the option-free reference, got: $splitSizes")
+ assert(splitSizes.contains(None),
+ s"expected an option-free reference, got: $splitSizes")
+ }
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala
index bcd8ba185d1d..406e8885e163 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala
@@ -385,6 +385,21 @@ class SupportsCatalogOptionsSuite extends SharedSparkSession with BeforeAndAfter
assert(relation.timeTravelSpec.contains(expectedTimeTravelSpec))
}
+ test("SPARK-58389: read options are forwarded to the catalog's loadTable") {
+ sql(s"create table $catalogName.t1 (id bigint) using $format")
+ val cat = catalog(catalogName).asInstanceOf[InMemoryTableCatalog]
+ cat.resetLoadTableCalls()
+
+ // spark.read.format(...).option(...).load() goes through the SupportsCatalogOptions read
+ // path in DataSourceV2Utils, which now forwards the user options to CatalogV2Util.getTable.
+ load("t1", Some(catalogName)).collect()
+
+ val opts = cat.lastLoadTableOptions
+ assert(opts.isDefined, "loadTable(context, options) was not invoked")
+ // The user-provided read options reach the catalog (e.g. the table name selector).
+ assert(opts.get.get("name") === "t1")
+ }
+
private def load(
name: String,
catalogOpt: Option[String],
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala
index c88ebb0d69ee..bcc4895616bd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala
@@ -30,12 +30,13 @@ import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog, TableWritePrivilege}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog, TableContext, TableWritePrivilege}
import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{BooleanType, IntegerType, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
abstract class AlignAssignmentsSuiteBase extends AnalysisTest {
@@ -163,6 +164,10 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest {
})
when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]()))
.thenCallRealMethod()
+ // The options-aware overload runs the real default dispatch, which delegates to the
+ // stubbed overloads above.
+ when(newCatalog.loadTable(any(), any[TableContext](), any[CaseInsensitiveStringMap]()))
+ .thenCallRealMethod()
when(newCatalog.name()).thenReturn("cat")
newCatalog
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
index a924b637a79f..ee3eaa789ffc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
@@ -27,7 +27,7 @@ import org.mockito.invocation.InvocationOnMock
import org.apache.spark.SparkUnsupportedOperationException
import org.apache.spark.sql.{AnalysisException, SaveMode}
import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, AnalysisTest, Analyzer, AsOfVersion, EmptyFunctionRegistry, NoSuchTableException, RelationResolution, ResolvedFieldName, ResolvedFieldPosition, ResolvedIdentifier, ResolvedTable, ResolveSessionCatalog, TimeTravelSpec, UnresolvedAttribute, UnresolvedFieldPosition, UnresolvedInlineTable, UnresolvedPartitionSpec, UnresolvedRelation, UnresolvedSubqueryColumnAliases, UnresolvedTable}
+import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, AnalysisTest, Analyzer, AsOfVersion, EmptyFunctionRegistry, NoSuchTableException, RelationCache, RelationResolution, ResolvedFieldName, ResolvedFieldPosition, ResolvedIdentifier, ResolvedTable, ResolveSessionCatalog, TimeTravelSpec, UnresolvedAttribute, UnresolvedFieldPosition, UnresolvedInlineTable, UnresolvedPartitionSpec, UnresolvedRelation, UnresolvedSubqueryColumnAliases, UnresolvedTable}
import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogStorageFormat, CatalogTable, CatalogTableType, InMemoryCatalog, SessionCatalog, TempVariableManager}
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, EqualTo, Expression, InSubquery, IntegerLiteral, ListQuery, Literal, StringLiteral}
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
@@ -36,7 +36,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{AlterColumns, AlterColumnSpe
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
import org.apache.spark.sql.connector.FakeV2Provider
-import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, TableChange, TableWritePrivilege, V1Table}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, TableChange, TableContext, TableWritePrivilege, V1Table}
import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform}
import org.apache.spark.sql.errors.QueryExecutionErrors
@@ -47,6 +47,7 @@ import org.apache.spark.sql.internal.SQLConf.{PARTITION_OVERWRITE_MODE, Partitio
import org.apache.spark.sql.sources.SimpleScanSource
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{BooleanType, CharType, DoubleType, IntegerType, LongType, StringType, StructField, StructType, VarcharType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.UTF8String
class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
@@ -190,6 +191,10 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
})
when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]()))
.thenCallRealMethod()
+ // The options-aware overload runs the real default dispatch, which delegates to the
+ // stubbed overloads above.
+ when(newCatalog.loadTable(any(), any[TableContext](), any[CaseInsensitiveStringMap]()))
+ .thenCallRealMethod()
when(newCatalog.name()).thenReturn("testcat")
newCatalog
}
@@ -209,6 +214,10 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
})
when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]()))
.thenCallRealMethod()
+ // The options-aware overload runs the real default dispatch, which delegates to the
+ // stubbed overloads above.
+ when(newCatalog.loadTable(any(), any[TableContext](), any[CaseInsensitiveStringMap]()))
+ .thenCallRealMethod()
when(newCatalog.name()).thenReturn(CatalogManager.SESSION_CATALOG_NAME)
newCatalog
}
@@ -3398,7 +3407,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
// after first resolution, cache should have 1 entry (without time travel)
assert(ctx.relationCache.size == 1)
- assert(ctx.relationCache.keys.head._2.isEmpty)
+ assert(ctx.relationCache.keys.head.timeTravelSpec.isEmpty)
// create unresolved relation with time travel spec
val timeTravelSpec = AsOfVersion("v1")
@@ -3456,6 +3465,54 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
}
}
+ test("SPARK-58389: shared relation cache is reused only when the read's options match") {
+ val ident = Identifier.of(Array.empty[String], "tab")
+ val cachedTable = testCat.loadTable(ident)
+
+ // A shared relation cache entry (as if left by an earlier CACHE TABLE) built with a specific
+ // set of options. The `id` tag lets the test tell a cache reuse apart from a fresh load, since
+ // both would otherwise carry the same mock `Table`.
+ def cachedRelationWith(opts: java.util.Map[String, String]): DataSourceV2Relation = {
+ val r = DataSourceV2Relation.create(
+ cachedTable, Some(testCat), Some(ident), new CaseInsensitiveStringMap(opts))
+ r.setTagValue(LogicalPlan.PLAN_ID_TAG, 4242L)
+ r
+ }
+
+ def resolveWith(
+ cacheOpts: java.util.Map[String, String],
+ readOpts: java.util.Map[String, String]): DataSourceV2Relation = {
+ AnalysisContext.withNewAnalysisContext {
+ val sharedCache: RelationCache = (_, _) => Some(cachedRelationWith(cacheOpts))
+ val rule = new RelationResolution(catalogManagerWithDefault, sharedCache)
+ val unresolved =
+ UnresolvedRelation(Seq("testcat", "tab"), new CaseInsensitiveStringMap(readOpts))
+ rule.resolveRelation(unresolved) match {
+ case Some(AsDataSourceV2Relation(relation)) => relation
+ case other => fail(s"failed to resolve as v2 relation: $other")
+ }
+ }
+ }
+
+ // Same options as the cached entry: the cache is reused (the tagged cached relation flows
+ // through, so the tag survives).
+ val reused = resolveWith(
+ java.util.Map.of("split-size", "5"), java.util.Map.of("split-size", "5"))
+ assert(reused.options.get("split-size") === "5")
+ assert(reused.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(4242L),
+ "matching options should reuse the cached relation")
+
+ // Different options: the cache is NOT reused. The relation is freshly loaded with this read's
+ // options, so it does not carry the cached entry's option or its tag.
+ val fresh = resolveWith(
+ java.util.Map.of("cachedOnly", "stale"), java.util.Map.of("split-size", "5"))
+ assert(fresh.options.get("split-size") === "5")
+ assert(!fresh.options.containsKey("cachedOnly"),
+ "differing options must not reuse the cached relation")
+ assert(fresh.getTagValue(LogicalPlan.PLAN_ID_TAG).isEmpty,
+ "differing options should freshly load, not reuse the cached relation")
+ }
+
private def resolve(
unresolvedRelation: UnresolvedRelation,
timeTravelSpec: Option[TimeTravelSpec] = None,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org