diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java index 23e9499932c1..58a40c237332 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java @@ -17,6 +17,7 @@ package org.apache.spark.sql.connector.catalog; +import org.apache.spark.SparkIllegalArgumentException; import org.apache.spark.annotation.Evolving; import org.apache.spark.sql.connector.expressions.Transform; import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; @@ -194,6 +195,50 @@ 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}. 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) + * @param options all options passed to the read, including any keys that are also parsed into + * {@code context} + * @return the table's metadata + * @throws NoSuchTableException If the table doesn't exist + * + * @since 4.3.0 + */ + default Table loadTable( + Identifier ident, + TableContext context, + CaseInsensitiveStringMap options) throws NoSuchTableException { + if (context.timeTravel().isPresent()) { + TimeTravel timeTravel = context.timeTravel().get(); + if (timeTravel instanceof TimeTravel.AsOfVersion v) { + return loadTable(ident, v.version()); + } else if (timeTravel instanceof TimeTravel.AsOfTimestamp ts) { + return loadTable(ident, ts.micros()); + } else { + throw new SparkIllegalArgumentException( + "INTERNAL_ERROR", + Map.of("message", "Unsupported time travel spec: " + timeTravel)); + } + } else if (!context.writePrivileges().isEmpty()) { + return loadTable(ident, context.writePrivileges()); + } else { + return loadTable(ident); + } + } + /** * Load a {@link Changelog} for the given table, representing the row-level changes within the * range specified by {@code context}. diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java new file mode 100644 index 000000000000..9437149895d2 --- /dev/null +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableContext.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector.catalog; + +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +import org.apache.spark.SparkIllegalArgumentException; +import org.apache.spark.annotation.Evolving; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +/** + * Encapsulates the parsed, Spark-recognized parameters of a table load request, passed from the + * analyzer / DataFrame API to the catalog's + * {@link TableCatalog#loadTable(Identifier, TableContext, CaseInsensitiveStringMap)} method. + *

+ * A load is either a read (optionally with time travel) or a write (carrying write privileges); + * time travel and write privileges are mutually exclusive. + * + * @since 4.3.0 + */ +@Evolving +public class TableContext { + + // null means no time travel. + private final TimeTravel timeTravel; + // Never null; an empty set means no write privileges (i.e. a read). + private final Set writePrivileges; + + public TableContext(TimeTravel timeTravel, Set privileges) { + this.timeTravel = timeTravel; + this.writePrivileges = privileges == null ? Set.of() : Set.copyOf(privileges); + if (timeTravel != null && !writePrivileges.isEmpty()) { + throw new SparkIllegalArgumentException( + "INTERNAL_ERROR", + Map.of("message", "Cannot set both time travel and write privileges")); + } + } + + /** Returns the time-travel spec, or empty for a current-version read. */ + public Optional timeTravel() { + 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..c9ecc80d65d5 --- /dev/null +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TimeTravel.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector.catalog; + +import org.apache.spark.annotation.Evolving; + +/** + * A time-travel specification for reading a table as of a specific version or point in time. + * + * @since 4.3.0 + */ +@Evolving +public sealed interface TimeTravel permits TimeTravel.AsOfVersion, TimeTravel.AsOfTimestamp { + + /** + * Time travel to a specific version of the table. + * + * @param version the version identifier (connector-defined) + */ + record AsOfVersion(String version) implements TimeTravel {} + + /** + * Time travel to a specific point in time. + * + * @param micros microseconds since 1970-01-01 00:00:00 UTC + */ + record AsOfTimestamp(long micros) implements TimeTravel {} +} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 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 0a085fcc2971..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 @@ -61,11 +61,10 @@ class RelationResolution( with LookupCatalog with SQLConfHelper { - type CacheKey = (Seq[String], Option[TimeTravelSpec]) - val v1SessionCatalog = catalogManager.v1SessionCatalog - private def relationCache: mutable.Map[CacheKey, LogicalPlan] = AnalysisContext.get.relationCache + private def relationCache: mutable.Map[RelationCacheKey, LogicalPlan] = + AnalysisContext.get.relationCache /** * If we are resolving database objects (relations, functions, etc.) inside views, we may need to @@ -236,24 +235,26 @@ class RelationResolution( finalTimeTravelSpec: Option[TimeTravelSpec]): Option[LogicalPlan] = { expandIdentifier(identifier) match { case CatalogAndIdentifier(catalog, ident) => - val key = toCacheKey(catalog, ident, finalTimeTravelSpec) val planId = u.getTagValue(LogicalPlan.PLAN_ID_TAG) val writePrivileges = u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES) val finalOptions = u.clearWritePrivileges.options + // Time travel applies to reads only; reject it on a write target (reachable via the option + // form, e.g. `INSERT INTO t WITH ('versionAsOf' = ...)`) with a user-facing error. + if (finalTimeTravelSpec.nonEmpty && writePrivileges != null) { + throw QueryCompilationErrors.timeTravelUnsupportedError(toSQLId(identifier)) + } + val key = toCacheKey(catalog, ident, finalTimeTravelSpec, finalOptions) // A reference that requires write privileges is never served from the per-query relation // cache. The catalog authorizes the write in `loadTable(ident, writePrivileges)` below, and // a cache hit would skip that call entirely. The hit happens whenever the write target is // also read in the same statement -- the target is resolved after its query (see // `ResolveRelations`), so it finds the relation the query already put in the cache, e.g. // for `INSERT INTO t SELECT * FROM t`. + // + // The cache key includes the options, so a hit means the options already match and each + // reference's own bag is honored without re-applying it here. val cached = if (writePrivileges == null) relationCache.get(key) else None cached - // The per-query relation cache is not keyed by options. When the same table is referenced - // more than once in a single statement with different dynamic options (e.g. a self-join, - // or a second reference sharing the target's cache entry), a cache hit would otherwise - // reuse the first reference's options and silently drop this reference's. Re-apply this - // reference's options to the cached relation so each reference honors its own bag. - .map(applyOptions(_, finalOptions)) .map(adaptCachedRelation(_, planId)) .orElse { // For a `RelationCatalog` with no time-travel / write privileges, the single-RPC @@ -280,7 +281,8 @@ class RelationResolution( catalog, ident, finalTimeTravelSpec, - Option(writePrivileges)) + Option(writePrivileges), + finalOptions) } else { None } @@ -308,14 +310,17 @@ class RelationResolution( // we don't share-cache views. val table: Option[Table] = relation.collect { case t: Table => t } + // Reuse a cached relation only when this read's options match: the lookup is by name + // and `Table.id`, so a differing-options read would otherwise get the cached read's + // `Table`. val sharedRelationCacheMatch = for { t <- table if finalTimeTravelSpec.isEmpty && writePrivileges == null && !u.isStreaming cached <- lookupSharedRelationCache(catalog, ident, t) + if cached.options == finalOptions } yield { - val updatedRelation = cached.copy(options = finalOptions) val nameParts = ident.toQualifiedNameParts(catalog) - val aliasedRelation = SubqueryAlias(nameParts, updatedRelation) + val aliasedRelation = SubqueryAlias(nameParts, cached) relationCache.update(key, aliasedRelation) adaptCachedRelation(aliasedRelation, planId) } @@ -370,19 +375,6 @@ class RelationResolution( CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident, table, conf) } - /** - * Re-applies `options` to the relation in a cached plan. Every `relationCache` entry holds a - * single relation for its own identifier (a view's body is still unresolved when it is cached), - * so this cannot reach another table's relation. - */ - private def applyOptions( - cached: LogicalPlan, - options: CaseInsensitiveStringMap): LogicalPlan = cached transform { - case r: DataSourceV2Relation => r.copy(options = options) - case r: UnresolvedCatalogRelation => r.copy(options = options) - case r: StreamingRelationV2 => r.copy(extraOptions = options) - } - private def adaptCachedRelation(cached: LogicalPlan, planId: Option[Long]): LogicalPlan = { val plan = cached transform { case multi: MultiInstanceRelation => @@ -496,7 +488,7 @@ class RelationResolution( } private def getOrLoadRelation(ref: V2TableReference): LogicalPlan = { - val key = toCacheKey(ref.catalog, ref.identifier) + val key = toCacheKey(ref.catalog, ref.identifier, None, ref.options) relationCache.get(key) match { case Some(cached) => adaptCachedRelation(cached, ref) @@ -556,8 +548,10 @@ class RelationResolution( private def toCacheKey( catalog: CatalogPlugin, ident: Identifier, - timeTravelSpec: Option[TimeTravelSpec] = None): CacheKey = { - ((catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq, timeTravelSpec) + timeTravelSpec: Option[TimeTravelSpec], + options: CaseInsensitiveStringMap): RelationCacheKey = { + RelationCacheKey( + (catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq, timeTravelSpec, options) } private def cloneWithPlanId(plan: LogicalPlan, planId: Option[Long]): LogicalPlan = { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala index c0905ede4f0d..28aaae4a81ff 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala @@ -472,9 +472,10 @@ private[sql] object CatalogV2Util { catalog: CatalogPlugin, ident: Identifier, timeTravelSpec: Option[TimeTravelSpec] = None, - writePrivilegesString: Option[String] = None): Option[Table] = + writePrivilegesString: Option[String] = None, + options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Option[Table] = try { - Option(getTable(catalog, ident, timeTravelSpec, writePrivilegesString)) + Option(getTable(catalog, ident, timeTravelSpec, writePrivilegesString, options)) } catch { case _: NoSuchTableException => None case _: NoSuchDatabaseException => None @@ -484,23 +485,29 @@ private[sql] object CatalogV2Util { catalog: CatalogPlugin, ident: Identifier, timeTravelSpec: Option[TimeTravelSpec] = None, - writePrivilegesString: Option[String] = None): Table = { - if (timeTravelSpec.nonEmpty) { - assert(writePrivilegesString.isEmpty, "Should not write to a table with time travel") - timeTravelSpec.get match { - case v: AsOfVersion => - catalog.asTableCatalog.loadTable(ident, v.version) - case ts: AsOfTimestamp => - catalog.asTableCatalog.loadTable(ident, ts.timestamp) - } - } else { - if (writePrivilegesString.isDefined) { - val writePrivileges = writePrivilegesString.get.split(",").map(_.trim) - .map(TableWritePrivilege.valueOf).toSet.asJava - catalog.asTableCatalog.loadTable(ident, writePrivileges) - } else { - catalog.asTableCatalog.loadTable(ident) - } + writePrivilegesString: Option[String] = None, + options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Table = { + val timeTravel: TimeTravel = timeTravelSpec match { + case Some(v: AsOfVersion) => new TimeTravel.AsOfVersion(v.version) + case Some(ts: AsOfTimestamp) => new TimeTravel.AsOfTimestamp(ts.timestamp) + case None => null + } + val context = new TableContext(timeTravel, parseWritePrivileges(writePrivilegesString)) + catalog.asTableCatalog.loadTable(ident, context, options) + } + + /** + * Parses the comma-separated write-privileges string (as carried in the internal + * [[org.apache.spark.sql.catalyst.analysis.UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES]] + * option) into a set of [[TableWritePrivilege]]. Returns an empty set when absent (a read). + */ + private def parseWritePrivileges( + writePrivilegesString: Option[String]): util.Set[TableWritePrivilege] = { + writePrivilegesString match { + case Some(str) => + str.split(",").map(_.trim).map(TableWritePrivilege.valueOf).toSet.asJava + case None => + util.Set.of() } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala index eda401ceb6bd..4b9d55be07e3 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogV2UtilSuite.scala @@ -17,15 +17,30 @@ package org.apache.spark.sql.connector.catalog -import org.mockito.Mockito.{mock, when} +import org.mockito.ArgumentMatchers.{any, eq => mockEq} +import org.mockito.Mockito.{mock, verify, when} -import org.apache.spark.SparkFunSuite +import org.apache.spark.{SparkFunSuite, SparkIllegalArgumentException} +import org.apache.spark.sql.catalyst.analysis.{AsOfTimestamp, AsOfVersion, TimeTravelSpec} import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation import org.apache.spark.sql.types.IntegerType +import org.apache.spark.sql.util.CaseInsensitiveStringMap class CatalogV2UtilSuite extends SparkFunSuite { - test("Load relation should encode the identifiers for V2Relations") { + + // CatalogV2Util.getTable routes through the options-aware TableCatalog.loadTable, whose default + // implementation dispatches to the existing overloads. Stub only that method to run the real + // default so the dispatch is exercised; the leaf overloads stay as plain mock methods (returning + // null) that we then `verify`. + private def mockCatalogWithRealDispatch(): TableCatalog = { val testCatalog = mock(classOf[TableCatalog]) + when(testCatalog.loadTable( + any[Identifier], any[TableContext], any[CaseInsensitiveStringMap])).thenCallRealMethod() + testCatalog + } + + test("Load relation should encode the identifiers for V2Relations") { + val testCatalog = mockCatalogWithRealDispatch() val ident = mock(classOf[Identifier]) val table = mock(classOf[Table]) when(table.columns()).thenReturn(Array(Column.create("i", IntegerType))) @@ -37,4 +52,65 @@ class CatalogV2UtilSuite extends SparkFunSuite { assert(v2Relation.catalog.exists(_ == testCatalog)) assert(v2Relation.identifier.exists(_ == ident)) } + + private def getTableAndVerifyDispatch( + timeTravelSpec: Option[TimeTravelSpec], + writePrivilegesString: Option[String])( + verifyOverload: TableCatalog => Unit): Unit = { + val testCatalog = mockCatalogWithRealDispatch() + val ident = mock(classOf[Identifier]) + CatalogV2Util.getTable(testCatalog, ident, timeTravelSpec, writePrivilegesString) + verifyOverload(testCatalog) + } + + test("getTable dispatches to loadTable(ident) with no time travel and no write privileges") { + getTableAndVerifyDispatch(None, None) { c => verify(c).loadTable(any[Identifier]) } + } + + test("getTable dispatches to loadTable(ident, writePrivileges) with write privileges") { + getTableAndVerifyDispatch(None, Some("INSERT,DELETE")) { c => + verify(c).loadTable( + any[Identifier], + mockEq(java.util.Set.of(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))) + } + } + + test("getTable dispatches to loadTable(ident, version) for version time travel") { + getTableAndVerifyDispatch(Some(AsOfVersion("v1")), None) { c => + verify(c).loadTable(any[Identifier], mockEq("v1")) + } + } + + test("getTable dispatches to loadTable(ident, timestamp) for timestamp time travel") { + getTableAndVerifyDispatch(Some(AsOfTimestamp(123L)), None) { c => + verify(c).loadTable(any[Identifier], mockEq(123L)) + } + } + + test("getTable rejects combining time travel and write privileges") { + val testCatalog = mockCatalogWithRealDispatch() + val ident = mock(classOf[Identifier]) + val e = intercept[SparkIllegalArgumentException] { + CatalogV2Util.getTable(testCatalog, ident, Some(AsOfVersion("v1")), Some("INSERT")) + } + assert(e.getMessage.contains("Cannot set both time travel and write privileges")) + } + + test("TableContext normalizes null time travel and null write privileges to empty") { + val context = new TableContext(null, null) + assert(context.timeTravel().isEmpty) + assert(context.writePrivileges().isEmpty) + } + + test("TableContext equals / hashCode / toString") { + val emptyPrivileges = java.util.Set.of[TableWritePrivilege]() + val a = new TableContext(new TimeTravel.AsOfVersion("v1"), emptyPrivileges) + val b = new TableContext(new TimeTravel.AsOfVersion("v1"), emptyPrivileges) + val c = new TableContext(new TimeTravel.AsOfTimestamp(1L), emptyPrivileges) + assert(a == b) + assert(a.hashCode() == b.hashCode()) + assert(a != c) + assert(a.toString.contains("timeTravel")) + assert(a.toString.contains("writePrivileges")) + } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala index bb137ba4830d..53a7fa05b40e 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala @@ -22,6 +22,7 @@ import java.util.Collections import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger +import scala.collection.mutable import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.InternalRow @@ -48,6 +49,19 @@ class BasicInMemoryTableCatalog extends TableCatalog { private var _name: Option[String] = None private var copyOnLoad: Boolean = false + // Records every (TableContext, options) pair passed to the options-aware loadTable(), in call + // order, so tests can verify that the analyzer / DataFrame API correctly constructed and + // forwarded them -- including how many times loadTable was called when the same table is + // referenced more than once in a statement with different options. + // "loadTable" is in the name because the subclass InMemoryChangelogCatalog has an analogous + // `lastOptions` recording the options passed to loadChangelog(); the two must not collide. + private val _loadTableCalls = mutable.ArrayBuffer.empty[(TableContext, CaseInsensitiveStringMap)] + def loadTableCalls: Seq[(TableContext, CaseInsensitiveStringMap)] = _loadTableCalls.toSeq + def resetLoadTableCalls(): Unit = _loadTableCalls.clear() + + def lastTableContext: Option[TableContext] = _loadTableCalls.lastOption.map(_._1) + def lastLoadTableOptions: Option[CaseInsensitiveStringMap] = _loadTableCalls.lastOption.map(_._2) + override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = { _name = Some(name) copyOnLoad = options.getBoolean("copyOnLoad", false) @@ -124,6 +138,16 @@ class BasicInMemoryTableCatalog extends TableCatalog { } } + // Records the forwarded context/options so tests can verify they reached the catalog, then + // defers to the default dispatch in TableCatalog (rather than reimplementing it here). + override def loadTable( + ident: Identifier, + context: TableContext, + options: CaseInsensitiveStringMap): Table = { + _loadTableCalls += ((context, options)) + super.loadTable(ident, context, options) + } + override def invalidateTable(ident: Identifier): Unit = { invalidatedTables.add(ident) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/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/main/scala/org/apache/spark/sql/execution/CacheManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala index 345c1d5d635f..3541c939909f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala @@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan, Resolv import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION import org.apache.spark.sql.catalyst.util.sideBySide import org.apache.spark.sql.classic.{Dataset, SparkSession} -import org.apache.spark.sql.connector.catalog.CatalogPlugin +import org.apache.spark.sql.connector.catalog.{CatalogPlugin, CatalogV2Util} import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.{IdentifierHelper, MultipartIdentifierHelper} import org.apache.spark.sql.connector.catalog.Identifier import org.apache.spark.sql.connector.catalog.transactions.Transaction @@ -419,9 +419,9 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper { try { EliminateSubqueryAliases(plan) match { case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if r.timeTravelSpec.isEmpty => - val table = catalog.loadTable(ident) + val table = CatalogV2Util.getTable(catalog, ident, options = r.options) if (r.table.id == table.id) { - Some(DataSourceV2Relation.create(table, Some(catalog), Some(ident))) + Some(DataSourceV2Relation.create(table, Some(catalog), Some(ident), r.options)) } else { None } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala index a3b5c5aeb799..6f7447faa621 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala @@ -141,7 +141,7 @@ private[sql] object DataSourceV2Utils extends Logging { } val timeTravel = TimeTravelSpec.create( timeTravelTimestamp, timeTravelVersion, conf.sessionLocalTimeZone) - val tbl = CatalogV2Util.getTable(catalog, ident, timeTravel) + val tbl = CatalogV2Util.getTable(catalog, ident, timeTravel, options = dsOptions) (tbl, Some(catalog), Some(ident), timeTravel) case _ => // TODO: Non-catalog paths for DSV2 are currently not well defined. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala index f1ff11b1a4a6..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 f785cac9c012..8cdf12b18f44 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala @@ -17,20 +17,35 @@ package org.apache.spark.sql.connector +import java.util.concurrent.atomic.AtomicInteger + import org.apache.spark.sql.{AnalysisException, Row} import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured +import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 -import org.apache.spark.sql.connector.catalog.{InMemoryBaseTable, InMemoryRowLevelOperationTableCatalog} +import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryBaseTable, InMemoryCatalog, InMemoryRowLevelOperationTableCatalog, Table, TimeTravel} import org.apache.spark.sql.execution.CommandResultExec import org.apache.spark.sql.execution.datasources.v2._ import org.apache.spark.sql.functions.lit +class LoadCountingInMemoryCatalog extends InMemoryCatalog { + val singleArgLoads = new AtomicInteger(0) + + override def loadTable(ident: Identifier): Table = { + singleArgLoads.incrementAndGet() + super.loadTable(ident) + } +} + class DataSourceV2OptionSuite extends DatasourceV2SQLBase { import testImplicits._ private val catalogAndNamespace = "testcat.ns1.ns2." + private def inMemoryCatalog: InMemoryCatalog = + catalog("testcat").asInstanceOf[InMemoryCatalog] + test("SPARK-36680: Supports Dynamic Table Options for SQL Select") { val t1 = s"${catalogAndNamespace}table" withTable(t1) { @@ -426,4 +441,348 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { assert (collected.size == 1) } } + + test("options are forwarded to loadTable - DataFrame API") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + spark.read.option("customOption", "customValue").table(t1).collect() + + val opts = inMemoryCatalog.lastLoadTableOptions + assert(opts.isDefined) + assert(opts.get.get("customOption") === "customValue") + } + } + + test("options are forwarded to loadTable - SQL") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"SELECT * FROM $t1 WITH ('customOption' = 'customValue')").collect() + + val opts = inMemoryCatalog.lastLoadTableOptions + assert(opts.isDefined) + assert(opts.get.get("customOption") === "customValue") + } + } + + test("options are forwarded to loadTable - DataStreamReader") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + // Trigger analysis of the streaming relation. + spark.readStream.option("customOption", "customValue").table(t1).queryExecution.analyzed + + val opts = inMemoryCatalog.lastLoadTableOptions + assert(opts.isDefined) + assert(opts.get.get("customOption") === "customValue") + } + } + + test("options are forwarded to loadTable alongside time travel") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + // versionAsOf is the default time-travel version option key + // (SQLConf.TIME_TRAVEL_VERSION_KEY). Pin a versioned copy so the versioned load succeeds. + inMemoryCatalog.pinTable(Identifier.of(Array("ns1", "ns2"), "table"), "v1") + + spark.read + .option("versionAsOf", "v1") + .option("customOption", "customValue") + .table(t1) + .collect() + + val ctx = inMemoryCatalog.lastTableContext + assert(ctx.isDefined) + assert(ctx.get.timeTravel().isPresent) + assert(ctx.get.timeTravel().get() === new TimeTravel.AsOfVersion("v1")) + + val opts = inMemoryCatalog.lastLoadTableOptions + assert(opts.isDefined) + assert(opts.get.get("customOption") === "customValue") + assert(opts.get.get("versionAsOf") === "v1") + } + } + + test("write privileges are carried in TableContext, internal key stripped") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a')") + + val ctx = inMemoryCatalog.lastTableContext + assert(ctx.isDefined) + assert(!ctx.get.writePrivileges().isEmpty) + + val opts = inMemoryCatalog.lastLoadTableOptions + assert(opts.isDefined) + // The internal write-privileges marker must not leak to the connector as a user option. + assert(opts.get.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES) === null) + } + } + + test("SPARK-58389: execution refresh forwards options on a plain table read") { + registerCatalog("loadcounting", classOf[LoadCountingInMemoryCatalog]) + val loadCountingCatalog = + catalog("loadcounting").asInstanceOf[LoadCountingInMemoryCatalog] + val t1 = "loadcounting.ns1.ns2.table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + loadCountingCatalog.resetLoadTableCalls() + loadCountingCatalog.singleArgLoads.set(0) + + spark.read.option("split-size", "5").table(t1).collect() + + // Both analysis and the execution-time refresh must enter through the options-aware + // overload. Each then delegates to the single-argument overload in this test catalog. + val optionAwareLoads = loadCountingCatalog.loadTableCalls + .map(_._2.get("split-size")) + .filter(_ != null) + assert(optionAwareLoads === Seq("5", "5"), + s"expected analysis and refresh to forward split-size=5, got: $optionAwareLoads") + assert(loadCountingCatalog.singleArgLoads.get() === 2, + s"expected two delegated single-argument loads, got: " + + loadCountingCatalog.singleArgLoads.get()) + } + } + + test("SPARK-58389: a self-join with different options loads the table once per option bag") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + inMemoryCatalog.resetLoadTableCalls() + + // The two references share a name but carry different options. Because a catalog's + // options-aware loadTable can return a different Table depending on the options, the analyzer + // relation cache is keyed on the options, so each reference triggers its own loadTable rather + // than reusing the first reference's Table. + val df = sql(s"SELECT a.id FROM $t1 WITH (`split-size` = 5) a " + + s"JOIN $t1 WITH (`split-size` = 9) b ON a.id = b.id") + df.queryExecution.analyzed + + // Each distinct option bag reaches the catalog as its own loadTable call during analysis. + val loadedOptions = inMemoryCatalog.loadTableCalls + .map(_._2.get("split-size")) + .filter(_ != null) + .sorted + assert(loadedOptions === Seq("5", "9"), + s"expected one loadTable per distinct option bag, got: $loadedOptions") + + // The execution-time refresh also reloads once per option bag instead of sharing one Table + // across the two references. + inMemoryCatalog.resetLoadTableCalls() + df.collect() + val refreshedOptions = inMemoryCatalog.loadTableCalls + .map(_._2.get("split-size")) + .filter(_ != null) + .sorted + assert(refreshedOptions === Seq("5", "9"), + s"expected refresh to load each distinct option bag, got: $refreshedOptions") + + // Each scan also keeps its own option end-to-end (neither reference inherits the other's). + val splitSizes = df.queryExecution.optimizedPlan.collect { + case s: DataSourceV2ScanRelation => s.relation.options.get("split-size") + }.sorted + assert(splitSizes === Seq("5", "9")) + } + } + + test("SPARK-58389: repeated references with the same options load the table once") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + inMemoryCatalog.resetLoadTableCalls() + + // Both references carry the same options, so they share one relation-cache entry: the table + // is loaded once (resolve-once-per-query is preserved for identical option bags). + val df = sql(s"SELECT a.id FROM $t1 WITH (`split-size` = 5) a " + + s"JOIN $t1 WITH (`split-size` = 5) b ON a.id = b.id") + df.queryExecution.analyzed + + val splitSizeLoads = inMemoryCatalog.loadTableCalls + .count(_._2.get("split-size") === "5") + assert(splitSizeLoads === 1, + s"expected a single loadTable for identical option bags, got: $splitSizeLoads") + + // The refresh phase also reuses one load for repeated references with identical options. + inMemoryCatalog.resetLoadTableCalls() + df.collect() + val refreshedSplitSizeLoads = inMemoryCatalog.loadTableCalls + .count(_._2.get("split-size") === "5") + assert(refreshedSplitSizeLoads === 1, + s"expected refresh to load identical option bags once, got: $refreshedSplitSizeLoads") + } + } + + test("SPARK-58389: time travel option on a write target is rejected with a user-facing error") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + + // A time-travel option on a write target is reachable via the option form (the `AS OF` + // syntax is blocked earlier by the parser). It must surface as a user-facing analysis error, + // not the internal TableContext mutual-exclusion guard (which would report INTERNAL_ERROR). + checkError( + exception = intercept[AnalysisException] { + sql(s"INSERT INTO $t1 WITH ('versionAsOf' = 'v1') VALUES (1, 'a')") + }, + condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL", + parameters = Map("relationId" -> "`testcat`.`ns1`.`ns2`.`table`")) + } + } + + test("SPARK-58389: CACHE TABLE result is not reused for a read carrying different options") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + + // Cache the option-free read. The CacheManager keys entries on the query plan, and a DSv2 + // relation's plan carries its `options`, so the cached entry's fingerprint is "no options". + val cached = spark.table(t1) + cached.cache() + try { + val cacheManager = spark.sharedState.cacheManager + + // An option-free read has the same plan fingerprint, so it reuses the cached result. + assert(cacheManager.lookupCachedData(spark.table(t1)).isDefined, + "an option-free read should hit the cached result") + + // A read carrying options has a different fingerprint, so it must NOT reuse the cached + // result -- otherwise the connector's options would be silently ignored on a cache hit. + assert( + cacheManager.lookupCachedData(spark.read.option("split-size", "5").table(t1)).isEmpty, + "a read carrying options must not reuse the option-free cached result") + } finally { + cached.unpersist() + } + } + } + + test("SPARK-58389: execution refresh does not reuse cached relation with different options") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + + val cached = spark.table(t1) + cached.cache() + try { + assert(cached.count() === 2) + + // Analysis correctly rejects the option-free shared relation cache entry. Reset after + // analysis to isolate the execution-time refresh, which must apply the same options check. + val df = spark.read.option("split-size", "5").table(t1).filter("id > 0") + df.queryExecution.analyzed + inMemoryCatalog.resetLoadTableCalls() + df.collect() + + val refreshLoads = inMemoryCatalog.loadTableCalls + .map(_._2.get("split-size")) + .filter(_ != null) + assert(refreshLoads === Seq("5"), + s"expected refresh to reject the option-free cache entry, got: $refreshLoads") + } finally { + cached.unpersist() + } + } + } + + test("SPARK-58389: recaching preserves and forwards table options") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + + val cached = spark.read.option("split-size", "5").table(t1) + cached.cache() + try { + assert(cached.count() === 2) + + // Refreshing a cached table rebuilds its CacheManager entry. The rebuilt relation must + // retain the original options, and the catalog must use them when it reloads the Table. + inMemoryCatalog.resetLoadTableCalls() + spark.catalog.refreshTable(t1) + + val recacheLoads = inMemoryCatalog.loadTableCalls.map(_._2.get("split-size")) + assert(recacheLoads.contains("5"), + s"expected recache to forward split-size=5, got: $recacheLoads") + + val cacheManager = spark.sharedState.cacheManager + val sameOptions = spark.read.option("split-size", "5").table(t1) + val recached = cacheManager.lookupCachedData(sameOptions) + assert(recached.isDefined, "a read with the original options should hit after recache") + + val recachedOptions = recached.get.plan.collect { + case r: DataSourceV2Relation => r.options.get("split-size") + } + assert(recachedOptions === Seq("5"), + s"expected the recached relation to retain split-size=5, got: $recachedOptions") + + assert(cacheManager.lookupCachedData(spark.table(t1)).isEmpty, + "an option-free read must not reuse the recached option-carrying result") + } finally { + spark.catalog.clearCache() + } + } + } + + test("SPARK-58389: recaching a non-relation plan forwards table options") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + + // The filter forces CacheManager.tryRefreshPlan through V2TableRefreshUtil instead of the + // bare-relation fast path covered by the preceding test. + val cached = spark.read.option("split-size", "5").table(t1).filter("id > 0") + cached.cache() + try { + assert(cached.count() === 2) + inMemoryCatalog.resetLoadTableCalls() + + spark.catalog.refreshTable(t1) + + val recacheLoads = inMemoryCatalog.loadTableCalls + .map(_._2.get("split-size")) + .filter(_ != null) + assert(recacheLoads.contains("5"), + s"expected non-relation recache to forward split-size=5, got: $recacheLoads") + + val samePlan = spark.read.option("split-size", "5").table(t1).filter("id > 0") + assert(spark.sharedState.cacheManager.lookupCachedData(samePlan).isDefined, + "the filtered plan should remain cached after refresh") + } finally { + spark.catalog.clearCache() + } + } + } + + test("SPARK-58389: a DataFrame temp view's options do not leak to a later reference") { + val t1 = s"${catalogAndNamespace}table" + withTable(t1) { + sql(s"CREATE TABLE $t1 (id bigint, data string)") + sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')") + + // The temp view resolves via the V2TableReference path, whose relation carries the view's + // options. A later option-free reference to the same table must not inherit them. + withTempView("v") { + spark.read.option("split-size", "5").table(t1).createOrReplaceTempView("v") + val df = sql(s"SELECT v.id FROM v JOIN $t1 b ON v.id = b.id") + + val splitSizes = df.queryExecution.analyzed.collect { + case r: DataSourceV2Relation => Option(r.options.get("split-size")) + } + // Exactly one reference (`v`) keeps its option; `b` (option-free) must not inherit it. + assert(splitSizes.flatten === Seq("5"), + s"option leaked to the option-free reference, got: $splitSizes") + assert(splitSizes.contains(None), + s"expected an option-free reference, got: $splitSizes") + } + } + } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala index bcd8ba185d1d..406e8885e163 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala @@ -385,6 +385,21 @@ class SupportsCatalogOptionsSuite extends SharedSparkSession with BeforeAndAfter assert(relation.timeTravelSpec.contains(expectedTimeTravelSpec)) } + test("SPARK-58389: read options are forwarded to the catalog's loadTable") { + sql(s"create table $catalogName.t1 (id bigint) using $format") + val cat = catalog(catalogName).asInstanceOf[InMemoryTableCatalog] + cat.resetLoadTableCalls() + + // spark.read.format(...).option(...).load() goes through the SupportsCatalogOptions read + // path in DataSourceV2Utils, which now forwards the user options to CatalogV2Util.getTable. + load("t1", Some(catalogName)).collect() + + val opts = cat.lastLoadTableOptions + assert(opts.isDefined, "loadTable(context, options) was not invoked") + // The user-provided read options reach the catalog (e.g. the table name selector). + assert(opts.get.get("name") === "t1") + } + private def load( name: String, catalogOpt: Option[String], diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala index c88ebb0d69ee..bcc4895616bd 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala @@ -30,12 +30,13 @@ import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull import org.apache.spark.sql.catalyst.parser.CatalystSqlParser import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog, TableWritePrivilege} +import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog, TableContext, TableWritePrivilege} import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform} import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{BooleanType, IntegerType, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap abstract class AlignAssignmentsSuiteBase extends AnalysisTest { @@ -163,6 +164,10 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { }) when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]())) .thenCallRealMethod() + // The options-aware overload runs the real default dispatch, which delegates to the + // stubbed overloads above. + when(newCatalog.loadTable(any(), any[TableContext](), any[CaseInsensitiveStringMap]())) + .thenCallRealMethod() when(newCatalog.name()).thenReturn("cat") newCatalog } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala index a924b637a79f..ee3eaa789ffc 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala @@ -27,7 +27,7 @@ import org.mockito.invocation.InvocationOnMock import org.apache.spark.SparkUnsupportedOperationException import org.apache.spark.sql.{AnalysisException, SaveMode} import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier} -import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, AnalysisTest, Analyzer, AsOfVersion, EmptyFunctionRegistry, NoSuchTableException, RelationResolution, ResolvedFieldName, ResolvedFieldPosition, ResolvedIdentifier, ResolvedTable, ResolveSessionCatalog, TimeTravelSpec, UnresolvedAttribute, UnresolvedFieldPosition, UnresolvedInlineTable, UnresolvedPartitionSpec, UnresolvedRelation, UnresolvedSubqueryColumnAliases, UnresolvedTable} +import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, AnalysisTest, Analyzer, AsOfVersion, EmptyFunctionRegistry, NoSuchTableException, RelationCache, RelationResolution, ResolvedFieldName, ResolvedFieldPosition, ResolvedIdentifier, ResolvedTable, ResolveSessionCatalog, TimeTravelSpec, UnresolvedAttribute, UnresolvedFieldPosition, UnresolvedInlineTable, UnresolvedPartitionSpec, UnresolvedRelation, UnresolvedSubqueryColumnAliases, UnresolvedTable} import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogStorageFormat, CatalogTable, CatalogTableType, InMemoryCatalog, SessionCatalog, TempVariableManager} import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, EqualTo, Expression, InSubquery, IntegerLiteral, ListQuery, Literal, StringLiteral} import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke @@ -36,7 +36,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{AlterColumns, AlterColumnSpe import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId import org.apache.spark.sql.connector.FakeV2Provider -import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, TableChange, TableWritePrivilege, V1Table} +import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, TableChange, TableContext, TableWritePrivilege, V1Table} import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform} import org.apache.spark.sql.errors.QueryExecutionErrors @@ -47,6 +47,7 @@ import org.apache.spark.sql.internal.SQLConf.{PARTITION_OVERWRITE_MODE, Partitio import org.apache.spark.sql.sources.SimpleScanSource import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types.{BooleanType, CharType, DoubleType, IntegerType, LongType, StringType, StructField, StructType, VarcharType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap import org.apache.spark.unsafe.types.UTF8String class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { @@ -190,6 +191,10 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { }) when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]())) .thenCallRealMethod() + // The options-aware overload runs the real default dispatch, which delegates to the + // stubbed overloads above. + when(newCatalog.loadTable(any(), any[TableContext](), any[CaseInsensitiveStringMap]())) + .thenCallRealMethod() when(newCatalog.name()).thenReturn("testcat") newCatalog } @@ -209,6 +214,10 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { }) when(newCatalog.loadTable(any(), any[java.util.Set[TableWritePrivilege]]())) .thenCallRealMethod() + // The options-aware overload runs the real default dispatch, which delegates to the + // stubbed overloads above. + when(newCatalog.loadTable(any(), any[TableContext](), any[CaseInsensitiveStringMap]())) + .thenCallRealMethod() when(newCatalog.name()).thenReturn(CatalogManager.SESSION_CATALOG_NAME) newCatalog } @@ -3398,7 +3407,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { // after first resolution, cache should have 1 entry (without time travel) assert(ctx.relationCache.size == 1) - assert(ctx.relationCache.keys.head._2.isEmpty) + assert(ctx.relationCache.keys.head.timeTravelSpec.isEmpty) // create unresolved relation with time travel spec val timeTravelSpec = AsOfVersion("v1") @@ -3456,6 +3465,54 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { } } + test("SPARK-58389: shared relation cache is reused only when the read's options match") { + val ident = Identifier.of(Array.empty[String], "tab") + val cachedTable = testCat.loadTable(ident) + + // A shared relation cache entry (as if left by an earlier CACHE TABLE) built with a specific + // set of options. The `id` tag lets the test tell a cache reuse apart from a fresh load, since + // both would otherwise carry the same mock `Table`. + def cachedRelationWith(opts: java.util.Map[String, String]): DataSourceV2Relation = { + val r = DataSourceV2Relation.create( + cachedTable, Some(testCat), Some(ident), new CaseInsensitiveStringMap(opts)) + r.setTagValue(LogicalPlan.PLAN_ID_TAG, 4242L) + r + } + + def resolveWith( + cacheOpts: java.util.Map[String, String], + readOpts: java.util.Map[String, String]): DataSourceV2Relation = { + AnalysisContext.withNewAnalysisContext { + val sharedCache: RelationCache = (_, _) => Some(cachedRelationWith(cacheOpts)) + val rule = new RelationResolution(catalogManagerWithDefault, sharedCache) + val unresolved = + UnresolvedRelation(Seq("testcat", "tab"), new CaseInsensitiveStringMap(readOpts)) + rule.resolveRelation(unresolved) match { + case Some(AsDataSourceV2Relation(relation)) => relation + case other => fail(s"failed to resolve as v2 relation: $other") + } + } + } + + // Same options as the cached entry: the cache is reused (the tagged cached relation flows + // through, so the tag survives). + val reused = resolveWith( + java.util.Map.of("split-size", "5"), java.util.Map.of("split-size", "5")) + assert(reused.options.get("split-size") === "5") + assert(reused.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(4242L), + "matching options should reuse the cached relation") + + // Different options: the cache is NOT reused. The relation is freshly loaded with this read's + // options, so it does not carry the cached entry's option or its tag. + val fresh = resolveWith( + java.util.Map.of("cachedOnly", "stale"), java.util.Map.of("split-size", "5")) + assert(fresh.options.get("split-size") === "5") + assert(!fresh.options.containsKey("cachedOnly"), + "differing options must not reuse the cached relation") + assert(fresh.getTagValue(LogicalPlan.PLAN_ID_TAG).isEmpty, + "differing options should freshly load, not reuse the cached relation") + } + private def resolve( unresolvedRelation: UnresolvedRelation, timeTravelSpec: Option[TimeTravelSpec] = None,