This is an automated email from the ASF dual-hosted git repository.
aokolnychyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 4b4a2f7b69d5 [SPARK-58389][SQL][FOLLOWUP] Pin DSv2 table instances by
state options during analysis
4b4a2f7b69d5 is described below
commit 4b4a2f7b69d555a2b28a130092f7d75411cb4c49
Author: yyanyy <[email protected]>
AuthorDate: Thu Aug 13 15:05:04 2026 -0700
[SPARK-58389][SQL][FOLLOWUP] Pin DSv2 table instances by state options
during analysis
### What changes were proposed in this pull request?
SPARK-58389 changed DSv2 relation caching to include all read options. This
is required
to preserve each reference's complete option bag, but it also means
references that differ
only in scan-specific options can independently call `loadTable` and
receive different
concrete table versions within one query.
This PR separates relation reuse from table-state pinning:
- The existing `relationCache` remains keyed by all read options and reuses
a finalized relation
only when the complete option bags match.
- A new query-scoped `tableCache` is keyed by catalog, identifier, time
travel, and table-state
options. References with the same table-state key reuse one concrete
`Table`, while retaining
their own complete options in their relations.
- References with different table-state options do not share a table pin.
- `sharedRelationCache` and CacheManager reuse match table identity and
table-state options. The
resolved relation still retains the current reference's complete option
bag.
This PR adds `tableStateOptionKeys()` directly to the evolving
`TableCatalog` API so a catalog can
declare which raw options may affect the table state selected by
`loadTable`, such as a branch, tag,
snapshot, or version. Spark passes only those declared options to the
options-aware `loadTable`.
The default implementation declares no table-state options, so catalogs
that do not override it
receive an empty option map during table loading. The complete option map
remains on the resolved
relation for scan and write planning.
While applying the same table-pinning model to cacheable `V2TableReference`
resolution, this PR
also fixes two existing gaps in `getOrLoadRelation`:
- It previously called `loadTable(identifier)` without passing the options
captured in the table
reference. A table-cache miss now uses the options-aware catalog API with
only the reference's
declared table-state options.
- It previously did not consult `sharedRelationCache`. Temporary-view
re-resolution now consults
`sharedRelationCache` while establishing the initial table pin, allowing
it to preserve a
`Table` already pinned through CacheManager.
Transaction references still use the `Table` loaded through the transaction
catalog and do not
consult `sharedRelationCache`. Write targets remain non-cacheable and
bypass the query-scoped read
caches.
The resulting resolution flow is:
1. Check `relationCache` using the full-option relation key.
2. On a miss, check `tableCache` using the table-state key.
3. On a `tableCache` hit, construct a relation from the pinned `Table` and
the current reference's
complete options. Do not call `loadTable` or consult
`sharedRelationCache`.
4. On a `tableCache` miss, load the current `Table` through the applicable
path. An options-aware
`loadTable` receives only the declared table-state options.
5. Where `sharedRelationCache` lookup applies, reuse its table only when
table identity and
table-state options match. The shared cached `Table` establishes the
initial pin on a match;
otherwise, the newly loaded `Table` establishes it.
6. Store the pinned `Table` in `tableCache` and the finalized relation in
`relationCache`.
Execution-time table refresh uses the same table-state option projection to
preserve this
first-resolution-wins behavior.
### Why are the changes needed?
A catalog may accept both table-state options and scan-specific options.
Using the complete option
bag for relation reuse is necessary, but using it as the only level of
caching can cause references
in the same table-state domain to load different concrete table versions
during one query.
The new `tableCache` pins one concrete `Table` per state key while
finalized relations remain
distinct when their complete option bags differ. Shared cache and
CacheManager lookup use the same
table-state projection while preserving the current reference's complete
options.
Forwarding declared table-state options from `V2TableReference` is
necessary because those options
may select the table state being reloaded. Temporary views additionally
need the
`sharedRelationCache` bridge to preserve a CacheManager-pinned `Table`.
Different state domains, including different parsed time-travel
specifications, continue to
resolve and pin independently.
### Does this PR introduce _any_ user-facing change?
Yes, for catalog implementors only.
This adds the default `tableStateOptionKeys()` method directly to the
evolving `TableCatalog` API.
Catalogs whose raw options select table state can override this method
together with the
options-aware `loadTable`. By default, no options are considered
table-state options and
`loadTable` receives an empty option map. Complete options are still
retained for subsequent scan
and write planning.
There is no new SQL syntax or configuration. The option-forwarding behavior
refined here was
introduced on the unreleased master branch by SPARK-58389.
### How was this patch tested?
Added regression coverage for table-state projection and load-option
filtering, table pinning,
default no-state-option behavior, `sharedRelationCache` and CacheManager
matching, nested analysis,
execution refresh, and temporary-view, transaction, and write-target
`V2TableReference` behavior.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
Closes #57799 from
yyanyy/yan-yan_data/spark-dsv2-table-state-cache-20260804.
Authored-by: yyanyy <[email protected]>
Signed-off-by: Anton Okolnychyi <[email protected]>
---
.../spark/sql/connector/catalog/TableCatalog.java | 36 +-
.../spark/sql/catalyst/analysis/Analyzer.scala | 5 +
.../sql/catalyst/analysis/RelationCache.scala | 11 +-
.../sql/catalyst/analysis/RelationCacheKey.scala | 6 +-
.../sql/catalyst/analysis/RelationResolution.scala | 152 +++--
...{RelationCacheKey.scala => TableCacheKey.scala} | 15 +-
.../sql/catalyst/analysis/V2TableReference.scala | 7 +
.../sql/connector/catalog/CatalogV2Util.scala | 45 +-
.../catalyst/analysis/TableLookupCacheSuite.scala | 17 +
.../sql/connector/catalog/CatalogV2UtilSuite.scala | 88 +++
.../connector/catalog/InMemoryTableCatalog.scala | 16 +-
.../apache/spark/sql/execution/CacheManager.scala | 25 +-
.../datasources/v2/V2TableRefreshUtil.scala | 13 +-
.../apache/spark/sql/internal/SharedState.scala | 3 +-
.../sql/connector/DataSourceV2OptionSuite.scala | 688 ++++++++++++++++++---
.../connector/SupportsCatalogOptionsSuite.scala | 14 +-
.../execution/command/PlanResolutionSuite.scala | 177 +++++-
17 files changed, 1108 insertions(+), 210 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java
index 58a40c237332..5d6e9d01f782 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
@@ -104,6 +104,26 @@ public interface TableCatalog extends CatalogPlugin {
*/
default Set<TableCatalogCapability> capabilities() { return Set.of(); }
+ /**
+ * Returns the connector-specific option keys that select the table state
(such as a branch, tag,
+ * snapshot, or version) and therefore must be known when the table is
loaded. Keys that Spark
+ * parses and handles itself, such as time travel, must not be listed here.
+ * <p>
+ * Spark may need to resolve the same table more than once while analyzing
or refreshing a query.
+ * Spark reuses one table instance only for references whose table-state
options match and passes
+ * only the declared options to {@code loadTable}. The complete user option
map remains on each
+ * resolved relation for subsequent scan and write planning.
+ * <p>
+ * The default implementation returns an empty set, treating all options as
unable to select a
+ * different table state. Option key matching is case-insensitive, while
option values remain
+ * case-sensitive.
+ *
+ * @return a non-null set of case-insensitive option keys
+ *
+ * @since 4.3.0
+ */
+ default Set<String> tableStateOptionKeys() { return Set.of(); }
+
/**
* List the tables in a namespace from the catalog.
*
@@ -196,12 +216,14 @@ public interface TableCatalog extends CatalogPlugin {
}
/**
- * Load table metadata by {@link Identifier identifier} from the catalog,
forwarding all
- * user-specified options.
+ * Load table metadata by {@link Identifier identifier} from the catalog,
forwarding the
+ * user-specified options that may affect table state.
* <p>
- * The default implementation ignores {@code options} and delegates to the
existing
+ * The default implementation ignores {@code stateOptions} 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.
+ * options while reading a table must override {@link
#tableStateOptionKeys()} and this method.
+ * Spark passes only the options declared by {@link
#tableStateOptionKeys()}. Spark retains the
+ * complete user option map on the resolved relation for subsequent scan and
write planning.
* <p>
* An override replaces that dispatch and must honor {@code context} itself:
apply the time
* travel in {@link TableContext#timeTravel()}, and authorize the requested
@@ -210,8 +232,8 @@ public interface TableCatalog extends CatalogPlugin {
*
* @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}
+ * @param stateOptions options declared to affect table state; Spark-parsed
state such as time
+ * travel is provided through {@code context} instead
* @return the table's metadata
* @throws NoSuchTableException If the table doesn't exist
*
@@ -220,7 +242,7 @@ public interface TableCatalog extends CatalogPlugin {
default Table loadTable(
Identifier ident,
TableContext context,
- CaseInsensitiveStringMap options) throws NoSuchTableException {
+ CaseInsensitiveStringMap stateOptions) throws NoSuchTableException {
if (context.timeTravel().isPresent()) {
TimeTravel timeTravel = context.timeTravel().get();
if (timeTravel instanceof TimeTravel.AsOfVersion v) {
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 4b2a1cc1114c..e790ea2e61a8 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
@@ -135,6 +135,9 @@ object FakeV2SessionCatalog extends TableCatalog with
FunctionCatalog with Suppo
* @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 tableCache A mapping from (catalog, identifier, time travel spec,
table-state options) to
+ * concrete tables. This pins one table state while allowing
references to keep
+ * different read-specific 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,
@@ -155,6 +158,7 @@ case class AnalysisContext(
nestedViewDepth: Int = 0,
maxNestedViewDepth: Int = -1,
relationCache: mutable.Map[RelationCacheKey, LogicalPlan] =
mutable.Map.empty,
+ tableCache: mutable.Map[TableCacheKey, Table] = 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)`.
@@ -249,6 +253,7 @@ object AnalysisContext {
nestedViewDepth = originContext.nestedViewDepth + 1,
maxNestedViewDepth = maxNestedViewDepth,
relationCache = originContext.relationCache,
+ tableCache = originContext.tableCache,
referredTempViewNames = viewDesc.viewReferredTempViewNames,
referredTempFunctionNames =
mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
referredTempVariableNames = viewDesc.viewReferredTempVariableNames,
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCache.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCache.scala
index 770a5e780b24..7c34ad8911f6 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCache.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCache.scala
@@ -18,11 +18,18 @@
package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
private[sql] trait RelationCache {
- def lookup(nameParts: Seq[String], resolver: Resolver): Option[LogicalPlan]
+ def lookup(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ tableId: Option[String],
+ stateOptions: CaseInsensitiveStringMap,
+ resolver: Resolver): Option[LogicalPlan]
}
private[sql] object RelationCache {
- val empty: RelationCache = (_, _) => None
+ val empty: RelationCache = (_, _, _, _, _) => None
}
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
index c89a9b0b9581..fc4217cdf701 100644
---
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
@@ -22,9 +22,9 @@ 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.
+ * The complete option map is part of the key because the resolved relation
carries it into scan
+ * and write planning. References may reuse the same `Table` when their
table-state options match,
+ * but references with different options must not reuse the same cached
relation.
*/
private[sql] case class RelationCacheKey(
nameParts: Seq[String],
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 d217219cd8f9..130ee920cf49 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
@@ -62,6 +62,7 @@ class RelationResolution(
val v1SessionCatalog = catalogManager.v1SessionCatalog
private def relationCache = AnalysisContext.get.relationCache
+ private def tableCache = AnalysisContext.get.tableCache
/**
* If we are resolving database objects (relations, functions, etc.) inside
views, we may need to
@@ -254,6 +255,14 @@ class RelationResolution(
cached
.map(adaptCachedRelation(_, planId))
.orElse {
+ lazy val tableKey =
+ toTableCacheKey(catalog, ident, finalTimeTravelSpec,
finalOptions)
+ val pinnedTable = if (writePrivileges == null &&
catalog.isInstanceOf[TableCatalog]) {
+ tableCache.get(tableKey)
+ } else {
+ 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
@@ -263,61 +272,66 @@ class RelationResolution(
// 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.
- val relation: Option[Relation] = catalog match {
- case mc: RelationCatalog if finalTimeTravelSpec.isEmpty &&
writePrivileges == null =>
- try {
- Some(mc.loadRelation(ident))
- } catch {
- case _: NoSuchTableException => None
- }
- case _ =>
- val tableSide: Option[Table] = if (
- CatalogV2Util.isSessionCatalog(catalog) ||
catalog.isInstanceOf[TableCatalog]
- ) {
- CatalogV2Util.loadTable(
- catalog,
- ident,
- finalTimeTravelSpec,
- Option(writePrivileges),
- finalOptions)
- } else {
- None
- }
- // Fallback to ViewCatalog for catalogs that host views but
where loadTable
- // returned None (or was skipped because there's no
TableCatalog mixin).
- // Time-travel / write privileges only apply to tables, not
views, so the
- // fallback only fires when both are absent.
- tableSide.orElse {
- if (finalTimeTravelSpec.isEmpty && writePrivileges == null) {
- catalog match {
- case vc: ViewCatalog =>
- try {
- Some(vc.loadView(ident))
- } catch {
- case _: NoSuchViewException => None
- }
- case _ => None
- }
+ val relation: Option[Relation] = pinnedTable.orElse {
+ catalog match {
+ case mc: RelationCatalog
+ if finalTimeTravelSpec.isEmpty && writePrivileges == null
=>
+ try {
+ Some(mc.loadRelation(ident))
+ } catch {
+ case _: NoSuchTableException => None
+ }
+ case _ =>
+ val tableSide: Option[Table] = if (
+ CatalogV2Util.isSessionCatalog(catalog) ||
catalog.isInstanceOf[TableCatalog]
+ ) {
+ CatalogV2Util.loadTable(
+ catalog,
+ ident,
+ finalTimeTravelSpec,
+ Option(writePrivileges),
+ finalOptions)
} else {
None
}
- }
+ // Fallback to ViewCatalog for catalogs that host views but
where loadTable
+ // returned None (or was skipped because there's no
TableCatalog mixin).
+ // Time-travel / write privileges only apply to tables, not
views, so the
+ // fallback only fires when both are absent.
+ tableSide.orElse {
+ if (finalTimeTravelSpec.isEmpty && writePrivileges ==
null) {
+ catalog match {
+ case vc: ViewCatalog =>
+ try {
+ Some(vc.loadView(ident))
+ } catch {
+ case _: NoSuchViewException => None
+ }
+ case _ => None
+ }
+ } else {
+ None
+ }
+ }
+ }
}
// `table` is `relation` filtered to tables only -- used for cache
lookup since
// 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`.
+ // Reuse a cached relation only when its table identity and state
options match. The
+ // returned relation still carries this read's complete option map.
val sharedRelationCacheMatch = for {
t <- table
- if finalTimeTravelSpec.isEmpty && writePrivileges == null &&
!u.isStreaming
- cached <- lookupSharedRelationCache(catalog, ident, t)
- if cached.options == finalOptions
+ if pinnedTable.isEmpty && finalTimeTravelSpec.isEmpty &&
+ writePrivileges == null && !u.isStreaming
+ cached <- lookupSharedRelationCache(catalog, ident, t,
finalOptions)
} yield {
+ val updatedRelation = cached.copy(options = finalOptions)
+ updatedRelation.copyTagsFrom(cached)
val nameParts = ident.toQualifiedNameParts(catalog)
- val aliasedRelation = SubqueryAlias(nameParts, cached)
+ val aliasedRelation = SubqueryAlias(nameParts, updatedRelation)
+ tableCache.update(tableKey, cached.table)
relationCache.update(key, aliasedRelation)
adaptCachedRelation(aliasedRelation, planId)
}
@@ -330,6 +344,12 @@ class RelationResolution(
finalOptions,
u.isStreaming,
finalTimeTravelSpec)
+ // A write target skips cache lookup above so authorization
always runs, but its
+ // freshly loaded Table is still published for subsequent reads,
matching the
+ // relation cache behavior below.
+ if (pinnedTable.isEmpty) {
+ table.foreach(tableCache.update(tableKey, _))
+ }
loaded.foreach(relationCache.update(key, _))
loaded.map(cloneWithPlanId(_, planId))
}
@@ -368,8 +388,9 @@ class RelationResolution(
private def lookupSharedRelationCache(
catalog: CatalogPlugin,
ident: Identifier,
- table: Table): Option[DataSourceV2Relation] = {
- CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident,
table, conf)
+ table: Table,
+ options: CaseInsensitiveStringMap): Option[DataSourceV2Relation] = {
+ CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident,
table, options, conf)
}
private def adaptCachedRelation(cached: LogicalPlan, planId: Option[Long]):
LogicalPlan = {
@@ -490,7 +511,27 @@ class RelationResolution(
case Some(cached) =>
adaptCachedRelation(cached, ref)
case None =>
- val relation = loadRelation(ref)
+ val catalog = catalogManager.catalog(ref.catalog.name).asTableCatalog
+ val tableKey = toTableCacheKey(catalog, ref.identifier, None,
ref.options)
+ val relation = tableCache.get(tableKey) match {
+ case Some(pinnedTable) =>
+ createRelation(ref, catalog, pinnedTable)
+ 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)
+ } else {
+ None
+ }
+ sharedCacheMatch match {
+ case Some(cached) =>
+ tableCache.update(tableKey, cached.table)
+ adaptCachedRelation(cached, ref)
+ case None =>
+ tableCache.update(tableKey, table)
+ createRelation(ref, catalog, table)
+ }
+ }
relationCache.update(key, relation)
relation
}
@@ -508,6 +549,13 @@ class RelationResolution(
private def loadRelation(ref: V2TableReference): LogicalPlan = {
val resolvedCatalog =
catalogManager.catalog(ref.catalog.name).asTableCatalog
val table = resolvedCatalog.loadTable(ref.identifier)
+ createRelation(ref, resolvedCatalog, table)
+ }
+
+ private def createRelation(
+ ref: V2TableReference,
+ resolvedCatalog: TableCatalog,
+ table: Table): DataSourceV2Relation = {
V2TableReferenceUtils.validateLoadedTable(table, ref)
DataSourceV2Relation(
table = table,
@@ -551,6 +599,18 @@ class RelationResolution(
RelationCacheKey(nameParts, timeTravelSpec, options)
}
+ private def toTableCacheKey(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ timeTravelSpec: Option[TimeTravelSpec],
+ options: CaseInsensitiveStringMap): TableCacheKey = {
+ TableCacheKey(
+ catalog,
+ ident,
+ timeTravelSpec,
+ CatalogV2Util.extractTableStateOptions(catalog, options))
+ }
+
private def cloneWithPlanId(plan: LogicalPlan, planId: Option[Long]):
LogicalPlan = {
planId match {
case Some(id) =>
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/TableCacheKey.scala
similarity index 67%
copy from
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationCacheKey.scala
copy to
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableCacheKey.scala
index c89a9b0b9581..186c4b9855a4 100644
---
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/TableCacheKey.scala
@@ -17,16 +17,17 @@
package org.apache.spark.sql.catalyst.analysis
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
/**
- * Key for the per-query relation cache in [[AnalysisContext]], shared by
[[RelationResolution]].
+ * Key for the per-query table-state cache in [[AnalysisContext]].
*
- * 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.
+ * Unlike [[RelationCacheKey]], this key contains only options declared to
affect table state. This
+ * lets references retain different scan options while sharing one concrete
table state.
*/
-private[sql] case class RelationCacheKey(
- nameParts: Seq[String],
+private[sql] case class TableCacheKey(
+ catalog: CatalogPlugin,
+ identifier: Identifier,
timeTravelSpec: Option[TimeTravelSpec],
- options: CaseInsensitiveStringMap)
+ stateOptions: CaseInsensitiveStringMap)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V2TableReference.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V2TableReference.scala
index 7bad6d149c60..2d4ef18b5f60 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V2TableReference.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V2TableReference.scala
@@ -86,22 +86,29 @@ private[sql] object V2TableReference {
metadataColumns: Seq[MetadataColumn])
sealed trait Context {
+ /** Whether re-resolution may reuse the per-query relation cache. */
def cacheable: Boolean
+
+ /** Whether re-resolution may reuse the shared (CACHE TABLE) relation
cache. */
+ def sharedCacheable: Boolean
}
/** Context for relations that are re-resolved on access of a dataframe temp
view. */
case class TemporaryViewContext(viewName: Seq[String]) extends Context {
val cacheable = true
+ val sharedCacheable = true
}
/** Context for relations that are re-resolved through a transaction
catalog. */
case object TransactionContext extends Context {
val cacheable = true
+ val sharedCacheable = false
}
/** Context for write targets. */
case object WriteTargetContext extends Context {
val cacheable = false
+ val sharedCacheable = false
}
def createForTempView(relation: DataSourceV2Relation, viewName:
Seq[String]): V2TableReference = {
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 a29fe880b915..8a9e77693824 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
@@ -481,6 +481,28 @@ private[sql] object CatalogV2Util {
case _: NoSuchDatabaseException => None
}
+ /**
+ * Extracts the options that may select table state from a complete option
map. These are the
+ * only options passed to `loadTable` and used to identify a pinned table
state.
+ */
+ def extractTableStateOptions(
+ catalog: CatalogPlugin,
+ options: CaseInsensitiveStringMap): CaseInsensitiveStringMap = {
+ val stateKeys = catalog.asTableCatalog.tableStateOptionKeys.asScala
+ .map(_.toLowerCase(Locale.ROOT))
+ .toSet
+ val projected = options.entrySet.asScala.collect {
+ case entry if stateKeys.contains(entry.getKey.toLowerCase(Locale.ROOT))
=>
+ entry.getKey -> entry.getValue
+ }.toMap
+ new CaseInsensitiveStringMap(projected.asJava)
+ }
+
+ /**
+ * Loads a table from the catalog. Callers may pass the complete option map,
but only the keys the
+ * catalog declares via `tableStateOptionKeys()` are forwarded to
`loadTable`, so the loaded table
+ * state stays independent of non-state options and of how many times the
table is referenced.
+ */
def getTable(
catalog: CatalogPlugin,
ident: Identifier,
@@ -493,7 +515,8 @@ private[sql] object CatalogV2Util {
case None => null
}
val context = new TableContext(timeTravel,
parseWritePrivileges(writePrivilegesString))
- catalog.asTableCatalog.loadTable(ident, context, options)
+ val stateOptions = extractTableStateOptions(catalog, options)
+ catalog.asTableCatalog.loadTable(ident, context, stateOptions)
}
/**
@@ -534,25 +557,19 @@ private[sql] object CatalogV2Util {
loadTable(catalog, ident).map(DataSourceV2Relation.create(_,
Some(catalog), Some(ident)))
}
- def isSameTable(
- rel: DataSourceV2Relation,
- catalog: CatalogPlugin,
- ident: Identifier,
- table: Table): Boolean = {
- rel.catalog.contains(catalog) && rel.identifier.contains(ident) &&
rel.table.id == table.id
- }
-
def lookupCachedRelation(
cache: RelationCache,
catalog: CatalogPlugin,
ident: Identifier,
table: Table,
+ options: CaseInsensitiveStringMap,
conf: SQLConf): Option[DataSourceV2Relation] = {
- val nameParts = ident.toQualifiedNameParts(catalog)
- val cached = cache.lookup(nameParts, conf.resolver)
- cached.collect {
- case r: DataSourceV2Relation if isSameTable(r, catalog, ident, table) =>
r
- }
+ cache.lookup(
+ catalog,
+ ident,
+ Some(table.id),
+ extractTableStateOptions(catalog, options),
+ conf.resolver).collect { case r: DataSourceV2Relation => r }
}
def isSessionCatalog(catalog: CatalogPlugin): Boolean = {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TableLookupCacheSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TableLookupCacheSuite.scala
index 75846aa49616..c078504ea33a 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TableLookupCacheSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TableLookupCacheSuite.scala
@@ -119,4 +119,21 @@ class TableLookupCacheSuite extends AnalysisTest with
Matchers {
verify(catalog, times(1)).getTable("default", "t1")
}
}
+
+ test("nested view analysis shares both query-scoped caches") {
+ AnalysisContext.withNewAnalysisContext {
+ val outer = AnalysisContext.get
+ val viewDesc = CatalogTable(
+ TableIdentifier("view", Some("default")),
+ CatalogTableType.VIEW,
+ CatalogStorageFormat.empty,
+ StructType(Seq(StructField("a", IntegerType))),
+ viewText = Some("select * from t1"))
+
+ AnalysisContext.withAnalysisContext(viewDesc) {
+ assert(AnalysisContext.get.relationCache eq outer.relationCache)
+ assert(AnalysisContext.get.tableCache eq outer.tableCache)
+ }
+ }
+ }
}
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 f124186903e0..f8919a8730f2 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
@@ -28,12 +28,19 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
class CatalogV2UtilSuite extends SparkFunSuite {
+ private def catalogWithStateOptions(keys: java.util.Set[String]):
TableCatalog = {
+ val catalog = mock(classOf[TableCatalog])
+ when(catalog.tableStateOptionKeys()).thenReturn(keys)
+ catalog
+ }
+
// 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.tableStateOptionKeys()).thenCallRealMethod()
when(testCatalog.loadTable(
any[Identifier], any[TableContext],
any[CaseInsensitiveStringMap])).thenCallRealMethod()
testCatalog
@@ -114,6 +121,87 @@ class CatalogV2UtilSuite extends SparkFunSuite {
assert(a.toString.contains("writePrivileges"))
}
+ test("getTable forwards only declared table-state options") {
+ val catalog = mock(classOf[TableCatalog])
+
when(catalog.tableStateOptionKeys()).thenReturn(java.util.Set.of("snapshot"))
+ val ident = mock(classOf[Identifier])
+ val options = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "5"))
+
+ CatalogV2Util.getTable(catalog, ident, options = options)
+
+ val expected = new CaseInsensitiveStringMap(java.util.Map.of("snapshot",
"s1"))
+ verify(catalog).loadTable(
+ mockEq(ident),
+ any[TableContext],
+ mockEq(expected))
+ }
+
+ test("getTable forwards no options when a catalog declares no table-state
options") {
+ val catalog = mock(classOf[TableCatalog])
+ when(catalog.tableStateOptionKeys()).thenCallRealMethod()
+ val ident = mock(classOf[Identifier])
+ val options = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "5"))
+
+ CatalogV2Util.getTable(catalog, ident, options = options)
+
+ verify(catalog).loadTable(
+ mockEq(ident),
+ any[TableContext],
+ mockEq(CaseInsensitiveStringMap.empty()))
+ }
+
+ test("extractTableStateOptions projects declared keys case-insensitively") {
+ val catalog = catalogWithStateOptions(java.util.Set.of("BrAnCh", "tag"))
+ val options = new CaseInsensitiveStringMap(java.util.Map.of(
+ "branch", "Main",
+ "TAG", "Release",
+ "split-size", "5"))
+
+ val stateOptions = CatalogV2Util.extractTableStateOptions(catalog, options)
+
+ assert(stateOptions.size() == 2)
+ assert(stateOptions.get("BRANCH") == "Main")
+ assert(stateOptions.get("tag") == "Release")
+ assert(!stateOptions.containsKey("split-size"))
+ }
+
+ test("extractTableStateOptions compares option keys case-insensitively") {
+ val catalog = catalogWithStateOptions(java.util.Set.of("SnApShOt"))
+ val lowerCaseKey = CatalogV2Util.extractTableStateOptions(
+ catalog,
+ new CaseInsensitiveStringMap(java.util.Map.of("snapshot", "main")))
+ val upperCaseKey = CatalogV2Util.extractTableStateOptions(
+ catalog,
+ new CaseInsensitiveStringMap(java.util.Map.of("SNAPSHOT", "main")))
+
+ assert(lowerCaseKey == upperCaseKey)
+ }
+
+ test("extractTableStateOptions compares option values case-sensitively") {
+ val catalog = catalogWithStateOptions(java.util.Set.of("snapshot"))
+ val lowerCaseValue = CatalogV2Util.extractTableStateOptions(
+ catalog,
+ new CaseInsensitiveStringMap(java.util.Map.of("snapshot", "main")))
+ val upperCaseValue = CatalogV2Util.extractTableStateOptions(
+ catalog,
+ new CaseInsensitiveStringMap(java.util.Map.of("snapshot", "MAIN")))
+
+ assert(lowerCaseValue != upperCaseValue)
+ }
+
+ test("extractTableStateOptions returns no options by default") {
+ val catalog = mock(classOf[TableCatalog])
+ when(catalog.tableStateOptionKeys()).thenCallRealMethod()
+ val options = new CaseInsensitiveStringMap(
+ java.util.Map.of("branch", "Main", "split-size", "5"))
+
+ val stateOptions = CatalogV2Util.extractTableStateOptions(catalog, options)
+
+ assert(stateOptions.isEmpty)
+ }
+
test("viewInfoBuilderFrom preserves the dependency list") {
val dependencies = DependencyList.of(Array(Dependency.table(Array("cat",
"ns", "events"))))
val existing = viewWithDependencies(Some(dependencies))
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 53a7fa05b40e..c46f96493ea7 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
@@ -49,10 +49,10 @@ 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.
+ // Records every (TableContext, table-state 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)]
@@ -138,14 +138,14 @@ class BasicInMemoryTableCatalog extends TableCatalog {
}
}
- // Records the forwarded context/options so tests can verify they reached
the catalog, then
+ // Records the forwarded context/state 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)
+ stateOptions: CaseInsensitiveStringMap): Table = {
+ _loadTableCalls += ((context, stateOptions))
+ super.loadTable(ident, context, stateOptions)
}
override def invalidateTable(ident: Identifier): Unit = {
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 3541c939909f..a7adfdf4049e 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
@@ -42,6 +42,7 @@ import org.apache.spark.sql.execution.command.CommandUtils
import org.apache.spark.sql.execution.datasources.{FileIndex,
HadoopFsRelation, LogicalRelation, LogicalRelationWithTable}
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2Relation, ExtractV2CatalogAndIdentifier, ExtractV2Table, FileTable,
V2TableRefreshUtil}
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK
@@ -436,17 +437,25 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
private[sql] def lookupCachedTable(
- name: Seq[String],
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ tableId: Option[String],
+ stateOptions: CaseInsensitiveStringMap,
resolver: Resolver): Option[LogicalPlan] = {
+ val name = ident.toQualifiedNameParts(catalog)
val cachedRelations = findCachedRelations(name, resolver)
- cachedRelations match {
- case cachedRelation +: _ =>
- CacheManager.logCacheOperation(
- log"Relation cache hit for table ${MDC(TABLE_NAME, name.quoted)}")
- Some(cachedRelation)
- case _ =>
- None
+ val cachedRelation = cachedRelations.collectFirst {
+ case r: DataSourceV2Relation
+ if r.catalog.contains(catalog) && r.identifier.contains(ident) &&
+ tableId.forall(_ == r.table.id) &&
+ CatalogV2Util.extractTableStateOptions(catalog, r.options) ==
stateOptions =>
+ r
+ }
+ cachedRelation.foreach { _ =>
+ CacheManager.logCacheOperation(
+ log"Relation cache hit for table ${MDC(TABLE_NAME, name.quoted)}")
}
+ cachedRelation
}
private def findCachedRelations(
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 10408aa51363..6653a10c87c1 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
@@ -88,10 +88,11 @@ private[sql] object V2TableRefreshUtil extends
SQLConfHelper with Logging {
plan transformWithSubqueries {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
if (r.isVersioned || !versionedOnly) && r.timeTravelSpec.isEmpty =>
- val currentTable = currentTables.getOrElseUpdate((catalog, ident,
r.options), {
+ 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) match {
- case Some(cached) if cached.options == r.options =>
+ lookupCachedRelation(spark, catalog, ident, r.table, r.options)
match {
+ case Some(cached) =>
logDebug(s"Refreshing table metadata for $tableName using shared
relation cache")
cached.table
case _ =>
@@ -110,8 +111,10 @@ private[sql] object V2TableRefreshUtil extends
SQLConfHelper with Logging {
spark: SparkSession,
catalog: TableCatalog,
ident: Identifier,
- table: Table): Option[DataSourceV2Relation] = {
- CatalogV2Util.lookupCachedRelation(spark.sharedState.relationCache,
catalog, ident, table, conf)
+ table: Table,
+ options: CaseInsensitiveStringMap): Option[DataSourceV2Relation] = {
+ CatalogV2Util.lookupCachedRelation(
+ spark.sharedState.relationCache, catalog, ident, table, options, conf)
}
// it is not safe to allow any schema changes in commands (e.g. CTAS, RTAS,
MERGE)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala
b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala
index 8e641294bf8c..3271f0e1f100 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala
@@ -101,7 +101,8 @@ private[sql] class SharedState(
* A relation cache backed by the cache manager.
*/
private[sql] val relationCache: RelationCache = {
- (nameParts, resolver) => cacheManager.lookupCachedTable(nameParts,
resolver)
+ (catalog, ident, tableId, stateOptions, resolver) =>
+ cacheManager.lookupCachedTable(catalog, ident, tableId, stateOptions,
resolver)
}
/** A global lock for all streaming query lifecycle tracking and management.
*/
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 8cdf12b18f44..82b5938ae813 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,17 +17,32 @@
package org.apache.spark.sql.connector
+import java.util
import java.util.concurrent.atomic.AtomicInteger
-import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
-import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.analysis.{
+ AnalysisContext,
+ RelationCache,
+ RelationResolution,
+ UnresolvedRelation,
+ V2TableReference}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
-import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryBaseTable,
InMemoryCatalog, InMemoryRowLevelOperationTableCatalog, Table, TimeTravel}
+import org.apache.spark.sql.connector.catalog.{
+ Identifier,
+ InMemoryBaseTable,
+ InMemoryCatalog,
+ InMemoryRowLevelOperationTableCatalog,
+ Table,
+ TableChange,
+ TableWritePrivilege,
+ TimeTravel}
import org.apache.spark.sql.execution.CommandResultExec
import org.apache.spark.sql.execution.datasources.v2._
import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
class LoadCountingInMemoryCatalog extends InMemoryCatalog {
val singleArgLoads = new AtomicInteger(0)
@@ -38,6 +53,13 @@ class LoadCountingInMemoryCatalog extends InMemoryCatalog {
}
}
+class StateAwareInMemoryCatalog extends LoadCountingInMemoryCatalog {
+ // 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] =
+ util.Set.of("snapshot", UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
+}
+
class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
import testImplicits._
@@ -46,6 +68,30 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
private def inMemoryCatalog: InMemoryCatalog =
catalog("testcat").asInstanceOf[InMemoryCatalog]
+ private def withStateAwareTable(
+ f: (StateAwareInMemoryCatalog, String) => Unit): Unit = {
+ withSQLConf(
+ "spark.sql.catalog.statecat" ->
classOf[StateAwareInMemoryCatalog].getName,
+ "spark.sql.catalog.statecat.copyOnLoad" -> "true") {
+ val tableName = "statecat.ns.table"
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (id bigint, data string)")
+ sql(s"INSERT INTO $tableName VALUES (1, 'a'), (2, 'b')")
+ f(catalog("statecat").asInstanceOf[StateAwareInMemoryCatalog],
tableName)
+ }
+ }
+ }
+
+ private def assertOnlySnapshotOptions(
+ catalog: StateAwareInMemoryCatalog,
+ expectedSnapshot: String): Unit = {
+ val loadOptions = catalog.loadTableCalls.map(_._2)
+ assert(loadOptions.nonEmpty, "expected at least one options-aware table
load")
+ assert(loadOptions.forall { options =>
+ options.size() == 1 && options.get("snapshot") == expectedSnapshot
+ }, s"expected only snapshot=$expectedSnapshot to be forwarded, got:
$loadOptions")
+ }
+
test("SPARK-36680: Supports Dynamic Table Options for SQL Select") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
@@ -442,44 +488,45 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
}
}
- 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()
+ test("only table-state options are forwarded to loadTable - DataFrame API") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ spark.read
+ .option("snapshot", "s1")
+ .option("customOption", "customValue")
+ .table(tableName)
+ .collect()
- val opts = inMemoryCatalog.lastLoadTableOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") === "customValue")
+ assertOnlySnapshotOptions(stateCatalog, "s1")
}
}
- 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()
+ test("only table-state options are forwarded to loadTable - SQL") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ sql(s"SELECT * FROM $tableName " +
+ "WITH ('snapshot' = 's1', 'customOption' = 'customValue')").collect()
- val opts = inMemoryCatalog.lastLoadTableOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") === "customValue")
+ assertOnlySnapshotOptions(stateCatalog, "s1")
}
}
- test("options are forwarded to loadTable - DataStreamReader") {
- val t1 = s"${catalogAndNamespace}table"
- withTable(t1) {
- sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ test("only table-state options are forwarded to loadTable -
DataStreamReader") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
// Trigger analysis of the streaming relation.
- spark.readStream.option("customOption",
"customValue").table(t1).queryExecution.analyzed
+ spark.readStream
+ .option("snapshot", "s1")
+ .option("customOption", "customValue")
+ .table(tableName)
+ .queryExecution
+ .analyzed
- val opts = inMemoryCatalog.lastLoadTableOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") === "customValue")
+ assertOnlySnapshotOptions(stateCatalog, "s1")
}
}
- test("options are forwarded to loadTable alongside time travel") {
+ test("time travel is passed in TableContext and excluded from load options")
{
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
@@ -500,29 +547,28 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
val opts = inMemoryCatalog.lastLoadTableOptions
assert(opts.isDefined)
- assert(opts.get.get("customOption") === "customValue")
- assert(opts.get.get("versionAsOf") === "v1")
+ assert(opts.get.isEmpty)
}
}
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')")
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ sql(s"INSERT INTO $tableName WITH ('snapshot' = 's1') VALUES (3, 'c')")
- val ctx = inMemoryCatalog.lastTableContext
+ val ctx = stateCatalog.lastTableContext
assert(ctx.isDefined)
assert(!ctx.get.writePrivileges().isEmpty)
- val opts = inMemoryCatalog.lastLoadTableOptions
+ val opts = stateCatalog.lastLoadTableOptions
assert(opts.isDefined)
+ assert(opts.get.get("snapshot") === "s1")
// 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") {
+ test("execution refresh filters load options when a catalog declares no
state options") {
registerCatalog("loadcounting", classOf[LoadCountingInMemoryCatalog])
val loadCountingCatalog =
catalog("loadcounting").asInstanceOf[LoadCountingInMemoryCatalog]
@@ -535,52 +581,47 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
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.
+ // Both analysis and the execution-time refresh enter through the
options-aware overload,
+ // but this catalog declares no table-state options, so each receives an
empty option map.
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(optionAwareLoads.size === 2,
+ s"expected one analysis load and one refresh load, got:
$optionAwareLoads")
+ assert(optionAwareLoads.forall(_._2.isEmpty),
+ s"expected analysis and refresh to filter split-size, 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") {
+ test("catalogs with no declared state options share table state across
option bags") {
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.
+ // The catalog declares no table-state option keys, so none of them
affect table state.
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 first reference loads the table and the second reuses that table
state.
+ val analysisLoads = inMemoryCatalog.loadTableCalls
+ assert(analysisLoads.size === 1,
+ s"expected the second option bag to reuse the loaded table state, got:
$analysisLoads")
+ assert(analysisLoads.head._2.isEmpty,
+ s"expected no table-state options to be forwarded, got:
$analysisLoads")
- // The execution-time refresh also reloads once per option bag instead
of sharing one Table
- // across the two references.
+ // The execution-time refresh loads the first reference and reuses its
table state for the
+ // second reference.
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")
+ val refreshLoads = inMemoryCatalog.loadTableCalls
+ assert(refreshLoads.size === 1,
+ s"expected refresh to share table state across option bags, got:
$refreshLoads")
+ assert(refreshLoads.head._2.isEmpty,
+ s"expected refresh to filter scan options, got: $refreshLoads")
// Each scan also keeps its own option end-to-end (neither reference
inherits the other's).
val splitSizes = df.queryExecution.optimizedPlan.collect {
@@ -590,6 +631,468 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
}
}
+ test("same table state shares one Table while preserving each reference's
options") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ stateCatalog.singleArgLoads.set(0)
+
+ val df = sql(s"SELECT a.id FROM $tableName " +
+ s"WITH (`SnApShOt` = 's1', `split-size` = 5) a JOIN $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) b ON a.id = b.id")
+
+ val analyzedRelations = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("split-size") =>
r
+ }
+ assert(analyzedRelations.size == 2)
+ assert(analyzedRelations.map(_.options.get("split-size")).sorted ==
Seq("5", "9"))
+ assert(analyzedRelations.map(_.options.get("snapshot")).distinct ==
Seq("s1"))
+ assert(analyzedRelations.head.table eq analyzedRelations.last.table)
+
+ val analysisLoads =
stateCatalog.loadTableCalls.filter(_._2.get("snapshot") == "s1")
+ assert(analysisLoads.size == 1,
+ s"expected one catalog load for state s1 during analysis, got:
$analysisLoads")
+ assert(analysisLoads.head._2.size() == 1,
+ s"expected scan options to be filtered during analysis, got:
$analysisLoads")
+
+ stateCatalog.resetLoadTableCalls()
+ stateCatalog.singleArgLoads.set(0)
+ assert(df.collect().toSeq == Seq(Row(1), Row(2)))
+
+ val refreshLoads =
stateCatalog.loadTableCalls.filter(_._2.get("snapshot") == "s1")
+ assert(refreshLoads.size == 1,
+ s"expected one catalog load for state s1 during refresh, got:
$refreshLoads")
+ assert(refreshLoads.head._2.size() == 1,
+ s"expected scan options to be filtered during refresh, got:
$refreshLoads")
+ val refreshedRelations = df.queryExecution.optimizedPlan.collect {
+ case s: DataSourceV2ScanRelation if
s.relation.options.containsKey("split-size") =>
+ s.relation
+ }
+ assert(refreshedRelations.size == 2)
+ assert(refreshedRelations.head.table eq refreshedRelations.last.table)
+ }
+ }
+
+ test("shared relation cache selects by state and preserves each reference's
full options") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ val cached = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(tableName)
+ val otherStateCached = spark.read
+ .option("snapshot", "s2")
+ .option("split-size", "7")
+ .table(tableName)
+ cached.cache()
+ otherStateCached.cache()
+ try {
+ cached.collect()
+ // Cache this relation last so it is searched first. An s1 lookup must
scan past it.
+ otherStateCached.collect()
+ val cachedTable = cached.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.getOrElse(fail("expected a cached v2 relation"))
+
+ def relations(df: DataFrame): Seq[DataSourceV2Relation] = {
+ df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("snapshot")
=> r
+ }
+ }
+
+ stateCatalog.resetLoadTableCalls()
+ val cachedFirst = sql(s"SELECT a.id FROM $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 5) a JOIN $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) b ON a.id = b.id")
+ val cachedFirstRelations = relations(cachedFirst)
+ assert(cachedFirstRelations.size == 2)
+ assert(cachedFirstRelations.forall(_.table eq cachedTable))
+ assert(cachedFirstRelations.map(_.options.get("split-size")).sorted ==
Seq("5", "9"))
+ assert(stateCatalog.loadTableCalls.count(_._2.get("snapshot") == "s1")
== 1)
+
+ stateCatalog.resetLoadTableCalls()
+ assert(cachedFirst.collect().map(_.getLong(0)).sorted.toSeq == Seq(1L,
2L))
+ assert(stateCatalog.loadTableCalls.isEmpty,
+ s"shared relation cache pin should avoid refresh reloads, got: " +
+ stateCatalog.loadTableCalls)
+
+ stateCatalog.resetLoadTableCalls()
+ val uncachedFirst = sql(s"SELECT a.id FROM $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) a JOIN $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 5) b ON a.id = b.id")
+ val uncachedFirstRelations = relations(uncachedFirst)
+ assert(uncachedFirstRelations.size == 2)
+ assert(uncachedFirstRelations.forall(_.table eq cachedTable))
+ assert(uncachedFirstRelations.map(_.options.get("split-size")).sorted
== Seq("5", "9"))
+ assert(stateCatalog.loadTableCalls.count(_._2.get("snapshot") == "s1")
== 1)
+
+ stateCatalog.resetLoadTableCalls()
+ assert(uncachedFirst.collect().map(_.getLong(0)).sorted.toSeq ==
Seq(1L, 2L))
+ assert(stateCatalog.loadTableCalls.isEmpty,
+ s"shared relation cache pin should avoid refresh reloads, got: " +
+ stateCatalog.loadTableCalls)
+ } finally {
+ otherStateCached.unpersist()
+ cached.unpersist()
+ }
+ }
+ }
+
+ test("shared relation cache makes table selection independent of reference
order") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ val ident = Identifier.of(Array("ns"), "table")
+ stateCatalog.alterTable(ident, TableChange.setProperty("version", "X"))
+
+ val cached = spark.read.option("split-size", "5").table(tableName)
+ cached.cache()
+ try {
+ cached.collect()
+ val versionX = cached.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.getOrElse(fail("expected a cached v2 relation"))
+ assert(versionX.properties().get("version") == "X")
+
+ // Simulate an external catalog update that does not invalidate
Spark's relation cache.
+ val versionY = stateCatalog.alterTable(
+ ident,
+ TableChange.setProperty("version", "Y"))
+ assert(versionY.properties().get("version") == "Y")
+ assert(versionY.id == versionX.id)
+ assert(versionY ne versionX)
+
+ def relations(df: DataFrame): Seq[DataSourceV2Relation] = {
+ df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if
r.options.containsKey("split-size") => r
+ }
+ }
+
+ val readA = spark.read.option("split-size", "5").table(tableName)
+ val readB = spark.read.option("split-size", "9").table(tableName)
+ assert(relations(readA).map(_.table) == Seq(versionX))
+ assert(relations(readB).map(_.table) == Seq(versionX))
+
+ val aThenB = sql(s"SELECT a.id FROM $tableName WITH (`split-size` = 5)
a " +
+ s"JOIN $tableName WITH (`split-size` = 9) b ON a.id = b.id")
+ val bThenA = sql(s"SELECT a.id FROM $tableName WITH (`split-size` = 9)
a " +
+ s"JOIN $tableName WITH (`split-size` = 5) b ON a.id = b.id")
+ assert(relations(aThenB).map(_.table) == Seq(versionX, versionX))
+ assert(relations(bThenA).map(_.table) == Seq(versionX, versionX))
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+
+ test("shared relation cache rejects a different declared table state") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ val cached = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(tableName)
+ cached.cache()
+ try {
+ cached.collect()
+ val cachedTable = cached.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.getOrElse(fail("expected a cached v2 relation"))
+
+ stateCatalog.resetLoadTableCalls()
+ val differentState = spark.read
+ .option("snapshot", "s2")
+ .option("split-size", "5")
+ .table(tableName)
+ val relation = differentState.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r
+ }.getOrElse(fail("expected a v2 relation"))
+
+ assert(relation.table ne cachedTable)
+ assert(relation.options.get("snapshot") == "s2")
+ assert(relation.options.get("split-size") == "5")
+ assert(stateCatalog.loadTableCalls.count(_._2.get("snapshot") == "s2")
== 1)
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+
+ test("streaming references participate in the query table-state cache") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ val df = sql(s"SELECT a.id FROM STREAM $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 5) a JOIN STREAM $tableName
" +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) b ON a.id = b.id")
+ val relations = df.queryExecution.analyzed.collect {
+ case r: StreamingRelationV2 if r.extraOptions.containsKey("snapshot")
=> r
+ }
+
+ assert(relations.size == 2)
+ assert(relations.map(_.extraOptions.get("split-size")).sorted ==
Seq("5", "9"))
+ assert(relations.head.table eq relations.last.table)
+ val stateLoads = stateCatalog.loadTableCalls.count(_._2.get("snapshot")
== "s1")
+ assert(stateLoads == 1, s"expected one streaming table load, got:
$stateLoads")
+ }
+ }
+
+ test("different table-state option values establish separate table pins") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ stateCatalog.singleArgLoads.set(0)
+
+ val df = sql(s"SELECT a.id FROM $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 5) a JOIN $tableName " +
+ s"WITH (`snapshot` = 's2', `split-size` = 9) b ON a.id = b.id")
+ val relations = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("snapshot") => r
+ }
+
+ assert(relations.size == 2)
+ assert(relations.map(_.options.get("snapshot")).sorted == Seq("s1",
"s2"))
+ assert(relations.head.table ne relations.last.table)
+ val loadedStates = stateCatalog.loadTableCalls
+ .map(_._2.get("snapshot"))
+ .filter(_ != null)
+ .sorted
+ assert(loadedStates == Seq("s1", "s2"))
+ }
+ }
+
+ test("persistent write targets bypass query-scoped cache lookups") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ val resolver = new RelationResolution(
+ spark.sessionState.catalogManager,
+ RelationCache.empty)
+ val options = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "5"))
+ val read = UnresolvedRelation(tableName.split("\\.").toSeq, options)
+ val write = read.requireWritePrivileges(Set(TableWritePrivilege.INSERT))
+
+ def resolve(relation: UnresolvedRelation): DataSourceV2Relation = {
+ resolver.resolveRelation(relation).flatMap(_.collectFirst {
+ case r: DataSourceV2Relation => r
+ }).getOrElse(fail(s"failed to resolve ${relation.name} as a v2
relation"))
+ }
+
+ AnalysisContext.withNewAnalysisContext {
+ val readRelation = resolve(read)
+ val writeRelation = resolve(write)
+
+ assert(readRelation.table ne writeRelation.table)
+ assert(AnalysisContext.get.tableCache.size == 1)
+ assert(AnalysisContext.get.relationCache.size == 1)
+ }
+
+ assert(stateCatalog.loadTableCalls.size == 2)
+ assert(stateCatalog.loadTableCalls.count(_._1.writePrivileges().isEmpty)
== 1)
+ assert(stateCatalog.loadTableCalls.count(
+ _._1.writePrivileges().contains(TableWritePrivilege.INSERT)) == 1)
+ assert(stateCatalog.loadTableCalls.forall(_._2.get("snapshot") == "s1"))
+ assert(stateCatalog.loadTableCalls.forall(_._2.size() == 1))
+ }
+ }
+
+ test("persistent write targets establish table pins for subsequent reads") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ stateCatalog.resetLoadTableCalls()
+ val resolver = new RelationResolution(
+ spark.sessionState.catalogManager,
+ RelationCache.empty)
+ val writeOptions = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "5"))
+ val readOptions = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "9"))
+ val write = UnresolvedRelation(tableName.split("\\.").toSeq,
writeOptions)
+ .requireWritePrivileges(Set(TableWritePrivilege.INSERT))
+ val read = UnresolvedRelation(tableName.split("\\.").toSeq, readOptions)
+
+ def resolve(relation: UnresolvedRelation): DataSourceV2Relation = {
+ resolver.resolveRelation(relation).flatMap(_.collectFirst {
+ case r: DataSourceV2Relation => r
+ }).getOrElse(fail(s"failed to resolve ${relation.name} as a v2
relation"))
+ }
+
+ AnalysisContext.withNewAnalysisContext {
+ val writeRelation = resolve(write)
+ val readRelation = resolve(read)
+
+ assert(writeRelation.table eq readRelation.table)
+ assert(writeRelation.options.get("split-size") == "5")
+ assert(readRelation.options.get("split-size") == "9")
+ assert(AnalysisContext.get.tableCache.size == 1)
+ assert(AnalysisContext.get.relationCache.size == 2)
+ }
+
+ assert(stateCatalog.loadTableCalls.size == 1)
+ assert(stateCatalog.loadTableCalls.head._1.writePrivileges().contains(
+ TableWritePrivilege.INSERT))
+ assertOnlySnapshotOptions(stateCatalog, "s1")
+ }
+ }
+
+ test("transaction V2TableReference skips shared lookup and writes bypass
query caches") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ val original = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(tableName)
+ .queryExecution
+ .analyzed
+ .collectFirst { case r: DataSourceV2Relation => r }
+ .getOrElse(fail("expected a v2 relation"))
+ val readRef = V2TableReference.createForTransaction(original)
+ val otherReadRef = V2TableReference.createForTransaction(original.copy(
+ options = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "9"))))
+ val writeRef = V2TableReference.createForWriteTarget(original)
+ var sharedRelationCacheLookups = 0
+ val sharedRelationCache: RelationCache = (_, _, _, _, _) => {
+ sharedRelationCacheLookups += 1
+ None
+ }
+ val resolver = new RelationResolution(
+ spark.sessionState.catalogManager,
+ sharedRelationCache)
+
+ stateCatalog.resetLoadTableCalls()
+ stateCatalog.singleArgLoads.set(0)
+ AnalysisContext.withNewAnalysisContext {
+ val readRelation =
resolver.resolveReference(readRef).asInstanceOf[DataSourceV2Relation]
+ val cachedReadRelation =
+ resolver.resolveReference(readRef).asInstanceOf[DataSourceV2Relation]
+ val otherReadRelation =
+
resolver.resolveReference(otherReadRef).asInstanceOf[DataSourceV2Relation]
+ val writeRelation =
resolver.resolveReference(writeRef).asInstanceOf[DataSourceV2Relation]
+ val readAfterWriteRelation =
+ resolver.resolveReference(readRef).asInstanceOf[DataSourceV2Relation]
+
+ assert(readRelation.table eq cachedReadRelation.table)
+ assert(readRelation.table eq otherReadRelation.table)
+ assert(readRelation.table eq readAfterWriteRelation.table)
+ assert(readRelation.table ne writeRelation.table)
+ assert(readRelation.options.get("split-size") == "5")
+ assert(otherReadRelation.options.get("split-size") == "9")
+ assert(AnalysisContext.get.tableCache.size == 1)
+ assert(AnalysisContext.get.relationCache.size == 2)
+ }
+
+ assert(sharedRelationCacheLookups == 0)
+ assert(stateCatalog.singleArgLoads.get() == 2)
+ assert(stateCatalog.loadTableCalls.size == 1)
+ assert(stateCatalog.loadTableCalls.head._2.get("snapshot") == "s1")
+ assert(stateCatalog.loadTableCalls.head._2.size() == 1)
+ }
+ }
+
+ test("nested view resolution shares the query table-state cache") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ withView("state_nested_view") {
+ sql(s"CREATE VIEW state_nested_view AS SELECT * FROM $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 5)")
+ stateCatalog.resetLoadTableCalls()
+
+ val df = sql(s"SELECT v.id FROM state_nested_view v JOIN $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) b ON v.id = b.id")
+ val relations = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("snapshot") =>
r
+ }
+
+ assert(relations.size == 2)
+ assert(relations.map(_.options.get("split-size")).sorted == Seq("5",
"9"))
+ assert(relations.head.table eq relations.last.table)
+ val stateLoads =
stateCatalog.loadTableCalls.count(_._2.get("snapshot") == "s1")
+ assert(stateLoads == 1, s"expected one nested-view table load, got:
$stateLoads")
+ }
+ }
+ }
+
+ test("temporary-view V2TableReference consults shared cache only for the
initial pin") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ val cached = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(tableName)
+ .queryExecution
+ .analyzed
+ .collectFirst { case r: DataSourceV2Relation => r }
+ .getOrElse(fail("expected a v2 relation"))
+ val initialRef = V2TableReference.createForTempView(cached,
Seq("state_view"))
+ val otherOptionsRef = V2TableReference.createForTempView(
+ cached.copy(options = new CaseInsensitiveStringMap(
+ java.util.Map.of("snapshot", "s1", "split-size", "9"))),
+ Seq("state_view"))
+ var sharedRelationCacheLookups = 0
+ val sharedRelationCache: RelationCache = (_, _, _, _, _) => {
+ sharedRelationCacheLookups += 1
+ Some(cached)
+ }
+ val resolver = new RelationResolution(
+ spark.sessionState.catalogManager,
+ sharedRelationCache)
+
+ stateCatalog.resetLoadTableCalls()
+ stateCatalog.singleArgLoads.set(0)
+ AnalysisContext.withNewAnalysisContext {
+ val initialRelation =
+
resolver.resolveReference(initialRef).asInstanceOf[DataSourceV2Relation]
+ val cachedRelation =
+
resolver.resolveReference(initialRef).asInstanceOf[DataSourceV2Relation]
+ val otherOptionsRelation =
+
resolver.resolveReference(otherOptionsRef).asInstanceOf[DataSourceV2Relation]
+
+ assert(initialRelation.table eq cached.table)
+ assert(cachedRelation.table eq cached.table)
+ assert(otherOptionsRelation.table eq cached.table)
+ assert(initialRelation.options.get("split-size") == "5")
+ assert(otherOptionsRelation.options.get("split-size") == "9")
+ assert(AnalysisContext.get.tableCache.size == 1)
+ assert(AnalysisContext.get.relationCache.size == 2)
+ }
+
+ assert(sharedRelationCacheLookups == 1)
+ assert(stateCatalog.singleArgLoads.get() == 1)
+ assert(stateCatalog.loadTableCalls.size == 1)
+ assert(stateCatalog.loadTableCalls.head._2.get("snapshot") == "s1")
+ assert(stateCatalog.loadTableCalls.head._2.size() == 1)
+ }
+ }
+
+ test("temporary-view re-resolution preserves the CacheManager table pin") {
+ withStateAwareTable { (stateCatalog, tableName) =>
+ withTempView("state_view") {
+ val cached = spark.read
+ .option("snapshot", "s1")
+ .option("split-size", "5")
+ .table(tableName)
+ cached.cache()
+ try {
+ cached.collect()
+ val cachedTable = cached.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.getOrElse(fail("expected a cached v2 relation"))
+ cached.createOrReplaceTempView("state_view")
+ stateCatalog.resetLoadTableCalls()
+ stateCatalog.singleArgLoads.set(0)
+
+ val df = sql(s"SELECT v.id FROM state_view v JOIN $tableName " +
+ s"WITH (`snapshot` = 's1', `split-size` = 9) b ON v.id = b.id")
+ val relations = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation if r.options.containsKey("snapshot")
=> r
+ }
+
+ assert(relations.size == 2)
+ assert(relations.map(_.options.get("split-size")).sorted == Seq("5",
"9"))
+ assert(relations.forall(_.table eq cachedTable))
+ assert(stateCatalog.singleArgLoads.get() == 1,
+ s"expected one table load while re-resolving the query, got: " +
+ stateCatalog.singleArgLoads.get())
+ assert(stateCatalog.loadTableCalls.size == 1)
+ assert(stateCatalog.loadTableCalls.head._2.get("snapshot") == "s1")
+ assert(stateCatalog.loadTableCalls.head._2.size() == 1)
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+ }
+
test("SPARK-58389: repeated references with the same options load the table
once") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
@@ -603,18 +1106,20 @@ class DataSourceV2OptionSuite extends
DatasourceV2SQLBase {
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")
+ val analysisLoads = inMemoryCatalog.loadTableCalls
+ assert(analysisLoads.size === 1,
+ s"expected a single loadTable for identical option bags, got:
$analysisLoads")
+ assert(analysisLoads.head._2.isEmpty,
+ s"expected scan options to be filtered during analysis, got:
$analysisLoads")
// 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")
+ val refreshLoads = inMemoryCatalog.loadTableCalls
+ assert(refreshLoads.size === 1,
+ s"expected refresh to load identical option bags once, got:
$refreshLoads")
+ assert(refreshLoads.head._2.isEmpty,
+ s"expected scan options to be filtered during refresh, got:
$refreshLoads")
}
}
@@ -663,7 +1168,7 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
- test("SPARK-58389: execution refresh does not reuse cached relation with
different options") {
+ test("execution refresh reuses cached table state with different scan
options") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
@@ -674,25 +1179,26 @@ class DataSourceV2OptionSuite extends
DatasourceV2SQLBase {
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.
+ // This catalog declares no state options, so analysis and refresh may
reuse the cached
+ // table while the relation keeps split-size=5 as its complete option
bag.
val df = spark.read.option("split-size", "5").table(t1).filter("id >
0")
- df.queryExecution.analyzed
+ val analyzedRelation = df.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r
+ }.getOrElse(fail("expected a v2 relation"))
+ assert(analyzedRelation.options.get("split-size") == "5")
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")
+ assert(inMemoryCatalog.loadTableCalls.isEmpty,
+ s"expected refresh to reuse the matching table state, got: " +
+ inMemoryCatalog.loadTableCalls)
} finally {
cached.unpersist()
}
}
}
- test("SPARK-58389: recaching preserves and forwards table options") {
+ test("recaching preserves relation options and filters table load options") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
@@ -703,14 +1209,15 @@ class DataSourceV2OptionSuite extends
DatasourceV2SQLBase {
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.
+ // Refreshing a cached table rebuilds its CacheManager entry. The
rebuilt relation retains
+ // the original options, but this catalog declares no state options
for the table reload.
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 recacheLoads = inMemoryCatalog.loadTableCalls
+ assert(recacheLoads.nonEmpty, "expected recache to reload the table")
+ assert(recacheLoads.forall(_._2.isEmpty),
+ s"expected recache to filter split-size, got: $recacheLoads")
val cacheManager = spark.sharedState.cacheManager
val sameOptions = spark.read.option("split-size", "5").table(t1)
@@ -731,7 +1238,7 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
- test("SPARK-58389: recaching a non-relation plan forwards table options") {
+ test("recaching a filtered plan filters table load options") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
sql(s"CREATE TABLE $t1 (id bigint, data string)")
@@ -748,10 +1255,9 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
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")
+ assert(recacheLoads.nonEmpty, "expected filtered recache to reload the
table")
+ assert(recacheLoads.forall(_._2.isEmpty),
+ s"expected filtered recache to filter split-size, got:
$recacheLoads")
val samePlan = spark.read.option("split-size",
"5").table(t1).filter("id > 0")
assert(spark.sharedState.cacheManager.lookupCachedData(samePlan).isDefined,
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 406e8885e163..891bc0490a89 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,19 +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") {
+ test("read options are preserved for scans but filtered from 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()
+ // The provider uses the options to identify the table, but this catalog
declares no
+ // table-state options. The relation still retains the complete option map
for scan planning.
+ val df = load("t1", Some(catalogName))
+ df.collect()
+ val relation = df.logicalPlan.asInstanceOf[DataSourceV2Relation]
+ assert(relation.options.get("name") === "t1")
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")
+ assert(opts.get.isEmpty)
}
private def load(
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 ee3eaa789ffc..1065c5baa585 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
@@ -184,7 +184,7 @@ class PlanResolutionSuite extends SharedSparkSession with
AnalysisTest {
val ident = invocation.getArguments()(0).asInstanceOf[Identifier]
val version = invocation.getArguments()(1).asInstanceOf[String]
(ident.name, version) match {
- case ("tab", "v1") => table
+ case ("tab", "v1" | "v2") => table
case ("tab", _) => throw new RuntimeException("Unknown version: " +
version)
case _ => throw new NoSuchTableException(Seq(ident.name))
}
@@ -3392,6 +3392,7 @@ class PlanResolutionSuite extends SharedSparkSession with
AnalysisTest {
AnalysisContext.withNewAnalysisContext {
val ctx = AnalysisContext.get
assert(ctx.relationCache.isEmpty)
+ assert(ctx.tableCache.isEmpty)
// create two unresolved relations without time travel
val unresolved1 = UnresolvedRelation(Seq("testcat", "tab"))
@@ -3408,6 +3409,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.timeTravelSpec.isEmpty)
+ assert(ctx.tableCache.size == 1)
// create unresolved relation with time travel spec
val timeTravelSpec = AsOfVersion("v1")
@@ -3422,6 +3424,17 @@ class PlanResolutionSuite extends SharedSparkSession
with AnalysisTest {
// after time travel resolution, cache should have 2 entries (with and
without time travel)
assert(ctx.relationCache.size == 2)
+ assert(ctx.tableCache.size == 2)
+
+ val otherTimeTravelSpec = AsOfVersion("v2")
+ val resolved4 = resolve(
+ UnresolvedRelation(Seq("testcat", "tab")),
+ Some(otherTimeTravelSpec))
+ assert(resolved4.timeTravelSpec.contains(otherTimeTravelSpec))
+
+ // Distinct parsed time-travel specs are distinct state pins even with
identical raw options.
+ assert(ctx.relationCache.size == 3)
+ assert(ctx.tableCache.size == 3)
}
}
@@ -3465,7 +3478,31 @@ class PlanResolutionSuite extends SharedSparkSession
with AnalysisTest {
}
}
- test("SPARK-58389: shared relation cache is reused only when the read's
options match") {
+ test("option-based time travel bypasses the shared relation cache") {
+ AnalysisContext.withNewAnalysisContext {
+ val ident = Identifier.of(Array.empty[String], "tab")
+ val cachedRelation = DataSourceV2Relation.create(table, Some(testCat),
Some(ident))
+ var sharedRelationCacheLookups = 0
+ val sharedRelationCache: RelationCache = (_, _, _, _, _) => {
+ sharedRelationCacheLookups += 1
+ Some(cachedRelation)
+ }
+ val rule = new RelationResolution(catalogManagerWithDefault,
sharedRelationCache)
+ val unresolved = UnresolvedRelation(
+ Seq("testcat", "tab"),
+ new CaseInsensitiveStringMap(java.util.Map.of("VeRsIoNaSoF", "v1")))
+
+ val resolved = rule.resolveRelation(unresolved) match {
+ case Some(AsDataSourceV2Relation(relation)) => relation
+ case other => fail(s"failed to resolve as v2 relation: $other")
+ }
+
+ assert(resolved.timeTravelSpec.contains(AsOfVersion("v1")))
+ assert(sharedRelationCacheLookups == 0)
+ }
+ }
+
+ test("shared relation cache reuses table state and preserves the read's full
options") {
val ident = Identifier.of(Array.empty[String], "tab")
val cachedTable = testCat.loadTable(ident)
@@ -3483,8 +3520,9 @@ class PlanResolutionSuite extends SharedSparkSession with
AnalysisTest {
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 sharedRelationCache: RelationCache =
+ (_, _, _, _, _) => Some(cachedRelationWith(cacheOpts))
+ val rule = new RelationResolution(catalogManagerWithDefault,
sharedRelationCache)
val unresolved =
UnresolvedRelation(Seq("testcat", "tab"), new
CaseInsensitiveStringMap(readOpts))
rule.resolveRelation(unresolved) match {
@@ -3502,15 +3540,130 @@ class PlanResolutionSuite extends SharedSparkSession
with AnalysisTest {
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(
+ // This catalog does not declare any state options, so a different full
option bag still
+ // reuses the cached table while the returned relation carries this read's
complete options.
+ val reusedWithDifferentOptions = 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")
+ assert(reusedWithDifferentOptions.options.get("split-size") === "5")
+ assert(!reusedWithDifferentOptions.options.containsKey("cachedOnly"))
+
assert(reusedWithDifferentOptions.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(4242L),
+ "scan-specific options should not prevent shared table reuse")
+ }
+
+ test("table-state cache consults shared relation cache only while
establishing the pin") {
+ def newTable(id: String): Table = {
+ val t = mock(classOf[Table])
+ when(t.id()).thenReturn(id)
+ when(t.name()).thenReturn("tab")
+ when(t.columns()).thenReturn(Array(Column.create("i", IntegerType)))
+ when(t.capabilities()).thenReturn(java.util.Set.of())
+ t
+ }
+
+ def options(splitSize: String): CaseInsensitiveStringMap = {
+ new CaseInsensitiveStringMap(
+ java.util.Map.of("state", "s1", "split-size", splitSize))
+ }
+
+ def run(
+ firstSplitSize: String,
+ secondSplitSize: String,
+ sharedRelationCacheEntry: (
+ TableCatalog,
+ Identifier,
+ Table,
+ Table) => Option[LogicalPlan]): (
+ DataSourceV2Relation,
+ DataSourceV2Relation,
+ Table,
+ Table,
+ Int,
+ Int) = {
+ val currentTable = newTable("table-id")
+ 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 loads = 0
+ when(catalog.loadTable(
+ any[Identifier],
+ any[TableContext],
+ any[CaseInsensitiveStringMap])).thenAnswer((_: InvocationOnMock) => {
+ loads += 1
+ currentTable
+ })
+
+ val manager = mock(classOf[CatalogManager])
+ when(manager.catalog(any())).thenReturn(catalog)
+ when(manager.v1SessionCatalog).thenReturn(v1SessionCatalog)
+ val ident = Identifier.of(Array.empty[String], "tab")
+ val sharedRelationCacheCandidate =
+ sharedRelationCacheEntry(catalog, ident, currentTable, cachedTable)
+ var sharedRelationCacheLookups = 0
+ val sharedRelationCache: RelationCache =
+ (_, _, _, _, _) => {
+ sharedRelationCacheLookups += 1
+ sharedRelationCacheCandidate
+ }
+ val resolver = new RelationResolution(manager, sharedRelationCache)
+
+ def resolveWith(splitSize: String): DataSourceV2Relation = {
+ val unresolved = UnresolvedRelation(Seq("statecat", "tab"),
options(splitSize))
+ resolver.resolveRelation(unresolved) match {
+ case Some(AsDataSourceV2Relation(relation)) => relation
+ case other => fail(s"failed to resolve as v2 relation: $other")
+ }
+ }
+
+ AnalysisContext.withNewAnalysisContext {
+ val first = resolveWith(firstSplitSize)
+ val second = resolveWith(secondSplitSize)
+ assert(AnalysisContext.get.tableCache.size == 1)
+ assert(AnalysisContext.get.relationCache.size == 2)
+ (first, second, currentTable, cachedTable, loads,
sharedRelationCacheLookups)
+ }
+ }
+
+ def cachedRelation(
+ catalog: TableCatalog,
+ ident: Identifier,
+ table: Table,
+ splitSize: String,
+ tag: Long): DataSourceV2Relation = {
+ val relation = DataSourceV2Relation.create(
+ table,
+ Some(catalog),
+ Some(ident),
+ options(splitSize))
+ relation.setTagValue(LogicalPlan.PLAN_ID_TAG, tag)
+ relation
+ }
+
+ // Situation A: the state-option cached match establishes the initial pin.
The later same-state
+ // lookup reuses that pin without consulting the shared relation cache.
+ val situationA = run("5", "9", { (catalog, ident, _, cachedTable) =>
+ Some(cachedRelation(catalog, ident, cachedTable, "5", 2L))
+ })
+ assert(situationA._1.table eq situationA._4)
+ assert(situationA._2.table eq situationA._4)
+ assert(situationA._1.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(2L))
+ assert(situationA._2.getTagValue(LogicalPlan.PLAN_ID_TAG).isEmpty)
+ assert(situationA._5 == 1)
+ assert(situationA._6 == 1)
+
+ // Situation B: scan-specific options differ on the first reference, but
the state options
+ // still match. The shared cache therefore establishes the same pin
regardless of reference
+ // order, and each returned relation keeps its own complete options.
+ val situationB = run("9", "5", { (catalog, ident, _, cachedTable) =>
+ Some(cachedRelation(catalog, ident, cachedTable, "5", 6L))
+ })
+ assert(situationB._1.table eq situationB._4)
+ assert(situationB._2.table eq situationB._4)
+ assert(situationB._1.options.get("split-size") == "9")
+ assert(situationB._2.options.get("split-size") == "5")
+ assert(situationB._1.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(6L))
+ assert(situationB._5 == 1)
+ assert(situationB._6 == 1)
}
private def resolve(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
