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/08/28 11:00:06 UTC
(spark) branch branch-4.3 updated: [SPARK-58392][SQL] Use loadTable for RelationCatalog reads with table-state options
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 675fc9836c15 [SPARK-58392][SQL] Use loadTable for RelationCatalog reads with table-state options
675fc9836c15 is described below
commit 675fc9836c1504ed51669c04b0833e4b33a81ecb
Author: yyanyy <yy...@gmail.com>
AuthorDate: Fri Aug 28 18:58:46 2026 +0800
[SPARK-58392][SQL] Use loadTable for RelationCatalog reads with table-state options
### What changes were proposed in this pull request?
This makes catalog-declared table-state options reach table loading without defining an
options-aware `RelationCatalog` API before relation-state semantics are complete.
The resolver now:
- uses the single-RPC `RelationCatalog.loadRelation(Identifier)` fast path only when the projected
table-state options are empty and there is no time travel or write privilege request;
- sends non-empty table-state options through
`TableCatalog.loadTable(Identifier, TableContext, CaseInsensitiveStringMap)`; and
- falls back to the existing option-less `ViewCatalog.loadView` path when the state-aware table
lookup reports that no table exists.
The proposed `RelationCatalog.loadRelation(Identifier, CaseInsensitiveStringMap)` overload and its
state-aware `loadTable` default are removed. The complete user option map still remains on the
resolved relation for scan or write planning, while only `tableStateOptionKeys()` are used to load
and pin table state. Existing time-travel and write-privilege dispatch remains table-only.
### Why are the changes needed?
Table state belongs to `TableCatalog`, but a relation-level state API would also need defined
semantics for View-definition state and option separation. Finalizing that API now could make it
difficult to add independent Table and View state projections later.
Keeping state-bearing reads on the existing state-aware `loadTable` API preserves table pinning
and option filtering without freezing a premature public API. The no-state path retains
`RelationCatalog`'s single-RPC optimization; state-bearing View reads accept the additional table
lookup before falling back to `loadView`.
### Does this PR introduce _any_ user-facing change?
There is no SQL syntax or configuration change. For connector implementors, catalogs that declare
table-state option keys consume them through the existing state-aware `TableCatalog.loadTable`
overload. `RelationCatalog` does not add an options-aware overload in this PR.
### How was this patch tested?
`DataSourceV2OptionSuite` covers state-aware `loadTable` dispatch, the no-state `loadRelation` fast
path, option-bearing View fallback, View-body option isolation, same-state sharing and
different-state table pinning, execution refresh, recache, `SupportsCatalogOptions`, and existing
time-travel/write-privilege dispatch.
Commands run with the Databricks Maven proxy:
```bash
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/mvn -pl sql/catalyst -DskipTests clean install
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/mvn -pl sql/core -Dtest=none \
-DwildcardSuites=org.apache.spark.sql.connector.DataSourceV2OptionSuite test
```
The Catalyst build succeeded with tests skipped. The focused wildcard suite ran 56/56 tests with
0 failures, ignored, canceled, or pending tests. `-Dtest=none` excluded JUnit tests; no ScalaTest
suite other than `DataSourceV2OptionSuite` was requested.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code Opus 4.8 and OpenAI Codex
Closes #57585 from yyanyy/spark-dsv2-loadrelation-options-20260727.
Authored-by: yyanyy <yy...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 6f568b2f567aea60ade054cca5badcef1d8d0239)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/catalyst/analysis/RelationResolution.scala | 19 +-
.../sql/connector/DataSourceV2OptionSuite.scala | 361 ++++++++++++++++++++-
2 files changed, 371 insertions(+), 9 deletions(-)
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 9a61138cd180..fcbec9db78c9 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
@@ -277,11 +277,13 @@ class RelationResolution(
None
}
- // For a `RelationCatalog` with no time-travel / write privileges, the single-RPC
- // `loadRelation` answers both "is there a table?" and "is there a view?" in one
- // call. Time-travel and write privileges apply to tables only, so for those the
- // lookup falls through to the table-only `loadTable` path below; views are not
- // reachable via the v2 fallback in those cases.
+ // When this lookup through a `RelationCatalog` has no table-state options, time travel,
+ // or write privileges, the single-RPC `loadRelation` answers both "is there a table?"
+ // and "is there a view?" in one call. Table-state options, time travel, and write
+ // privileges apply to tables only, so for those the lookup falls through to the
+ // table-only `loadTable` path below. A state-aware table miss can still fall back to
+ // `loadView`;
+ // 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
@@ -289,7 +291,8 @@ class RelationResolution(
val relation: Option[Relation] = pinnedTable.orElse {
catalog match {
case mc: RelationCatalog
- if finalTimeTravelSpec.isEmpty && writePrivileges == null =>
+ if tableKey.stateOptions.isEmpty &&
+ finalTimeTravelSpec.isEmpty && writePrivileges == null =>
try {
Some(mc.loadRelation(ident))
} catch {
@@ -339,7 +342,7 @@ class RelationResolution(
t <- table
if pinnedTable.isEmpty && finalTimeTravelSpec.isEmpty &&
writePrivileges == null && !u.isStreaming
- cached <- lookupSharedRelationCache(catalog, ident, t, finalOptions)
+ cached <- lookupSharedRelationCache(catalog, ident, t, tableKey.stateOptions)
} yield {
val updatedRelation = cached.copy(options = finalOptions)
updatedRelation.copyTagsFrom(cached)
@@ -533,7 +536,7 @@ class RelationResolution(
case None =>
val table = CatalogV2Util.getTable(catalog, ref.identifier, options = ref.options)
val sharedCacheMatch = if (ref.context.sharedCacheable) {
- lookupSharedRelationCache(catalog, ref.identifier, table, ref.options)
+ lookupSharedRelationCache(catalog, ref.identifier, table, tableKey.stateOptions)
} 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 1490871e4440..0c9032fbeb00 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
@@ -20,6 +20,7 @@ package org.apache.spark.sql.connector
import java.util
import java.util.concurrent.atomic.AtomicInteger
+import scala.collection.mutable
import scala.jdk.CollectionConverters._
import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
@@ -27,6 +28,7 @@ import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.analysis.{
AnalysisContext,
AsOfVersion,
+ NoSuchTableException,
RelationCache,
RelationResolution,
UnresolvedRelation,
@@ -34,18 +36,24 @@ import org.apache.spark.sql.catalyst.analysis.{
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
import org.apache.spark.sql.connector.catalog.{
+ DelegatingTable,
Identifier,
InMemoryBaseTable,
InMemoryCatalog,
+ InMemoryRelationCatalog,
InMemoryRowLevelOperationTableCatalog,
+ InMemoryTable,
InMemoryTableCatalog,
+ Relation,
StagedTable,
StagingTableCatalog,
Table,
TableChange,
+ TableContext,
TableInfo,
TableWritePrivilege,
- TimeTravel}
+ TimeTravel,
+ View}
import org.apache.spark.sql.connector.write.Write
import org.apache.spark.sql.execution.{CommandResultExec, QueryExecution, SparkPlan}
import org.apache.spark.sql.execution.datasources.v2._
@@ -97,6 +105,102 @@ class NullReturningStagingInMemoryCatalog extends InMemoryCatalog with StagingTa
}
}
+class V2InMemoryRelationCatalog extends InMemoryRelationCatalog {
+ private val v2Tables = mutable.HashMap.empty[(Identifier, Option[String]), InMemoryTable]
+ private val _loadRelationCalls = mutable.ArrayBuffer.empty[Identifier]
+
+ protected def loadV2Relation(ident: Identifier, state: Option[String]): Relation = {
+ super.loadRelation(ident) match {
+ case table: DelegatingTable =>
+ v2Tables.getOrElseUpdate(
+ (ident, state),
+ new InMemoryTable(
+ table.name(),
+ table.columns(),
+ table.partitioning(),
+ table.properties(),
+ table.constraints()))
+ case relation => relation
+ }
+ }
+
+ override def loadRelation(ident: Identifier): Relation = {
+ _loadRelationCalls += ident
+ loadV2Relation(ident, None)
+ }
+
+ def loadRelationCalls: Seq[Identifier] = _loadRelationCalls.toSeq
+ def resetLoadRelationCalls(): Unit = _loadRelationCalls.clear()
+
+ override def dropTable(ident: Identifier): Boolean = {
+ v2Tables.keys.filter(_._1 == ident).toSeq.foreach(v2Tables.remove)
+ super.dropTable(ident)
+ }
+}
+
+class StateAwareV2InMemoryRelationCatalog extends V2InMemoryRelationCatalog {
+ private val _loadTableCalls =
+ mutable.ArrayBuffer.empty[(Identifier, TableContext, CaseInsensitiveStringMap)]
+ private val _loadViewCalls = mutable.ArrayBuffer.empty[Identifier]
+
+ override def tableStateOptionKeys(): util.Set[String] = util.Set.of("snapshot")
+
+ override def loadTable(
+ ident: Identifier,
+ context: TableContext,
+ stateOptions: CaseInsensitiveStringMap): Table = {
+ _loadTableCalls += ((ident, context, stateOptions))
+ if (context.timeTravel().isPresent || !context.writePrivileges().isEmpty) {
+ super.loadTable(ident, context, stateOptions)
+ } else {
+ loadV2Relation(ident, Option(stateOptions.get("snapshot"))) match {
+ case table: Table => table
+ case _ => throw new NoSuchTableException(ident)
+ }
+ }
+ }
+
+ override def loadView(ident: Identifier): View = {
+ _loadViewCalls += ident
+ super.loadView(ident)
+ }
+
+ def loadTableCalls: Seq[(Identifier, TableContext, CaseInsensitiveStringMap)] =
+ _loadTableCalls.toSeq
+ def loadViewCalls: Seq[Identifier] = _loadViewCalls.toSeq
+
+ def resetDispatchCalls(): Unit = {
+ resetLoadRelationCalls()
+ _loadTableCalls.clear()
+ _loadViewCalls.clear()
+ }
+}
+
+class DispatchTrackingRelationCatalog extends InMemoryRelationCatalog {
+ private var _loadedVersion: Option[String] = None
+ private var _loadedWritePrivileges: Option[util.Set[TableWritePrivilege]] = None
+
+ def loadedVersion: Option[String] = _loadedVersion
+ def loadedWritePrivileges: Option[util.Set[TableWritePrivilege]] = _loadedWritePrivileges
+
+ def resetDispatchCalls(): Unit = {
+ _loadedVersion = None
+ _loadedWritePrivileges = None
+ }
+
+ override def loadTable(ident: Identifier, version: String): Table = {
+ _loadedVersion = Some(version)
+ loadTable(ident)
+ }
+
+ override def loadTable(
+ ident: Identifier,
+ writePrivileges: util.Set[TableWritePrivilege]): Table = {
+ _loadedWritePrivileges = Some(writePrivileges)
+ loadTable(ident)
+ }
+}
+
class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
import testImplicits._
@@ -129,6 +233,18 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}, s"expected only snapshot=$expectedSnapshot to be forwarded, got: $loadOptions")
}
+ private def assertOnlySnapshotTableOptions(
+ catalog: StateAwareV2InMemoryRelationCatalog,
+ expectedSnapshot: String,
+ expectedCalls: Int): Unit = {
+ val loadOptions = catalog.loadTableCalls.map(_._3)
+ assert(loadOptions.size === expectedCalls,
+ s"expected $expectedCalls state-aware table loads, got: $loadOptions")
+ assert(loadOptions.forall { options =>
+ options.size() == 1 && options.get("snapshot") == expectedSnapshot
+ }, s"expected only snapshot=$expectedSnapshot to be forwarded, got: $loadOptions")
+ }
+
private val loadOption = "load-Option"
private val loadOptionValue = "load-value"
private val writeOption = "write-option"
@@ -789,6 +905,249 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
+ test("SPARK-58392: RelationCatalog uses loadTable only for non-empty table state") {
+ registerCatalog("testrelcat", classOf[StateAwareV2InMemoryRelationCatalog])
+ val t1 = "testrelcat.ns1.ns2.table"
+ val ident = Identifier.of(Array("ns1", "ns2"), "table")
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+
+ val relCatalog =
+ catalog("testrelcat").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+ relCatalog.resetDispatchCalls()
+ val df = spark.read
+ .option("SnApShOt", "s1")
+ .option("split-size", "5")
+ .table(t1)
+ val relations = df.queryExecution.analyzed.collect { case r: DataSourceV2Relation => r }
+
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(relCatalog.loadRelationCalls.isEmpty)
+ assert(relations.size === 1)
+ assert(relations.head.options.size() === 2)
+ assert(relations.head.options.get("snapshot") === "s1")
+ assert(relations.head.options.get("split-size") === "5")
+
+ relCatalog.resetDispatchCalls()
+ spark.table(t1).queryExecution.analyzed
+ assert(relCatalog.loadTableCalls.isEmpty)
+ assert(relCatalog.loadRelationCalls === Seq(ident))
+ }
+ }
+
+ test("SPARK-58392: option-bearing View uses loadTable then option-less loadView") {
+ registerCatalog("testrelcat", classOf[StateAwareV2InMemoryRelationCatalog])
+ val v1 = "testrelcat.ns1.ns2.view"
+ val ident = Identifier.of(Array("ns1", "ns2"), "view")
+ withView(v1) {
+ sql(s"CREATE VIEW $v1 AS SELECT 1 AS x")
+
+ val relCatalog =
+ catalog("testrelcat").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+ relCatalog.resetDispatchCalls()
+ spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(v1)
+ .queryExecution
+ .analyzed
+
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(relCatalog.loadViewCalls === Seq(ident))
+ assert(relCatalog.loadRelationCalls === Seq(ident))
+ }
+ }
+
+ test("SPARK-58392: view and nested table keep independent state options") {
+ registerCatalog("nestedviewrel", classOf[StateAwareV2InMemoryRelationCatalog])
+ val t1 = "nestedviewrel.ns1.ns2.table"
+ val v1 = "nestedviewrel.ns1.ns2.view"
+ withTable(t1) {
+ withView(v1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+ sql(s"CREATE VIEW $v1 AS SELECT * FROM $t1 WITH ('snapshot' = 'inner')")
+
+ val relCatalog =
+ catalog("nestedviewrel").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+ relCatalog.resetDispatchCalls()
+ spark.read
+ .option("snapshot", "outer")
+ .option("split-size", "5")
+ .table(v1)
+ .collect()
+
+ val calls = relCatalog.loadTableCalls.map(_._3)
+ assert(calls.map(_.get("snapshot")) === Seq("outer", "inner", "inner"))
+ assert(calls.forall(_.size() === 1))
+ val viewIdent = Identifier.of(Array("ns1", "ns2"), "view")
+ assert(relCatalog.loadViewCalls === Seq(viewIdent))
+ assert(relCatalog.loadRelationCalls === Seq(viewIdent))
+ }
+ }
+ }
+
+ test("SPARK-58392: RelationCatalog table pins use the state-option projection") {
+ registerCatalog("pinningrel", classOf[StateAwareV2InMemoryRelationCatalog])
+ val t1 = "pinningrel.ns1.ns2.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+ val relCatalog =
+ catalog("pinningrel").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+
+ relCatalog.resetDispatchCalls()
+ val sameState = sql(s"SELECT a.id FROM $t1 " +
+ s"WITH (`snapshot` = 's1', `split-size` = 5) a JOIN $t1 " +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) b ON a.id = b.id")
+ val sameStateRelations = sameState.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("split-size") => r
+ }
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(sameStateRelations.size === 2)
+ assert(sameStateRelations.map(_.options.get("split-size")).sorted === Seq("5", "9"))
+ assert(sameStateRelations.head.table eq sameStateRelations.last.table)
+ assert(relCatalog.loadRelationCalls.isEmpty)
+
+ relCatalog.resetDispatchCalls()
+ val differentStates = sql(s"SELECT a.id FROM $t1 " +
+ s"WITH (`snapshot` = 's1') a JOIN $t1 " +
+ s"WITH (`snapshot` = 's2') b ON a.id = b.id")
+ val differentStateRelations = differentStates.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("snapshot") => r
+ }
+ assert(differentStateRelations.size === 2)
+ assert(differentStateRelations.head.table ne differentStateRelations.last.table)
+ val loadedStates = relCatalog.loadTableCalls.map(_._3)
+ assert(loadedStates.map(_.get("snapshot")).sorted === Seq("s1", "s2"))
+ assert(loadedStates.forall(_.size() === 1))
+ assert(relCatalog.loadRelationCalls.isEmpty)
+ }
+ }
+
+ test("SPARK-58392: execution refresh forwards only table-state options") {
+ registerCatalog("loadcountingrel", classOf[StateAwareV2InMemoryRelationCatalog])
+ val relCatalog =
+ catalog("loadcountingrel").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+ val t1 = "loadcountingrel.ns1.ns2.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+ relCatalog.resetDispatchCalls()
+
+ spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(t1)
+ .collect()
+
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 2)
+ assert(relCatalog.loadRelationCalls.isEmpty)
+ }
+ }
+
+ test("SPARK-58392: recache forwards RelationCatalog table-state options") {
+ registerCatalog("recacherel", classOf[StateAwareV2InMemoryRelationCatalog])
+ val relCatalog = catalog("recacherel").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+ val t1 = "recacherel.ns1.ns2.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+ val cached = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(t1)
+ .filter("id >= 0")
+ cached.cache()
+ try {
+ cached.collect()
+ relCatalog.resetDispatchCalls()
+
+ spark.catalog.refreshTable(t1)
+
+ val recacheOptions = relCatalog.loadTableCalls.map(_._3)
+ assert(recacheOptions.nonEmpty, "expected recache to reload the table")
+ assert(
+ recacheOptions.forall { options =>
+ options.size() == 1 && options.get("snapshot") == "s1"
+ },
+ s"expected recache to forward only snapshot=s1, got: $recacheOptions")
+
+ val samePlan = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(t1)
+ .filter("id >= 0")
+ val recached = spark.sharedState.cacheManager.lookupCachedData(samePlan)
+ assert(recached.isDefined, "the option-bearing plan should remain cached after refresh")
+ val recachedOptions = recached.get.plan.collect { case r: DataSourceV2Relation =>
+ r.options.get("split-size")
+ }
+ assert(recachedOptions === Seq("5"))
+ } finally {
+ spark.catalog.clearCache()
+ }
+ }
+ }
+
+ test("SPARK-58392: SupportsCatalogOptions uses RelationCatalog state-aware loadTable") {
+ registerCatalog("providerrel", classOf[StateAwareV2InMemoryRelationCatalog])
+ val relCatalog = catalog("providerrel").asInstanceOf[StateAwareV2InMemoryRelationCatalog]
+ val t1 = "providerrel.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+ relCatalog.resetDispatchCalls()
+
+ val df = spark.read
+ .format(classOf[CatalogSupportingInMemoryTableProvider].getName)
+ .option("catalog", "providerrel")
+ .option("name", "table")
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .load()
+ val relation = df.queryExecution.analyzed
+ .collectFirst { case r: DataSourceV2Relation =>
+ r
+ }
+ .getOrElse(fail("expected a v2 relation"))
+
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(relCatalog.loadRelationCalls.isEmpty)
+ assert(relation.options.get("catalog") === "providerrel")
+ assert(relation.options.get("name") === "table")
+ assert(relation.options.get("snapshot") === "s1")
+ assert(relation.options.get("split-size") === "5")
+
+ relCatalog.resetDispatchCalls()
+ df.collect()
+ assertOnlySnapshotTableOptions(relCatalog, "s1", expectedCalls = 1)
+ assert(relCatalog.loadRelationCalls.isEmpty)
+ }
+ }
+
+ test("SPARK-58392: TableCatalog default preserves RelationCatalog table-only context") {
+ registerCatalog("dispatchtrackingrel", classOf[DispatchTrackingRelationCatalog])
+ val relCatalog =
+ catalog("dispatchtrackingrel").asInstanceOf[DispatchTrackingRelationCatalog]
+ val ident = Identifier.of(Array("ns1", "ns2"), "table")
+ val t1 = "dispatchtrackingrel.ns1.ns2.table"
+ val stateOptions = new CaseInsensitiveStringMap(util.Map.of("snapshot", "s1"))
+
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string) USING parquet")
+ relCatalog.resetDispatchCalls()
+
+ relCatalog.loadTable(
+ ident,
+ new TableContext(new TimeTravel.AsOfVersion("v1"), util.Set.of()),
+ stateOptions)
+ assert(relCatalog.loadedVersion === Some("v1"))
+
+ relCatalog.resetDispatchCalls()
+ relCatalog.loadTable(
+ ident,
+ new TableContext(null, util.Set.of(TableWritePrivilege.INSERT)),
+ stateOptions)
+ assert(relCatalog.loadedWritePrivileges === Some(util.Set.of(TableWritePrivilege.INSERT)))
+ }
+ }
+
test("only table-state options are forwarded to loadTable - SQL") {
withStateAwareTable { (stateCatalog, tableName) =>
stateCatalog.resetLoadTableCalls()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org