From d22835b6fc016f3071e9876dc50e1f32ff26a31f Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Mon, 27 Jul 2026 20:06:46 +0000 Subject: [PATCH 1/8] [SPARK-56961][SQL] Pass all options while loading tables 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 `Version`/`Timestamp` 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`. This PR does not touch the `RelationCatalog` single-RPC `loadRelation(Identifier)` read path, which for a table-and-view catalog is the primary path for a plain read (no time travel / write privileges) and so does not forward options. That is an independent improvement -- it needs nothing from this change (a new `loadRelation(Identifier, CaseInsensitiveStringMap)` overload plus wiring) -- and will be a separate PR. 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. No. New tests in `CatalogV2UtilSuite` (the default-dispatch mapping for each `TableContext` shape, and the time-travel/write-privileges mutual-exclusion invariant) and `DataSourceV2OptionSuite` (end-to-end option forwarding via the DataFrame API, SQL, streaming, time travel, and the write path). Existing `SupportsCatalogOptionsSuite`, `ChangelogResolutionSuite`, `ChangelogEndToEndSuite`, and the DataSourceV2 SQL/DataFrame suites pass. --- .../sql/connector/catalog/TableCatalog.java | 40 +++++++++ .../sql/connector/catalog/TableContext.java | 86 +++++++++++++++++++ .../sql/connector/catalog/TimeTravel.java | 51 +++++++++++ .../analysis/RelationResolution.scala | 3 +- .../sql/connector/catalog/CatalogV2Util.scala | 36 ++++---- .../catalog/CatalogV2UtilSuite.scala | 82 +++++++++++++++++- .../catalog/InMemoryTableCatalog.scala | 21 +++++ .../datasources/v2/DataSourceV2Utils.scala | 2 +- .../connector/DataSourceV2OptionSuite.scala | 86 ++++++++++++++++++- 9 files changed, 383 insertions(+), 24 deletions(-) create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TimeTravel.java 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..17de16a53ceb 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 @@ -194,6 +194,46 @@ 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. + *

+ * The default implementation ignores {@code options} and delegates to the existing + * {@code loadTable} overloads based on {@code context}: {@link #loadTable(Identifier, String)} + * or {@link #loadTable(Identifier, long)} when time travel is present, + * {@link #loadTable(Identifier, Set)} when write privileges are present, otherwise + * {@link #loadTable(Identifier)}. Catalogs that want to receive the user options while reading a + * table (e.g. to customize the scan) must override this method. + * + * @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.2.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.Version v) { + return loadTable(ident, v.version()); + } else if (timeTravel instanceof TimeTravel.Timestamp ts) { + return loadTable(ident, ts.micros()); + } else { + throw new IllegalArgumentException("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..ff20b1b91476 --- /dev/null +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java @@ -0,0 +1,86 @@ +/* + * 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.Collections; +import java.util.HashSet; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +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. + *

+ * 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.2.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 writePrivileges; + + public TableContext(TimeTravel timeTravel, Set writePrivileges) { + Set privileges = (writePrivileges == null) + ? Collections.emptySet() + : Collections.unmodifiableSet(new HashSet<>(writePrivileges)); + if (timeTravel != null && !privileges.isEmpty()) { + throw new IllegalArgumentException("Should not write to a table with time travel"); + } + this.timeTravel = timeTravel; + this.writePrivileges = privileges; + } + + /** Returns the time-travel spec, or empty for a current-version read. */ + public Optional timeTravel() { + return Optional.ofNullable(timeTravel); + } + + /** Returns the requested write privileges; empty for a read. */ + public Set 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..6265bf28a37c --- /dev/null +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TimeTravel.java @@ -0,0 +1,51 @@ +/* + * 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; + +/** + * Represents a time-travel specification for reading a table as of a specific version or point in + * time, passed to the catalog via + * {@link TableCatalog#loadTable(Identifier, TableContext, org.apache.spark.sql.util.CaseInsensitiveStringMap)}. + *

+ * This sealed interface has two implementations: + *

+ * + * @since 4.2.0 + */ +@Evolving +public sealed interface TimeTravel permits TimeTravel.Version, TimeTravel.Timestamp { + + /** + * Time travel to a specific version of the table. + * + * @param version the version identifier (connector-defined) + */ + record Version(String version) implements TimeTravel {} + + /** + * Time travel to a specific point in time. + * + * @param micros microseconds since 1970-01-01 00:00:00 UTC + */ + record Timestamp(long micros) implements TimeTravel {} +} 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 0a085fcc2971..464af0fe7952 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 @@ -280,7 +280,8 @@ class RelationResolution( catalog, ident, finalTimeTravelSpec, - Option(writePrivileges)) + Option(writePrivileges), + finalOptions) } else { None } 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..a509376f9f42 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,24 +485,21 @@ 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.Version(v.version) + case Some(ts: AsOfTimestamp) => new TimeTravel.Timestamp(ts.timestamp) + case None => null + } + val writePrivileges: util.Set[TableWritePrivilege] = writePrivilegesString match { + case Some(str) => + str.split(",").map(_.trim).map(TableWritePrivilege.valueOf).toSet.asJava + case None => + Collections.emptySet() } + val context = new TableContext(timeTravel, writePrivileges) + catalog.asTableCatalog.loadTable(ident, context, options) } /** 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..2b017a22fff6 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,32 @@ package org.apache.spark.sql.connector.catalog -import org.mockito.Mockito.{mock, when} +import java.util.Collections + +import org.mockito.ArgumentMatchers.{any, eq => mockEq} +import org.mockito.Mockito.{mock, verify, when} import org.apache.spark.SparkFunSuite +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 +54,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[IllegalArgumentException] { + CatalogV2Util.getTable(testCatalog, ident, Some(AsOfVersion("v1")), Some("INSERT")) + } + assert(e.getMessage.contains("Should not write to a table with time travel")) + } + + 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 = Collections.emptySet[TableWritePrivilege]() + val a = new TableContext(new TimeTravel.Version("v1"), emptyPrivileges) + val b = new TableContext(new TimeTravel.Version("v1"), emptyPrivileges) + val c = new TableContext(new TimeTravel.Timestamp(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..dbf0373f3d3e 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 @@ -48,6 +48,16 @@ class BasicInMemoryTableCatalog extends TableCatalog { private var _name: Option[String] = None private var copyOnLoad: Boolean = false + // Stores the most recent TableContext and options passed to the options-aware loadTable(), so + // tests can verify that the analyzer / DataFrame API correctly constructed and forwarded them. + // "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 var _lastTableContext: Option[TableContext] = None + def lastTableContext: Option[TableContext] = _lastTableContext + + private var _lastLoadTableOptions: Option[CaseInsensitiveStringMap] = None + def lastLoadTableOptions: Option[CaseInsensitiveStringMap] = _lastLoadTableOptions + override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = { _name = Some(name) copyOnLoad = options.getBoolean("copyOnLoad", false) @@ -124,6 +134,17 @@ 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 = { + _lastTableContext = Some(context) + _lastLoadTableOptions = Some(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/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/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala index f785cac9c012..44ac46fad774 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 @@ -19,9 +19,10 @@ package org.apache.spark.sql.connector 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, TimeTravel} import org.apache.spark.sql.execution.CommandResultExec import org.apache.spark.sql.execution.datasources.v2._ import org.apache.spark.sql.functions.lit @@ -31,6 +32,9 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { 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 +430,84 @@ 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.Version("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) + } + } } From 24cdfef8a8b5fb0717a17dd2f0289116eddfc796 Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Thu, 30 Jul 2026 20:17:49 +0000 Subject: [PATCH 2/8] [SPARK-58389][SQL] Address review comments on loadTable options API - Bump @since on the new TableContext, TimeTravel, and the loadTable overload to 4.3.0. - Rename TimeTravel.Version/Timestamp to AsOfVersion/AsOfTimestamp to match the SQL (AS OF) syntax and the catalyst AsOfVersion/AsOfTimestamp specs. - Throw Spark exceptions (SparkIllegalArgumentException with INTERNAL_ERROR) instead of plain IllegalArgumentException in TableContext and the default loadTable dispatch. - Simplify the TableContext constructor (Set.of()/Set.copyOf, flipped check order) and extract the write-privileges string parsing into a helper. - Stub the new loadTable(ident, TableContext, options) overload (thenCallRealMethod) in the Mockito TableCatalog mocks in PlanResolutionSuite and AlignAssignmentsSuiteBase, and update CatalogV2UtilSuite for the new Spark exception. Resolution now routes through the new overload, so an unstubbed mock returned null and left the relation unresolved. --- .../sql/connector/catalog/TableCatalog.java | 11 ++++++---- .../sql/connector/catalog/TableContext.java | 21 +++++++++---------- .../sql/connector/catalog/TimeTravel.java | 12 +++++------ .../sql/connector/catalog/CatalogV2Util.scala | 19 ++++++++++++----- .../catalog/CatalogV2UtilSuite.scala | 12 +++++------ .../connector/DataSourceV2OptionSuite.scala | 2 +- .../command/AlignAssignmentsSuiteBase.scala | 7 ++++++- .../command/PlanResolutionSuite.scala | 11 +++++++++- 8 files changed, 60 insertions(+), 35 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 17de16a53ceb..df982f26ca71 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; @@ -212,7 +213,7 @@ default Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExce * @return the table's metadata * @throws NoSuchTableException If the table doesn't exist * - * @since 4.2.0 + * @since 4.3.0 */ default Table loadTable( Identifier ident, @@ -220,12 +221,14 @@ default Table loadTable( CaseInsensitiveStringMap options) throws NoSuchTableException { if (context.timeTravel().isPresent()) { TimeTravel timeTravel = context.timeTravel().get(); - if (timeTravel instanceof TimeTravel.Version v) { + if (timeTravel instanceof TimeTravel.AsOfVersion v) { return loadTable(ident, v.version()); - } else if (timeTravel instanceof TimeTravel.Timestamp ts) { + } else if (timeTravel instanceof TimeTravel.AsOfTimestamp ts) { return loadTable(ident, ts.micros()); } else { - throw new IllegalArgumentException("Unsupported time travel spec: " + timeTravel); + throw new SparkIllegalArgumentException( + "INTERNAL_ERROR", + Map.of("message", "Unsupported time travel spec: " + timeTravel)); } } else if (!context.writePrivileges().isEmpty()) { return loadTable(ident, context.writePrivileges()); 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 index ff20b1b91476..9437149895d2 100644 --- 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 @@ -17,12 +17,12 @@ package org.apache.spark.sql.connector.catalog; -import java.util.Collections; -import java.util.HashSet; +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; @@ -34,7 +34,7 @@ * 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.2.0 + * @since 4.3.0 */ @Evolving public class TableContext { @@ -44,15 +44,14 @@ public class TableContext { // Never null; an empty set means no write privileges (i.e. a read). private final Set writePrivileges; - public TableContext(TimeTravel timeTravel, Set writePrivileges) { - Set privileges = (writePrivileges == null) - ? Collections.emptySet() - : Collections.unmodifiableSet(new HashSet<>(writePrivileges)); - if (timeTravel != null && !privileges.isEmpty()) { - throw new IllegalArgumentException("Should not write to a table with time travel"); - } + public TableContext(TimeTravel timeTravel, Set privileges) { this.timeTravel = timeTravel; - this.writePrivileges = privileges; + 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. */ 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 index 6265bf28a37c..436a3a727509 100644 --- 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 @@ -26,26 +26,26 @@ *

* This sealed interface has two implementations: *

    - *
  • {@link Version} -- read the table as of a specific version identifier
    • - *
  • {@link Timestamp} -- read the table as of a specific point in time
    • + *
  • {@link AsOfVersion} -- read the table as of a specific version identifier
    • + *
  • {@link AsOfTimestamp} -- read the table as of a specific point in time
    • *
* - * @since 4.2.0 + * @since 4.3.0 */ @Evolving -public sealed interface TimeTravel permits TimeTravel.Version, TimeTravel.Timestamp { +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 Version(String version) implements TimeTravel {} + 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 Timestamp(long micros) implements TimeTravel {} + record AsOfTimestamp(long micros) implements TimeTravel {} } 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 a509376f9f42..baf2954e036e 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 @@ -488,18 +488,27 @@ private[sql] object CatalogV2Util { writePrivilegesString: Option[String] = None, options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Table = { val timeTravel: TimeTravel = timeTravelSpec match { - case Some(v: AsOfVersion) => new TimeTravel.Version(v.version) - case Some(ts: AsOfTimestamp) => new TimeTravel.Timestamp(ts.timestamp) + case Some(v: AsOfVersion) => new TimeTravel.AsOfVersion(v.version) + case Some(ts: AsOfTimestamp) => new TimeTravel.AsOfTimestamp(ts.timestamp) case None => null } - val writePrivileges: util.Set[TableWritePrivilege] = writePrivilegesString match { + 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 => Collections.emptySet() } - val context = new TableContext(timeTravel, writePrivileges) - catalog.asTableCatalog.loadTable(ident, context, options) } /** 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 2b017a22fff6..36af10953a9e 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 @@ -22,7 +22,7 @@ import java.util.Collections 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 @@ -92,10 +92,10 @@ class CatalogV2UtilSuite extends SparkFunSuite { test("getTable rejects combining time travel and write privileges") { val testCatalog = mockCatalogWithRealDispatch() val ident = mock(classOf[Identifier]) - val e = intercept[IllegalArgumentException] { + val e = intercept[SparkIllegalArgumentException] { CatalogV2Util.getTable(testCatalog, ident, Some(AsOfVersion("v1")), Some("INSERT")) } - assert(e.getMessage.contains("Should not write to a table with time travel")) + assert(e.getMessage.contains("Cannot set both time travel and write privileges")) } test("TableContext normalizes null time travel and null write privileges to empty") { @@ -106,9 +106,9 @@ class CatalogV2UtilSuite extends SparkFunSuite { test("TableContext equals / hashCode / toString") { val emptyPrivileges = Collections.emptySet[TableWritePrivilege]() - val a = new TableContext(new TimeTravel.Version("v1"), emptyPrivileges) - val b = new TableContext(new TimeTravel.Version("v1"), emptyPrivileges) - val c = new TableContext(new TimeTravel.Timestamp(1L), emptyPrivileges) + 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) 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 44ac46fad774..6c06dd013be9 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 @@ -485,7 +485,7 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { val ctx = inMemoryCatalog.lastTableContext assert(ctx.isDefined) assert(ctx.get.timeTravel().isPresent) - assert(ctx.get.timeTravel().get() === new TimeTravel.Version("v1")) + assert(ctx.get.timeTravel().get() === new TimeTravel.AsOfVersion("v1")) val opts = inMemoryCatalog.lastLoadTableOptions assert(opts.isDefined) 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..cc555a8e4544 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 @@ -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 } From 391e9f9270df63d7a81ba58d19f9c0cbe94a0ca3 Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Thu, 30 Jul 2026 21:26:00 +0000 Subject: [PATCH 3/8] [SPARK-58389][SQL] Key the analyzer relation cache on options Now that a catalog's options-aware loadTable can return a different Table depending on the user options (schema, partitioning, snapshot, ...), the per-query relation cache must not hand a reference the Table that a previous reference built with different options. - Add the (write-privilege-stripped) options to the fixed-point relation cache key, extracted into a named RelationCacheKey(nameParts, timeTravelSpec, options) shared by AnalysisContext and RelationResolution, so two references to the same identifier with different options resolve independently. This matches the single-pass resolver's RelationId, which already keys on options, so the two analyzers now agree. - Drop the now-unnecessary applyOptions-on-cache-hit added in SPARK-58330: with options in the key, a hit already implies matching options. - Reject a time-travel spec on a write target (reachable via the option form, e.g. INSERT INTO t WITH ('versionAsOf' = ...)) at the analyzer with a user-facing UNSUPPORTED_FEATURE.TIME_TRAVEL error, rather than letting it fall through to the TableContext mutual-exclusion guard as an INTERNAL_ERROR. Tests: - Record every loadTable(context, options) call in the in-memory catalog and assert that a self-join with different options loads once per option bag, while identical options still load once. - Regression coverage that the two shared caches already respect options: CACHE TABLE's materialized result is not reused for a read carrying different options (CacheManager keys on the plan, which carries options), and a shared relation cache hit re-applies the current read's options via copy(options=). - A time-travel option on a write target surfaces UNSUPPORTED_FEATURE.TIME_TRAVEL. --- .../sql/connector/catalog/TableCatalog.java | 7 +- .../sql/connector/catalog/TimeTravel.java | 10 +- .../sql/catalyst/analysis/Analyzer.scala | 9 +- .../catalyst/analysis/RelationCacheKey.scala | 32 +++++++ .../analysis/RelationResolution.scala | 40 +++----- .../sql/connector/catalog/CatalogV2Util.scala | 2 +- .../catalog/InMemoryTableCatalog.scala | 19 ++-- .../connector/DataSourceV2OptionSuite.scala | 95 +++++++++++++++++++ .../command/PlanResolutionSuite.scala | 41 +++++++- 9 files changed, 200 insertions(+), 55 deletions(-) create mode 100644 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCacheKey.scala 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 df982f26ca71..6afad1fbcbf5 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 @@ -200,11 +200,8 @@ default Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExce * user-specified options. *

* The default implementation ignores {@code options} and delegates to the existing - * {@code loadTable} overloads based on {@code context}: {@link #loadTable(Identifier, String)} - * or {@link #loadTable(Identifier, long)} when time travel is present, - * {@link #loadTable(Identifier, Set)} when write privileges are present, otherwise - * {@link #loadTable(Identifier)}. Catalogs that want to receive the user options while reading a - * table (e.g. to customize the scan) must override this method. + * {@code loadTable} overloads based on {@code context}. Catalogs that want to receive the user + * options while reading a table must override this method. * * @param ident a table identifier * @param context the parsed load parameters (time travel, write privileges) 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 index 436a3a727509..c9ecc80d65d5 100644 --- 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 @@ -20,15 +20,7 @@ import org.apache.spark.annotation.Evolving; /** - * Represents a time-travel specification for reading a table as of a specific version or point in - * time, passed to the catalog via - * {@link TableCatalog#loadTable(Identifier, TableContext, org.apache.spark.sql.util.CaseInsensitiveStringMap)}. - *

- * This sealed interface has two implementations: - *

    - *
  • {@link AsOfVersion} -- read the table as of a specific version identifier
    • - *
  • {@link AsOfTimestamp} -- read the table as of a specific point in time
    • - *
+ * A time-travel specification for reading a table as of a specific version or point in time. * * @since 4.3.0 */ 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 d6eac52ebbb2..5e8ca047e960 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 464af0fe7952..0e9ebf57c285 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 @@ -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 @@ -371,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 => @@ -557,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] = None, + options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): RelationCacheKey = { + RelationCacheKey( + (catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq, 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 baf2954e036e..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 @@ -507,7 +507,7 @@ private[sql] object CatalogV2Util { case Some(str) => str.split(",").map(_.trim).map(TableWritePrivilege.valueOf).toSet.asJava case None => - Collections.emptySet() + util.Set.of() } } 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 dbf0373f3d3e..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,15 +49,18 @@ class BasicInMemoryTableCatalog extends TableCatalog { private var _name: Option[String] = None private var copyOnLoad: Boolean = false - // Stores the most recent TableContext and options passed to the options-aware loadTable(), so - // tests can verify that the analyzer / DataFrame API correctly constructed and forwarded them. + // 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 var _lastTableContext: Option[TableContext] = None - def lastTableContext: Option[TableContext] = _lastTableContext + private val _loadTableCalls = mutable.ArrayBuffer.empty[(TableContext, CaseInsensitiveStringMap)] + def loadTableCalls: Seq[(TableContext, CaseInsensitiveStringMap)] = _loadTableCalls.toSeq + def resetLoadTableCalls(): Unit = _loadTableCalls.clear() - private var _lastLoadTableOptions: Option[CaseInsensitiveStringMap] = None - def lastLoadTableOptions: Option[CaseInsensitiveStringMap] = _lastLoadTableOptions + 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) @@ -140,8 +144,7 @@ class BasicInMemoryTableCatalog extends TableCatalog { ident: Identifier, context: TableContext, options: CaseInsensitiveStringMap): Table = { - _lastTableContext = Some(context) - _lastLoadTableOptions = Some(options) + _loadTableCalls += ((context, options)) super.loadTable(ident, context, options) } 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 6c06dd013be9..f87d8ed022fa 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 @@ -510,4 +510,99 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { assert(opts.get.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES) === null) } } + + 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.collect() + + // Each distinct option bag reaches the catalog as its own loadTable call. + 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") + + // 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). + 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").collect() + + val splitSizeLoads = inMemoryCatalog.loadTableCalls + .count(_._2.get("split-size") === "5") + assert(splitSizeLoads === 1, + s"expected a single loadTable for identical option bags, got: $splitSizeLoads") + } + } + + 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() + } + } + } } 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 cc555a8e4544..09355a4dd052 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 @@ -3407,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") @@ -3465,6 +3465,43 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { } } + test("SPARK-58389: shared relation cache hit re-applies the current read's options") { + AnalysisContext.withNewAnalysisContext { + val ident = Identifier.of(Array.empty[String], "tab") + // testCat.loadTable returns a stable `table` mock, so a relation built from it shares the + // same Table.id and therefore passes `isSameTable` in the shared-cache lookup. + val cachedTable = testCat.loadTable(ident) + // The entry sitting in the shared relation cache (as if left by an earlier CACHE TABLE) + // carries a distinctive option that must NOT leak onto this query's relation. + val cachedRelation = DataSourceV2Relation.create( + cachedTable, Some(testCat), Some(ident), + new CaseInsensitiveStringMap(java.util.Map.of("cachedOnly", "stale"))) + + // A shared relation cache that always returns the stale entry by name -- this is the same + // RelationCache abstraction that SharedState backs with the CacheManager (df.cache()). + val sharedCache: RelationCache = (_, _) => Some(cachedRelation) + + val rule = new RelationResolution(catalogManagerWithDefault, sharedCache) + + // This read carries its own option, different from the cached entry's. + val unresolved = UnresolvedRelation( + Seq("testcat", "tab"), + new CaseInsensitiveStringMap(java.util.Map.of("split-size", "5"))) + + val resolved = rule.resolveRelation(unresolved) match { + case Some(AsDataSourceV2Relation(relation)) => relation + case other => fail(s"failed to resolve as v2 relation: $other") + } + + // The shared-cache entry was reused (same Table), proving we went through that branch. + assert(resolved.table == cachedTable) + // ... but the relation carries THIS read's option, not the stale cached one: the branch + // re-applies `finalOptions` via `cached.copy(options = ...)`. + assert(resolved.options.get("split-size") === "5") + assert(!resolved.options.containsKey("cachedOnly")) + } + } + private def resolve( unresolvedRelation: UnresolvedRelation, timeTravelSpec: Option[TimeTravelSpec] = None, From a58468c0e8928880ca30d355796ac83cec39d308 Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Fri, 31 Jul 2026 18:18:21 +0000 Subject: [PATCH 4/8] [SPARK-58389][SQL] Address review feedback on loadTable options Follow-up to the review by @peter-toth. Blocking findings: - Key the V2TableReference cache path (getOrLoadRelation) on the reference's options. It stored a relation carrying ref.options under an option-free key, so a later option-free reference to the same table inherited them; removing applyOptions-on-hit had unmasked this. Also remove toCacheKey's defaults so a caller cannot silently omit the options again. - Gate shared relation cache (CACHE TABLE) reuse on the read's options matching the cached relation's. The lookup matches by name and Table.id, neither of which sees options, and reuse buys nothing when they differ (the resulting plan no longer matches the cached fingerprint anyway). - Document that overriding loadTable(ident, context, options) takes over honoring context: applying time travel and authorizing write privileges as in loadTable(ident, Set); Spark does not re-check. Non-blocking: - Forward the user options on the SupportsCatalogOptions write path (DataFrameWriter), mirroring the read path in DataSourceV2Utils. - Add a test for the SupportsCatalogOptions read path forwarding options to the catalog's loadTable. Tests: a temp view's options do not leak to a later reference; shared relation cache reused only when options match; read options forwarded on the SupportsCatalogOptions path. --- .../sql/connector/catalog/TableCatalog.java | 5 ++ .../analysis/RelationResolution.scala | 13 ++-- .../spark/sql/classic/DataFrameWriter.scala | 5 +- .../connector/DataSourceV2OptionSuite.scala | 24 ++++++ .../SupportsCatalogOptionsSuite.scala | 15 ++++ .../command/PlanResolutionSuite.scala | 77 +++++++++++-------- 6 files changed, 100 insertions(+), 39 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 6afad1fbcbf5..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 @@ -202,6 +202,11 @@ default Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExce * 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. + *

+ * 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) 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 0e9ebf57c285..853d610f5460 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 @@ -310,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) } @@ -485,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) @@ -545,8 +548,8 @@ class RelationResolution( private def toCacheKey( catalog: CatalogPlugin, ident: Identifier, - timeTravelSpec: Option[TimeTravelSpec] = None, - options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): RelationCacheKey = { + timeTravelSpec: Option[TimeTravelSpec], + options: CaseInsensitiveStringMap): RelationCacheKey = { RelationCacheKey( (catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq, timeTravelSpec, options) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala index a9f16ffa87be..88df3799f1ab 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameWriter.scala @@ -176,7 +176,10 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) extends sql.DataFram val catalog = CatalogV2Util.getTableProviderCatalog( supportsExtract, catalogManager, dsOptions) - (catalog.loadTable(ident), Some(catalog), Some(ident)) + // Forward the user options to the catalog, mirroring the read path in + // DataSourceV2Utils.loadV2Source. + (CatalogV2Util.getTable(catalog, ident, options = dsOptions), + Some(catalog), Some(ident)) case _: TableProvider => val t = getTable if (t.supports(BATCH_WRITE)) { 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 f87d8ed022fa..d9311b7e79a2 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 @@ -605,4 +605,28 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { } } } + + 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/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala index 09355a4dd052..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 @@ -3465,41 +3465,52 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { } } - test("SPARK-58389: shared relation cache hit re-applies the current read's options") { - AnalysisContext.withNewAnalysisContext { - val ident = Identifier.of(Array.empty[String], "tab") - // testCat.loadTable returns a stable `table` mock, so a relation built from it shares the - // same Table.id and therefore passes `isSameTable` in the shared-cache lookup. - val cachedTable = testCat.loadTable(ident) - // The entry sitting in the shared relation cache (as if left by an earlier CACHE TABLE) - // carries a distinctive option that must NOT leak onto this query's relation. - val cachedRelation = DataSourceV2Relation.create( - cachedTable, Some(testCat), Some(ident), - new CaseInsensitiveStringMap(java.util.Map.of("cachedOnly", "stale"))) - - // A shared relation cache that always returns the stale entry by name -- this is the same - // RelationCache abstraction that SharedState backs with the CacheManager (df.cache()). - val sharedCache: RelationCache = (_, _) => Some(cachedRelation) - - val rule = new RelationResolution(catalogManagerWithDefault, sharedCache) - - // This read carries its own option, different from the cached entry's. - val unresolved = UnresolvedRelation( - Seq("testcat", "tab"), - new CaseInsensitiveStringMap(java.util.Map.of("split-size", "5"))) - - val resolved = rule.resolveRelation(unresolved) match { - case Some(AsDataSourceV2Relation(relation)) => relation - case other => fail(s"failed to resolve as v2 relation: $other") + 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") + } } - - // The shared-cache entry was reused (same Table), proving we went through that branch. - assert(resolved.table == cachedTable) - // ... but the relation carries THIS read's option, not the stale cached one: the branch - // re-applies `finalOptions` via `cached.copy(options = ...)`. - assert(resolved.options.get("split-size") === "5") - assert(!resolved.options.containsKey("cachedOnly")) } + + // 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( From 96b9ddab6787ce6201e6d554653cb0cfabe8af9e Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Fri, 31 Jul 2026 19:45:39 +0000 Subject: [PATCH 5/8] [SPARK-58389][SQL] Preserve table options when recaching a cached table tryRefreshPlan rebuilt the CacheManager entry for a cached table with a fresh DataSourceV2Relation that dropped the read options, so after REFRESH TABLE a read carrying the original options no longer matched the cached plan and missed the cache for no reason. Reload the Table with the relation's options and rebuild the relation with them. Test: a cached read with options, after refreshTable, forwards the options to loadTable and keeps them on the recached relation, while an option-free read still does not reuse it. --- .../spark/sql/execution/CacheManager.scala | 6 +-- .../connector/DataSourceV2OptionSuite.scala | 39 +++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) 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/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala index d9311b7e79a2..aa4bb88398a9 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 @@ -606,6 +606,45 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { } } + 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: a DataFrame temp view's options do not leak to a later reference") { val t1 = s"${catalogAndNamespace}table" withTable(t1) { From cbba1920a93ccd7d299649d39633de7fc9c538d7 Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Fri, 31 Jul 2026 21:43:27 +0000 Subject: [PATCH 6/8] [SPARK-58389][SQL] Preserve options when refreshing V2 tables --- .../datasources/v2/V2TableRefreshUtil.scala | 12 ++++---- .../connector/DataSourceV2OptionSuite.scala | 28 ++++++++++++++++--- 2 files changed, 31 insertions(+), 9 deletions(-) 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..19d45a1280a8 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,6 +26,7 @@ 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 @@ -81,19 +82,20 @@ 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[(TableCatalog, Identifier, CaseInsensitiveStringMap), 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 aa4bb88398a9..cd440bc6024b 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 @@ -524,9 +524,9 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { // 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.collect() + df.queryExecution.analyzed - // Each distinct option bag reaches the catalog as its own loadTable call. + // 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) @@ -534,6 +534,17 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { 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") @@ -551,13 +562,22 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { // 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). - 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").collect() + 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") } } From 76fc51de35da6b73a5e508601e7f810073497aa7 Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Fri, 31 Jul 2026 21:54:27 +0000 Subject: [PATCH 7/8] [SPARK-58389][SQL] Add V2 refresh regression coverage --- .../connector/DataSourceV2OptionSuite.scala | 99 ++++++++++++++++++- 1 file changed, 98 insertions(+), 1 deletion(-) 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 cd440bc6024b..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,16 +17,27 @@ 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.{Identifier, InMemoryBaseTable, InMemoryCatalog, InMemoryRowLevelOperationTableCatalog, TimeTravel} +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._ @@ -511,6 +522,32 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { } } + 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) { @@ -626,6 +663,35 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { } } + 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) { @@ -665,6 +731,37 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { } } + 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) { From 3377cd992f1dcc7fa46efe6838870088293508c5 Mon Sep 17 00:00:00 2001 From: Yan Yan Date: Sat, 1 Aug 2026 00:42:31 +0000 Subject: [PATCH 8/8] [SPARK-58389][SQL] Fix Scala style --- .../spark/sql/connector/catalog/CatalogV2UtilSuite.scala | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) 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 36af10953a9e..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,8 +17,6 @@ package org.apache.spark.sql.connector.catalog -import java.util.Collections - import org.mockito.ArgumentMatchers.{any, eq => mockEq} import org.mockito.Mockito.{mock, verify, when} @@ -105,7 +103,7 @@ class CatalogV2UtilSuite extends SparkFunSuite { } test("TableContext equals / hashCode / toString") { - val emptyPrivileges = Collections.emptySet[TableWritePrivilege]() + 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)