You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/09/07 12:51:58 UTC
(spark) branch branch-4.3 updated: [SPARK-59163][SQL] Reuse projected options and make CaseInsensitiveStringMap read-only
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 74197c98452d [SPARK-59163][SQL] Reuse projected options and make CaseInsensitiveStringMap read-only
74197c98452d is described below
commit 74197c98452d677eb90551c93a25e2cd2d3acba4
Author: yyanyy <yy...@gmail.com>
AuthorDate: Mon Sep 7 20:50:58 2026 +0800
[SPARK-59163][SQL] Reuse projected options and make CaseInsensitiveStringMap read-only
### What changes were proposed in this pull request?
This is tracked by [SPARK-59163](https://issues.apache.org/jira/browse/SPARK-59163). It follows
#57585 and completes the table-state option reuse requested in
[this review comment](https://github.com/apache/spark/pull/57585#discussion_r3819899362). The
earlier PR introduced the projection and has already landed; this PR is its subsequent internal
cleanup.
`RelationResolution` already computed the projected table-state options for `TableCacheKey`, but
table loading projected the full option map again. The projected map passed to
`CatalogV2Util.lookupCachedRelation` was also projected a second time inside that method.
This PR:
- as a minor refactoring, adds consistently named `*WithStateOptions` entry points and makes the
existing full-option helpers project once before delegating to them;
- reuses `TableCacheKey.stateOptions` for shared relation cache lookup in both persistent relation
and `V2TableReference` resolution, without repeating the projection for table loading;
- reuses the single projection in `V2TableRefreshUtil` for refresh deduplication and shared cache
lookup, without projecting the full option map again for catalog loading;
- makes `CaseInsensitiveStringMap` consistently read-only, including `keySet`, `values`, and
`entrySet`, so catalog loading can safely consume the same projected instance used by Spark's
cache and deduplication keys.
### Why are the changes needed?
The previous flow repeatedly normalized the catalog's state-option key set and materialized a full
projection during an uncached relation resolution or execution refresh. The new flow reuses each
projection for catalog loading, Spark's table pin or refresh deduplication, and shared cache lookup.
Because that same instance may be retained as a cache or deduplication key, it must remain stable
after being passed to connector code. `CaseInsensitiveStringMap` already rejected direct mutation;
making its normalized map unmodifiable also rejects mutation through collection views. This keeps
the reuse safe without reconstructing the map, preserving its original-case entries and the
effective value already selected for keys that differ only by case.
### Does this PR introduce _any_ user-facing change?
Yes, narrowly for connector implementations. Mutation through `CaseInsensitiveStringMap`'s
collection views now throws `UnsupportedOperationException`, consistent with its existing direct
mutators. Read behavior is unchanged.
### How was this patch tested?
`CaseInsensitiveStringMapSuite`, `CatalogV2UtilSuite`, `DataSourceV2OptionSuite`, and
`PlanResolutionSuite` cover collection-view immutability, duplicate-case preservation, table-state
option filtering, persistent relation resolution, `V2TableReference` resolution, shared relation
cache matching, and execution refresh. Focused call-count assertions verify that already-projected
paths do not project again, and catalog-side mutation assertions verify that retained cache keys
cannot be changed through collection views. Five additional SQL suites cover existing connector
option-map consumers.
```bash
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 \
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/sbt \
'catalyst/testOnly org.apache.spark.sql.util.CaseInsensitiveStringMapSuite org.apache.spark.sql.connector.catalog.CatalogV2UtilSuite' \
'sql/testOnly org.apache.spark.sql.connector.DataSourceV2OptionSuite org.apache.spark.sql.execution.command.PlanResolutionSuite org.apache.spark.sql.connector.V1WriteFallbackSuite org.apache.spark.sql.connector.AppendDataTransactionSuite org.apache.spark.sql.connector.SupportsCatalogOptionsSuite org.apache.spark.sql.execution.datasources.v2.FileTableSuite org.apache.spark.sql.sources.ExternalCommandRunnerSuite'
```
The nine-suite run completed 243/243 tests: `CaseInsensitiveStringMapSuite` 9/9,
`CatalogV2UtilSuite` 21/21, `DataSourceV2OptionSuite` 56/56, `PlanResolutionSuite` 96/96,
`V1WriteFallbackSuite` 13/13, `AppendDataTransactionSuite` 19/19,
`SupportsCatalogOptionsSuite` 20/20, `FileTableSuite` 8/8, and `ExternalCommandRunnerSuite` 1/1.
There were no failures, cancellations, ignores, or pending tests. The run selected these nine suites
but used no test-case filters within them; SBT's incremental compilation cache was used. Several
SQL suites emitted non-failing possible thread-leak warnings.
After the final `asCaseSensitiveMap` entry-mutation assertion was added, the focused suite was
rerun:
```bash
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 \
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/sbt \
'catalyst/testOnly org.apache.spark.sql.util.CaseInsensitiveStringMapSuite'
```
`CaseInsensitiveStringMapSuite` ran 9/9 tests, with no failures, cancellations, ignores, or pending
tests. No test-case filter was used; the run used SBT's incremental compilation cache.
```bash
XDG_RUNTIME_DIR=/tmp/yan-sbt-runtime \
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 \
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/sbt \
'catalyst / Compile / scalastyle' \
'catalyst / Test / scalastyle' \
'sql / Compile / scalastyle' \
'sql / Test / scalastyle' \
'catalyst / Compile / checkstyle'
```
The four scalastyle tasks processed 718 Catalyst main files, 440 Catalyst test files, 826 SQL core
main files, and 1,106 SQL core test files with no errors or warnings. Catalyst main checkstyle also
completed with no issues.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
Closes #58462 from yyanyy/spark-map-lookup-cleanup-20260831.
Authored-by: yyanyy <yy...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 4dc04afa85a97daa1b8e12ee3f222687fd25f82d)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/util/CaseInsensitiveStringMap.java | 9 +--
.../sql/catalyst/analysis/RelationResolution.scala | 18 +++---
.../sql/connector/catalog/CatalogV2Util.scala | 44 ++++++++++++--
.../sql/connector/catalog/CatalogV2UtilSuite.scala | 67 +++++++++++++++++++++-
.../sql/util/CaseInsensitiveStringMapSuite.scala | 29 ++++++++++
.../datasources/v2/V2TableRefreshUtil.scala | 10 ++--
.../sql/connector/DataSourceV2OptionSuite.scala | 50 ++++++++++++++--
.../execution/command/PlanResolutionSuite.scala | 13 ++++-
8 files changed, 209 insertions(+), 31 deletions(-)
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java b/sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java
index 0c85f9b41cba..3a0de51c96b0 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java
@@ -39,7 +39,7 @@ import org.apache.spark.SparkUnsupportedOperationException;
* This is used to pass options to v2 implementations to ensure consistent case insensitivity.
* <p>
* Methods that return keys in this map, like {@link #entrySet()} and {@link #keySet()}, return
- * keys converted to lower case. This map doesn't allow null key.
+ * keys converted to lower case. This map is read-only and does not allow null keys.
*
* @since 3.0.0
*/
@@ -58,15 +58,16 @@ public class CaseInsensitiveStringMap implements Map<String, String> {
public CaseInsensitiveStringMap(Map<String, String> originalMap) {
original = new HashMap<>(originalMap);
- delegate = new HashMap<>(originalMap.size());
+ Map<String, String> normalizedMap = new HashMap<>(originalMap.size());
for (Map.Entry<String, String> entry : originalMap.entrySet()) {
String key = toLowerCase(entry.getKey());
- if (delegate.containsKey(key)) {
+ if (normalizedMap.containsKey(key)) {
logger.warn("Converting duplicated key {} into CaseInsensitiveStringMap.",
MDC.of(LogKeys.KEY, entry.getKey()));
}
- delegate.put(key, entry.getValue());
+ normalizedMap.put(key, entry.getValue());
}
+ delegate = Collections.unmodifiableMap(normalizedMap);
}
@Override
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 fcbec9db78c9..89e5db467c9e 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
@@ -286,8 +286,8 @@ class RelationResolution(
// time-travel and write-privilege loads cannot.
//
// Skip the table-side lookup entirely for view-only catalogs (no `TableCatalog`
- // mixin): `CatalogV2Util.loadTable` would call `asTableCatalog` and throw
- // MISSING_CATALOG_ABILITY.TABLES, masking the legitimate view-resolution path.
+ // mixin): `CatalogV2Util.loadTableWithStateOptions` would call `asTableCatalog` and
+ // throw MISSING_CATALOG_ABILITY.TABLES, masking the legitimate view-resolution path.
val relation: Option[Relation] = pinnedTable.orElse {
catalog match {
case mc: RelationCatalog
@@ -302,12 +302,12 @@ class RelationResolution(
val tableSide: Option[Table] = if (
CatalogV2Util.isSessionCatalog(catalog) || catalog.isInstanceOf[TableCatalog]
) {
- CatalogV2Util.loadTable(
+ CatalogV2Util.loadTableWithStateOptions(
catalog,
ident,
+ tableKey.stateOptions,
finalTimeTravelSpec,
- Option(writePrivileges),
- finalOptions)
+ Option(writePrivileges))
} else {
None
}
@@ -406,8 +406,9 @@ class RelationResolution(
catalog: CatalogPlugin,
ident: Identifier,
table: Table,
- options: CaseInsensitiveStringMap): Option[DataSourceV2Relation] = {
- CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident, table, options, conf)
+ stateOptions: CaseInsensitiveStringMap): Option[DataSourceV2Relation] = {
+ CatalogV2Util.lookupCachedRelationWithStateOptions(
+ sharedRelationCache, catalog, ident, table, stateOptions, conf)
}
private def adaptCachedRelation(cached: LogicalPlan, planId: Option[Long]): LogicalPlan = {
@@ -534,7 +535,8 @@ class RelationResolution(
case Some(pinnedTable) =>
createRelation(ref, catalog, pinnedTable)
case None =>
- val table = CatalogV2Util.getTable(catalog, ref.identifier, options = ref.options)
+ val table = CatalogV2Util.getTableWithStateOptions(
+ catalog, ref.identifier, tableKey.stateOptions)
val sharedCacheMatch = if (ref.context.sharedCacheable) {
lookupSharedRelationCache(catalog, ref.identifier, table, tableKey.stateOptions)
} else {
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 9660ede92121..39dc1b853e05 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
@@ -475,9 +475,24 @@ private[sql] object CatalogV2Util {
ident: Identifier,
timeTravelSpec: Option[TimeTravelSpec] = None,
writePrivilegesString: Option[String] = None,
- options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Option[Table] =
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Option[Table] = {
+ val stateOptions = extractTableStateOptions(catalog, options)
+ loadTableWithStateOptions(
+ catalog, ident, stateOptions, timeTravelSpec, writePrivilegesString)
+ }
+
+ /**
+ * Loads a table using table-state options already projected by the caller.
+ */
+ def loadTableWithStateOptions(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ stateOptions: CaseInsensitiveStringMap,
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ writePrivilegesString: Option[String] = None): Option[Table] =
try {
- Option(getTable(catalog, ident, timeTravelSpec, writePrivilegesString, options))
+ Option(getTableWithStateOptions(
+ catalog, ident, stateOptions, timeTravelSpec, writePrivilegesString))
} catch {
case _: NoSuchTableException => None
case _: NoSuchDatabaseException => None
@@ -510,13 +525,26 @@ private[sql] object CatalogV2Util {
timeTravelSpec: Option[TimeTravelSpec] = None,
writePrivilegesString: Option[String] = None,
options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()): Table = {
+ val stateOptions = extractTableStateOptions(catalog, options)
+ getTableWithStateOptions(
+ catalog, ident, stateOptions, timeTravelSpec, writePrivilegesString)
+ }
+
+ /**
+ * Loads a table using table-state options already projected by the caller.
+ */
+ def getTableWithStateOptions(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ stateOptions: CaseInsensitiveStringMap,
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ writePrivilegesString: Option[String] = None): 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))
- val stateOptions = extractTableStateOptions(catalog, options)
catalog.asTableCatalog.loadTable(ident, context, stateOptions)
}
@@ -604,18 +632,22 @@ private[sql] object CatalogV2Util {
loadTable(catalog, ident).map(DataSourceV2Relation.create(_, Some(catalog), Some(ident)))
}
- def lookupCachedRelation(
+ /**
+ * Looks up a cached relation using table-state options already projected by the caller.
+ * Reusing the projection keeps table pinning and shared-cache lookup on the same state key.
+ */
+ def lookupCachedRelationWithStateOptions(
cache: RelationCache,
catalog: CatalogPlugin,
ident: Identifier,
table: Table,
- options: CaseInsensitiveStringMap,
+ stateOptions: CaseInsensitiveStringMap,
conf: SQLConf): Option[DataSourceV2Relation] = {
cache.lookup(
catalog,
ident,
Some(table.id),
- extractTableStateOptions(catalog, options),
+ stateOptions,
conf.resolver).collect { case r: DataSourceV2Relation => r }
}
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 e128adebdff8..7c2de5e69890 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,14 +17,17 @@
package org.apache.spark.sql.connector.catalog
+import scala.collection.mutable
+
import org.mockito.ArgumentCaptor
import org.mockito.ArgumentMatchers.{any, eq => mockEq}
import org.mockito.Mockito.{mock, verify, when}
+import org.mockito.invocation.InvocationOnMock
import org.apache.spark.{SparkFunSuite, SparkIllegalArgumentException}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.{
- AsOfTimestamp, AsOfVersion, TimeTravelSpec, UnresolvedRelation}
+ AsOfTimestamp, AsOfVersion, TableCacheKey, TimeTravelSpec, UnresolvedRelation}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StructType}
@@ -197,6 +200,68 @@ class CatalogV2UtilSuite extends SparkFunSuite {
mockEq(expected))
}
+ test("getTableWithStateOptions rejects catalog mutation of a retained cache key") {
+ val catalog = mock(classOf[TableCatalog])
+ val ident = Identifier.of(Array("ns"), "table")
+ val table = mock(classOf[Table])
+ var catalogOptions: CaseInsensitiveStringMap = null
+ var mutationBlocked = false
+ when(catalog.loadTable(any[Identifier], any[TableContext], any[CaseInsensitiveStringMap]))
+ .thenAnswer((invocation: InvocationOnMock) => {
+ catalogOptions = invocation.getArgument[CaseInsensitiveStringMap](2)
+ intercept[UnsupportedOperationException] {
+ catalogOptions.entrySet().iterator().next().setValue("s2")
+ }
+ mutationBlocked = true
+ table
+ })
+ val stateOptions =
+ new CaseInsensitiveStringMap(java.util.Map.of("SnApShOt", "s1"))
+ val retainedKey = TableCacheKey(catalog, ident, None, stateOptions)
+
+ val loaded = CatalogV2Util.getTableWithStateOptions(catalog, ident, stateOptions)
+ val tableCache = mutable.Map(retainedKey -> loaded)
+ val sameStateKey = TableCacheKey(
+ catalog,
+ ident,
+ None,
+ new CaseInsensitiveStringMap(java.util.Map.of("snapshot", "s1")))
+ val differentStateKey = TableCacheKey(
+ catalog,
+ ident,
+ None,
+ new CaseInsensitiveStringMap(java.util.Map.of("snapshot", "s2")))
+
+ assert(mutationBlocked, "the catalog fixture did not attempt to mutate its map")
+ assert(catalogOptions.get("snapshot") == "s1")
+ assert(catalogOptions.asCaseSensitiveMap().containsKey("SnApShOt"))
+ assert(tableCache.get(sameStateKey).contains(table))
+ assert(!tableCache.contains(differentStateKey))
+ }
+
+ test("getTableWithStateOptions preserves effective values for duplicate-case keys") {
+ val catalog = mock(classOf[TableCatalog])
+ val ident = Identifier.of(Array("ns"), "table")
+ val table = mock(classOf[Table])
+ var catalogOptions: CaseInsensitiveStringMap = null
+ when(catalog.loadTable(any[Identifier], any[TableContext], any[CaseInsensitiveStringMap]))
+ .thenAnswer((invocation: InvocationOnMock) => {
+ catalogOptions = invocation.getArgument[CaseInsensitiveStringMap](2)
+ table
+ })
+ val options = new java.util.LinkedHashMap[String, String]()
+ options.put("snapshot", "lower-value")
+ options.put("SNAPSHOT", "upper-value")
+ val stateOptions = new CaseInsensitiveStringMap(options)
+
+ CatalogV2Util.getTableWithStateOptions(catalog, ident, stateOptions)
+
+ assert(stateOptions.asCaseSensitiveMap().size() == 2)
+ assert(stateOptions.get("snapshot") == "upper-value")
+ assert(catalogOptions.get("snapshot") == stateOptions.get("snapshot"))
+ assert(catalogOptions.asCaseSensitiveMap() == stateOptions.asCaseSensitiveMap())
+ }
+
test("getTable forwards no options when a catalog declares no table-state options") {
val catalog = mock(classOf[TableCatalog])
when(catalog.tableStateOptionKeys()).thenCallRealMethod()
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/util/CaseInsensitiveStringMapSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/util/CaseInsensitiveStringMapSuite.scala
index 932fb0a73337..fb5e39861fb5 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/util/CaseInsensitiveStringMapSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/util/CaseInsensitiveStringMapSuite.scala
@@ -45,6 +45,31 @@ class CaseInsensitiveStringMapSuite extends SparkFunSuite {
assert(options.values().asScala.toSeq == Seq("valUE"))
}
+ test("collection views are read-only") {
+ def checkMutationBlocked(mutate: CaseInsensitiveStringMap => Any): Unit = {
+ val options = new CaseInsensitiveStringMap(Map("kEy" -> "valUE").asJava)
+
+ intercept[UnsupportedOperationException] {
+ mutate(options)
+ }
+
+ assert(options.get("key") == "valUE")
+ assert(options.asCaseSensitiveMap().asScala == Map("kEy" -> "valUE"))
+ }
+
+ checkMutationBlocked(_.keySet().remove("key"))
+ checkMutationBlocked(_.values().remove("valUE"))
+ checkMutationBlocked(_.entrySet().iterator().next().setValue("newValue"))
+ checkMutationBlocked { options =>
+ val iterator = options.entrySet().iterator()
+ iterator.next()
+ iterator.remove()
+ }
+ checkMutationBlocked { options =>
+ options.replaceAll((_, _) => "newValue")
+ }
+ }
+
test("getInt") {
val options = new CaseInsensitiveStringMap(Map("numFOo" -> "1", "foo" -> "bar").asJava)
assert(options.getInt("numFOO", 10) == 1)
@@ -107,5 +132,9 @@ class CaseInsensitiveStringMapSuite extends SparkFunSuite {
intercept[UnsupportedOperationException] {
caseSensitiveMap.put("kEy", "valUE")
}
+ intercept[UnsupportedOperationException] {
+ caseSensitiveMap.entrySet().iterator().next().setValue("newValue")
+ }
+ assert(caseSensitiveMap.equals(originalMap))
}
}
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 6653a10c87c1..da5f930795e7 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
@@ -91,13 +91,13 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper with Logging {
val stateOptions = CatalogV2Util.extractTableStateOptions(catalog, r.options)
val currentTable = currentTables.getOrElseUpdate((catalog, ident, stateOptions), {
val tableName = V2TableUtil.toQualifiedName(catalog, ident)
- lookupCachedRelation(spark, catalog, ident, r.table, r.options) match {
+ lookupCachedRelation(spark, catalog, ident, r.table, stateOptions) match {
case Some(cached) =>
logDebug(s"Refreshing table metadata for $tableName using shared relation cache")
cached.table
case _ =>
logDebug(s"Refreshing table metadata for $tableName using catalog")
- CatalogV2Util.getTable(catalog, ident, options = r.options)
+ CatalogV2Util.getTableWithStateOptions(catalog, ident, stateOptions)
}
})
validateTableIdentity(currentTable, r)
@@ -112,9 +112,9 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper with Logging {
catalog: TableCatalog,
ident: Identifier,
table: Table,
- options: CaseInsensitiveStringMap): Option[DataSourceV2Relation] = {
- CatalogV2Util.lookupCachedRelation(
- spark.sharedState.relationCache, catalog, ident, table, options, conf)
+ stateOptions: CaseInsensitiveStringMap): Option[DataSourceV2Relation] = {
+ CatalogV2Util.lookupCachedRelationWithStateOptions(
+ spark.sharedState.relationCache, catalog, ident, table, stateOptions, conf)
}
// it is not safe to allow any schema changes in commands (e.g. CTAS, RTAS, MERGE)
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 0c9032fbeb00..8b7fd2bb80e3 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
@@ -71,10 +71,17 @@ class LoadCountingInMemoryCatalog extends InMemoryCatalog {
}
class StateAwareInMemoryCatalog extends LoadCountingInMemoryCatalog {
+ private val _tableStateOptionKeyCalls = new AtomicInteger(0)
+
+ def tableStateOptionKeyCalls: Int = _tableStateOptionKeyCalls.get()
+ def resetTableStateOptionKeyCalls(): Unit = _tableStateOptionKeyCalls.set(0)
+
// Include Spark's internal marker so the write-context test detects if it leaks before state
// option projection. Production catalogs should declare only raw user option keys.
- override def tableStateOptionKeys(): util.Set[String] =
+ override def tableStateOptionKeys(): util.Set[String] = {
+ _tableStateOptionKeyCalls.incrementAndGet()
util.Set.of("snapshot", UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
+ }
}
class NullReturningInMemoryCatalog extends InMemoryCatalog {
@@ -142,8 +149,12 @@ class StateAwareV2InMemoryRelationCatalog extends V2InMemoryRelationCatalog {
private val _loadTableCalls =
mutable.ArrayBuffer.empty[(Identifier, TableContext, CaseInsensitiveStringMap)]
private val _loadViewCalls = mutable.ArrayBuffer.empty[Identifier]
+ private val _tableStateOptionKeyCalls = new AtomicInteger(0)
- override def tableStateOptionKeys(): util.Set[String] = util.Set.of("snapshot")
+ override def tableStateOptionKeys(): util.Set[String] = {
+ _tableStateOptionKeyCalls.incrementAndGet()
+ util.Set.of("snapshot")
+ }
override def loadTable(
ident: Identifier,
@@ -168,11 +179,13 @@ class StateAwareV2InMemoryRelationCatalog extends V2InMemoryRelationCatalog {
def loadTableCalls: Seq[(Identifier, TableContext, CaseInsensitiveStringMap)] =
_loadTableCalls.toSeq
def loadViewCalls: Seq[Identifier] = _loadViewCalls.toSeq
+ def tableStateOptionKeyCalls: Int = _tableStateOptionKeyCalls.get()
def resetDispatchCalls(): Unit = {
resetLoadRelationCalls()
_loadTableCalls.clear()
_loadViewCalls.clear()
+ _tableStateOptionKeyCalls.set(0)
}
}
@@ -1032,13 +1045,24 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
relCatalog.resetDispatchCalls()
- spark.read
+ val df = spark.read
.option("snapshot", "s1")
.option("split-size", "5")
.table(t1)
- .collect()
+ df.queryExecution.analyzed
+
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(
+ relCatalog.tableStateOptionKeyCalls == 1,
+ s"expected one analysis projection, got: ${relCatalog.tableStateOptionKeyCalls}")
- assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 2)
+ relCatalog.resetDispatchCalls()
+ df.collect()
+
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(
+ relCatalog.tableStateOptionKeyCalls == 1,
+ s"expected one refresh projection, got: ${relCatalog.tableStateOptionKeyCalls}")
assert(relCatalog.loadRelationCalls.isEmpty)
}
}
@@ -1703,12 +1727,28 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
stateCatalog.resetLoadTableCalls()
stateCatalog.singleArgLoads.set(0)
AnalysisContext.withNewAnalysisContext {
+ stateCatalog.resetTableStateOptionKeyCalls()
val initialRelation =
resolver.resolveReference(initialRef).asInstanceOf[DataSourceV2Relation]
+ assert(
+ stateCatalog.tableStateOptionKeyCalls == 1,
+ s"expected one initial projection, got: ${stateCatalog.tableStateOptionKeyCalls}")
+
+ stateCatalog.resetTableStateOptionKeyCalls()
val cachedRelation =
resolver.resolveReference(initialRef).asInstanceOf[DataSourceV2Relation]
+ assert(
+ stateCatalog.tableStateOptionKeyCalls == 0,
+ s"expected no projection on a relation-cache hit, got: " +
+ stateCatalog.tableStateOptionKeyCalls)
+
+ stateCatalog.resetTableStateOptionKeyCalls()
val otherOptionsRelation =
resolver.resolveReference(otherOptionsRef).asInstanceOf[DataSourceV2Relation]
+ assert(
+ stateCatalog.tableStateOptionKeyCalls == 1,
+ s"expected one table-cache lookup projection, got: " +
+ stateCatalog.tableStateOptionKeyCalls)
assert(initialRelation.table eq cached.table)
assert(cachedRelation.table eq cached.table)
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 1065c5baa585..f960507b300e 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
@@ -3583,7 +3583,11 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
val cachedTable = newTable("table-id")
val catalog = mock(classOf[TableCatalog])
when(catalog.name()).thenReturn("statecat")
- when(catalog.tableStateOptionKeys()).thenReturn(java.util.Set.of("state"))
+ var tableStateOptionKeyCalls = 0
+ when(catalog.tableStateOptionKeys()).thenAnswer((_: InvocationOnMock) => {
+ tableStateOptionKeyCalls += 1
+ java.util.Set.of("state")
+ })
var loads = 0
when(catalog.loadTable(
any[Identifier],
@@ -3608,11 +3612,16 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest {
val resolver = new RelationResolution(manager, sharedRelationCache)
def resolveWith(splitSize: String): DataSourceV2Relation = {
+ tableStateOptionKeyCalls = 0
val unresolved = UnresolvedRelation(Seq("statecat", "tab"), options(splitSize))
- resolver.resolveRelation(unresolved) match {
+ val resolved = resolver.resolveRelation(unresolved) match {
case Some(AsDataSourceV2Relation(relation)) => relation
case other => fail(s"failed to resolve as v2 relation: $other")
}
+ assert(
+ tableStateOptionKeyCalls == 1,
+ s"expected one table-state option projection, got: $tableStateOptionKeyCalls")
+ resolved
}
AnalysisContext.withNewAnalysisContext {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org