This is an automated email from the ASF dual-hosted git repository.
gengliangwang 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 a5b7f0dcfc1b [SPARK-59841][4.3][SQL] Align changelog state options and
metadata reuse
a5b7f0dcfc1b is described below
commit a5b7f0dcfc1b5babdbb6167323ea8a7e392a712c
Author: Gengliang Wang <[email protected]>
AuthorDate: Tue Sep 29 22:03:28 2026 -0700
[SPARK-59841][4.3][SQL] Align changelog state options and metadata reuse
### What changes were proposed in this pull request?
Backport #59119 to `branch-4.3` by cherry-picking
`a32c06b2cfe0275b5e6643810dbff8c91b74e5c6`.
Add `TableCatalog.changelogStateOptionKeys()`, filter catalog-level
changelog options, and reuse changelog metadata for reads with the same context
and state options. Preserve bounded CDC reads during refresh and temporary-view
access, and reload open-ended reads through the changelog API.
Resolve the `views.scala` import conflict by adding `ChangelogTable` and
retaining only `StaticSQLConf`; branch-4.3 has no corresponding `SQLConf`
usage. The new API retains `since 4.3.0`, as 4.3.0 has not yet been released.
### Why are the changes needed?
Bring the [SPARK-59841](https://issues.apache.org/jira/browse/SPARK-59841)
fixes to branch-4.3. Reads that differ only in scan options should reuse
changelog metadata, while different ranges or state options must remain
separate. Bounded CDC reads must retain their original range during refresh and
temporary-view access.
### Does this PR introduce _any_ user-facing change?
Yes, the same behavior changes as #59119. `loadChangelog` receives only
options declared by `changelogStateOptionKeys()`, which defaults to
`tableStateOptionKeys()`. Connectors must declare options required for metadata
loading; scan-only options remain available to scan builders.
### How was this patch tested?
- Compared the added and removed lines in all changed files against the
original commit; they match exactly.
- Passed `git diff --check`, added-source ASCII and line-length checks, and
error-conditions JSON parsing.
- Included the original regression tests. Local build and test suites were
not rerun for this backport. The original PR reports 323 passing tests plus
focused reruns; see #59119 for its validation details and dependency workaround.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-6)
Closes #59150 from gengliangwang/codex/spark-59841-branch-4.3.
Authored-by: Gengliang Wang <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 5 +
.../spark/sql/connector/catalog/Changelog.java | 5 +
.../spark/sql/connector/catalog/TableCatalog.java | 33 +-
.../spark/sql/catalyst/analysis/Analyzer.scala | 30 +-
.../sql/catalyst/analysis/ChangelogCacheKey.scala | 31 ++
.../sql/catalyst/analysis/CheckAnalysis.scala | 4 +-
.../sql/catalyst/analysis/RelationResolution.scala | 59 ++-
.../catalyst/analysis/ResolveChangelogTable.scala | 69 +---
.../analysis/UnresolveRelationsInTransaction.scala | 4 +-
.../{V2TableReference.scala => V2Reference.scala} | 119 ++++--
.../sql/connector/catalog/CatalogV2Util.scala | 23 +-
.../sql/connector/catalog/LookupCatalog.scala | 4 +-
.../spark/sql/errors/QueryCompilationErrors.scala | 10 +
.../execution/datasources/v2/ChangelogTable.scala | 75 +++-
.../catalyst/analysis/TableLookupCacheSuite.scala | 3 +-
.../sql/catalyst/analysis/V2ReferenceSuite.scala | 133 +++++++
.../sql/connector/catalog/CatalogV2UtilSuite.scala | 85 +++++
.../catalog/InMemoryChangelogCatalog.scala | 46 ++-
.../datasources/v2/ChangelogTableSuite.scala | 223 +++++++++++
.../apache/spark/sql/execution/CacheManager.scala | 10 +-
.../apache/spark/sql/execution/command/views.scala | 9 +-
.../datasources/v2/V2TableRefreshUtil.scala | 46 ++-
.../streaming/runtime/MicroBatchExecution.scala | 4 +-
.../sql/connector/ChangelogResolutionSuite.scala | 411 ++++++++++++++++++---
.../sql/connector/DataSourceV2OptionSuite.scala | 12 +-
25 files changed, 1205 insertions(+), 248 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index b7eca1e8dfc1..05bc9b98dfa9 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -3207,6 +3207,11 @@
"Detected incompatible changes to table <tableName> after
DataFrame/Dataset has been resolved and analyzed, meaning the underlying plan
is out of sync. Please, re-create DataFrame/Dataset before attempting to
execute the query again."
],
"subClass" : {
+ "CHANGELOG_METADATA_MISMATCH" : {
+ "message" : [
+ "Changelog metadata used for post-processing has changed:
<changedProperties>."
+ ]
+ },
"COLUMNS_MISMATCH" : {
"message" : [
"Data columns have changed:",
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Changelog.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Changelog.java
index 2c1dc896c1ba..2504784350b8 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Changelog.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Changelog.java
@@ -29,6 +29,11 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
* post-processing (carry-over removal, update detection, net change
computation) based on
* the properties declared by the connector.
* <p>
+ * Spark uses the connector's {@code equals} and {@code hashCode}
implementations when comparing
+ * changelog reads for plan reuse. Implementations that override these methods
must include all
+ * captured state affecting the schema, metadata, and data read. Spark also
compares the requested
+ * changelog context and each relation's read options.
+ * <p>
* The columns returned by {@link #columns()} must include the following
metadata columns:
* <ul>
* <li>{@code _change_type} (STRING) — the kind of change: {@code insert},
{@code delete},
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 36916f907160..cd8f01fb025d 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
@@ -124,6 +124,25 @@ public interface TableCatalog extends CatalogPlugin {
*/
default Set<String> tableStateOptionKeys() { return Set.of(); }
+ /**
+ * Returns the connector-specific option keys needed to load a changelog or
determine its
+ * metadata, such as branch, tag, or schema selection. Parameters that Spark
parses into
+ * {@link ChangelogContext}, such as the range and post-processing options,
must not be listed.
+ * <p>
+ * Spark passes only the declared options to {@link #loadChangelog} and uses
them, together with
+ * the changelog context, to identify reusable changelog state. The complete
user option map
+ * remains on each resolved relation for subsequent scan planning.
+ * <p>
+ * The default implementation returns {@link #tableStateOptionKeys()}. An
override declares the
+ * complete set of changelog state option keys; Spark does not add the table
state option keys.
+ * 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> changelogStateOptionKeys() { return
tableStateOptionKeys(); }
+
/**
* List the tables in a namespace from the catalog.
*
@@ -266,13 +285,17 @@ public interface TableCatalog extends CatalogPlugin {
* Load a {@link Changelog} for the given table, representing the row-level
changes within the
* range specified by {@code context}.
* <p>
- * The default implementation throws an analysis exception indicating that
the catalog does
- * not support CDC. Catalogs that support CDC must override this method.
+ * Spark passes only the options declared by {@link
#changelogStateOptionKeys()}. The range and
+ * post-processing parameters are supplied through {@code context}. The
complete user option
+ * map is retained for {@link
Changelog#newScanBuilder(CaseInsensitiveStringMap)}.
+ * <p>
+ * The default implementation throws {@link UnsupportedOperationException}
to indicate that
+ * the catalog does not support CDC. Catalogs that support CDC must override
this method.
*
* @param ident a table identifier
* @param context the CDC query context (range, deduplication mode, etc.)
- * @param options all options passed to the changelog query, including the
CDC-recognized
- * keys (range, deduplication mode, etc.) that are also
parsed into {@code context}
+ * @param stateOptions options declared to affect changelog state;
Spark-parsed parameters are
+ * provided through {@code context} instead
* @return a Changelog instance for the requested table and range
* @throws NoSuchTableException If the table doesn't exist
*
@@ -281,7 +304,7 @@ public interface TableCatalog extends CatalogPlugin {
default Changelog loadChangelog(
Identifier ident,
ChangelogContext context,
- CaseInsensitiveStringMap options) throws NoSuchTableException {
+ CaseInsensitiveStringMap stateOptions) throws NoSuchTableException {
throw new UnsupportedOperationException(name() + " does not support Change
Data Capture (CDC)");
}
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 8f11f21df672..6a1aa21f87bb 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
@@ -60,7 +60,7 @@ import
org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.connector.catalog.procedures.{BoundProcedure,
ProcedureParameter, UnboundProcedure}
import org.apache.spark.sql.connector.expressions.{FieldReference,
IdentityTransform}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.{PartitionOverwriteMode,
StoreAssignmentPolicy}
import org.apache.spark.sql.internal.connector.V1Function
@@ -138,6 +138,8 @@ object FakeV2SessionCatalog extends TableCatalog with
FunctionCatalog with Suppo
* @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 changelogCache A mapping from (catalog, identifier, changelog
context, state options) to
+ * changelog metadata. References retain their own scan
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,
@@ -159,6 +161,7 @@ case class AnalysisContext(
maxNestedViewDepth: Int = -1,
relationCache: mutable.Map[RelationCacheKey, LogicalPlan] =
mutable.Map.empty,
tableCache: mutable.Map[TableCacheKey, Table] = mutable.Map.empty,
+ changelogCache: mutable.Map[ChangelogCacheKey, ChangelogTable] =
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)`.
@@ -254,6 +257,7 @@ object AnalysisContext {
maxNestedViewDepth = maxNestedViewDepth,
relationCache = originContext.relationCache,
tableCache = originContext.tableCache,
+ changelogCache = originContext.changelogCache,
referredTempViewNames = viewDesc.viewReferredTempViewNames,
referredTempFunctionNames =
mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
referredTempVariableNames = viewDesc.viewReferredTempVariableNames,
@@ -1155,7 +1159,7 @@ class Analyzer(
case view: View if !view.child.resolved =>
ViewResolution
.resolve(view, options, resolveChild = executeSameContext,
checkAnalysis = checkAnalysis)
- // V2TableReference is a placeholder for DSv2 tables that needs to be
resolved to
+ // V2Reference is a placeholder for DSv2 tables or changelogs that must
be resolved to
// DataSourceV2Relation on each view access. Only dataframe temp view
may contain it
// as it stores resolved plans directly.
case view: View if view.isTempViewStoringAnalyzedPlan =>
@@ -1165,7 +1169,7 @@ class Analyzer(
case _ => plan
}
- // Unwrap temp views storing analyzed plans and resolve V2TableReference
nodes in the child.
+ // Unwrap temp views storing analyzed plans and resolve V2Reference nodes
in the child.
private def unwrapRelationPlan(plan: LogicalPlan): LogicalPlan = {
EliminateSubqueryAliases(plan) match {
case v: View if v.isTempViewStoringAnalyzedPlan =>
resolveTableReferencesInTempView(v.child)
@@ -1173,12 +1177,12 @@ class Analyzer(
}
}
- // Resolve V2TableReference nodes inside temp view plans. These are
created by
- // V2TableReference.createForTempView. We only need to resolve it when
returning
+ // Resolve V2Reference nodes inside temp view plans. These are created by
+ // V2Reference.createForTempView. We only need to resolve it when returning
// the plan of temp views (in resolveViews and unwrapRelationPlan).
private def resolveTableReferencesInTempView(plan: LogicalPlan):
LogicalPlan = {
plan.resolveOperatorsUp {
- case r: V2TableReference if
r.context.isInstanceOf[V2TableReference.TemporaryViewContext] =>
+ case r: V2Reference if
r.context.isInstanceOf[V2Reference.TemporaryViewContext] =>
relationResolution.resolveReference(r)
}
}
@@ -1195,7 +1199,7 @@ class Analyzer(
// Inserting into a file-based temporary view is allowed.
// (e.g., spark.read.parquet("path").createOrReplaceTempView("t").
// Thus, we need to look at the raw plan if `relation` is a temporary
view.
- // unwrapRelationPlan also resolves V2TableReference nodes in temp
view plans.
+ // unwrapRelationPlan also resolves V2Reference nodes in temp view
plans.
unwrapRelationPlan(relation) match {
case v: View if i.replaceCriteriaOpt.exists(_.isReplaceWhere) =>
throw
QueryCompilationErrors.writeIntoViewNotAllowedError(v.desc.identifier, i)
@@ -1206,11 +1210,11 @@ class Analyzer(
case write: V2StreamingWriteCommand =>
write.table match {
- case ref: V2TableReference =>
+ case ref: V2Reference =>
relationResolution.resolveReference(ref) match {
case r: NamedRelation => write.withNewTable(r)
case other => throw SparkException.internalError(
- s"Expected V2TableReference write target to resolve to a
NamedRelation, " +
+ s"Expected V2Reference write target to resolve to a
NamedRelation, " +
s"but got ${other.getClass.getName}")
}
case _ => write
@@ -1236,7 +1240,7 @@ class Analyzer(
case u: UnresolvedRelation =>
resolveRelation(u).map(resolveViews(_, u.options)).getOrElse(u)
- case r: V2TableReference =>
+ case r: V2Reference =>
relationResolution.resolveReference(r)
case r @ RelationTimeTravel(u: UnresolvedRelation, timestamp, version)
@@ -2672,13 +2676,13 @@ class Analyzer(
case a: FunctionTableSubqueryArgumentExpression if !a.plan.resolved =>
resolveSubQuery(a, outer)(
(plan, outerAttrs) => a.copy(plan = plan, outerAttrs = outerAttrs))
- // The subquery's plan is already resolved. Replace any
V2TableReferences without
+ // The subquery's plan is already resolved. Replace any V2References
without
// re-running any analyzer rules.
case se: SubqueryExpression
if se.plan.resolved &&
- se.plan.collectFirstWithSubqueries { case _: V2TableReference
=> () }.isDefined =>
+ se.plan.collectFirstWithSubqueries { case _: V2Reference => ()
}.isDefined =>
val newPlan = se.plan.transformWithSubqueries {
- case r: V2TableReference => relationResolution.resolveReference(r)
+ case r: V2Reference => relationResolution.resolveReference(r)
}
se.withNewPlan(newPlan)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ChangelogCacheKey.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ChangelogCacheKey.scala
new file mode 100644
index 000000000000..e4e1fef5125c
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ChangelogCacheKey.scala
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin,
ChangelogContext, Identifier}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Key for changelog metadata reused within an analysis or refresh pass. The
context identifies
+ * the requested range and processing mode; scan options remain on each
relation.
+ */
+private[sql] case class ChangelogCacheKey(
+ catalog: CatalogPlugin,
+ identifier: Identifier,
+ changelogContext: ChangelogContext,
+ stateOptions: CaseInsensitiveStringMap)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index d769ad258e80..42f0357c00bc 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -471,9 +471,9 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
context = u.origin.getQueryContext,
summary = u.origin.context.summary)
- case r: V2TableReference =>
+ case r: V2Reference =>
throw SparkException.internalError(
- s"V2TableReference should be resolved during analysis: ${r.name}")
+ s"V2Reference should be resolved during analysis: ${r.name}")
case u: UnresolvedInlineTable if
unresolvedInlineTableContainsScalarSubquery(u) =>
throw QueryCompilationErrors.inlineTableContainsScalarSubquery(u)
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 89e5db467c9e..762f1eb6e54d 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
@@ -63,6 +63,7 @@ class RelationResolution(
private def relationCache = AnalysisContext.get.relationCache
private def tableCache = AnalysisContext.get.tableCache
+ private def changelogCache = AnalysisContext.get.changelogCache
/**
* If we are resolving database objects (relations, functions, etc.) inside
views, we may need to
@@ -383,13 +384,9 @@ class RelationResolution(
expandIdentifier(u.multipartIdentifier) match {
case CatalogAndIdentifier(catalog, ident) =>
val tableCatalog = catalog.asTableCatalog
- val changelog = try {
- tableCatalog.loadChangelog(ident, ctx, u.options)
- } catch {
- case _: UnsupportedOperationException =>
- throw
QueryCompilationErrors.cdcNotSupportedError(tableCatalog.name())
- }
- val changelogTable = ChangelogTable(changelog, ctx)
+ val key = toChangelogCacheKey(catalog, ident, ctx, u.options)
+ val changelogTable = changelogCache.getOrElseUpdate(key,
+ ChangelogTable.load(tableCatalog, ident, ctx, key.stateOptions))
val relation = if (u.isStreaming) {
StreamingRelationV2(
None, changelogTable.name, changelogTable, u.options,
@@ -513,16 +510,36 @@ class RelationResolution(
}
}
- def resolveReference(ref: V2TableReference): LogicalPlan = {
- val relation = if (ref.context.cacheable) {
- getOrLoadRelation(ref)
- } else {
- loadRelation(ref)
+ def resolveReference(ref: V2Reference): LogicalPlan = {
+ val relation = ref match {
+ case changelogRef: V2ChangelogReference =>
+ resolveChangelogReference(changelogRef)
+ case tableRef: V2TableReference if tableRef.context.cacheable =>
+ getOrLoadRelation(tableRef)
+ case tableRef: V2TableReference =>
+ loadRelation(tableRef)
}
val planId = ref.getTagValue(LogicalPlan.PLAN_ID_TAG)
cloneWithPlanId(relation, planId)
}
+ private def resolveChangelogReference(ref: V2ChangelogReference):
DataSourceV2Relation = {
+ val catalog = catalogManager.catalog(ref.catalog.name).asTableCatalog
+ val captured = ref.changelog
+ val key = toChangelogCacheKey(
+ catalog, ref.identifier, captured.changelogContext, ref.options)
+ def load(): ChangelogTable = {
+ ChangelogTable.load(catalog, ref.identifier, captured.changelogContext,
key.stateOptions)
+ }
+ val current = if (ref.context.cacheable) {
+ changelogCache.getOrElseUpdate(key, load())
+ } else {
+ load()
+ }
+ captured.validateRefresh(current)
+ createRelation(ref, catalog, current.copy(resolved = captured.resolved))
+ }
+
private def getOrLoadRelation(ref: V2TableReference): LogicalPlan = {
val key = toCacheKey(ref.catalog, ref.identifier, None, ref.options)
relationCache.get(key) match {
@@ -575,10 +592,10 @@ class RelationResolution(
}
private def createRelation(
- ref: V2TableReference,
+ ref: V2Reference,
resolvedCatalog: TableCatalog,
table: Table): DataSourceV2Relation = {
- V2TableReferenceUtils.validateLoadedTable(table, ref)
+ V2ReferenceUtils.validateLoadedTable(table, ref)
DataSourceV2Relation(
table = table,
output = ref.output,
@@ -590,7 +607,7 @@ class RelationResolution(
private def adaptCachedRelation(cached: LogicalPlan, ref: V2TableReference):
LogicalPlan = {
cached transform {
case r: DataSourceV2Relation if matchesReference(r, ref) =>
- V2TableReferenceUtils.validateLoadedTable(r.table, ref)
+ V2ReferenceUtils.validateLoadedTable(r.table, ref)
r.copy(output = ref.output, options = ref.options)
}
}
@@ -633,6 +650,18 @@ class RelationResolution(
CatalogV2Util.extractTableStateOptions(catalog, options))
}
+ private def toChangelogCacheKey(
+ catalog: CatalogPlugin,
+ ident: Identifier,
+ context: ChangelogContext,
+ options: CaseInsensitiveStringMap): ChangelogCacheKey = {
+ ChangelogCacheKey(
+ catalog,
+ ident,
+ context,
+ CatalogV2Util.extractChangelogStateOptions(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/ResolveChangelogTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala
index ed9333d3a27a..1822e393903c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala
@@ -35,8 +35,7 @@ import
org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
-import org.apache.spark.sql.connector.catalog.{Changelog, ChangelogContext}
-import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.connector.catalog.Changelog
import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
import org.apache.spark.sql.streaming.{OutputMode, StatefulProcessor}
import org.apache.spark.sql.types.{BooleanType, DataType, IntegerType,
MetadataBuilder, StringType, StructField, StructType}
@@ -44,8 +43,7 @@ import org.apache.spark.unsafe.types.CalendarInterval
/**
* Post-processes a resolved [[ChangelogTable]] read to apply CDC option
semantics
- * (carry-over removal, update detection, net change computation) and to
enforce
- * supported option combinations.
+ * (carry-over removal, update detection, net change computation).
*
* Fires after [[ResolveRelations]] has wrapped the connector's [[Changelog]]
in a
* [[ChangelogTable]]. Both batch ([[DataSourceV2Relation]]) and streaming
@@ -123,15 +121,14 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp
{
case rel @ DataSourceV2Relation(table: ChangelogTable, _, _, _, _, _) if
!table.resolved =>
val changelog = table.changelog
- val req = evaluateRequirements(changelog, table.changelogContext)
val resolvedRel = rel.copy(table = table.copy(resolved = true))
var updatedRel: LogicalPlan = resolvedRel
- if (req.requiresCarryOverRemoval || req.requiresUpdateDetection) {
+ if (table.requiresCarryoverRemoval || table.requiresUpdateDetection) {
updatedRel = addRowLevelPostProcessing(
- resolvedRel, changelog, req.requiresCarryOverRemoval,
req.requiresUpdateDetection)
+ resolvedRel, changelog, table.requiresCarryoverRemoval,
table.requiresUpdateDetection)
}
- if (req.requiresNetChanges) {
+ if (table.requiresNetChangeCollapse) {
// Resolve rowId against the bare DataSourceV2Relation.
V2ExpressionUtils.resolveRefs
// requires a V2-shaped plan; addRowLevelPostProcessing may have
wrapped the relation
// in Project/Window, which would break resolution against
`updatedRel`. Catalyst
@@ -147,14 +144,13 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
case rel @ StreamingRelationV2(_, _, table: ChangelogTable, _, _, _, _, _,
_)
if !table.resolved =>
val changelog = table.changelog
- val req = evaluateRequirements(changelog, table.changelogContext)
val resolvedRel = rel.copy(table = table.copy(resolved = true))
var updatedRel: LogicalPlan = resolvedRel
- if (req.requiresCarryOverRemoval || req.requiresUpdateDetection) {
+ if (table.requiresCarryoverRemoval || table.requiresUpdateDetection) {
updatedRel = addStreamingRowLevelPostProcessing(
- resolvedRel, changelog, req.requiresCarryOverRemoval,
req.requiresUpdateDetection)
+ resolvedRel, changelog, table.requiresCarryoverRemoval,
table.requiresUpdateDetection)
}
- if (req.requiresNetChanges) {
+ if (table.requiresNetChangeCollapse) {
// Resolve the rowId references against `updatedRel` (the
post-row-level plan)
// rather than the bare `resolvedRel`. The streaming row-level rewrite
uses
// Aggregate + Generate(Inline), neither of which preserves the
original
@@ -169,55 +165,6 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
updatedRel
}
- //
---------------------------------------------------------------------------
- // Option validation & Requirement Computation
- //
---------------------------------------------------------------------------
-
- /**
- * Captures which post-processing passes a CDC query requires, derived from
the
- * user-provided [[ChangelogContext]] and the connector-declared
[[Changelog]]
- * capability flags.
- */
- private case class PostProcessingRequirements(
- requiresCarryOverRemoval: Boolean,
- requiresUpdateDetection: Boolean,
- requiresNetChanges: Boolean) {
- def needsAny: Boolean =
- requiresCarryOverRemoval || requiresUpdateDetection || requiresNetChanges
- }
-
- /**
- * Validates CDC option/capability combinations and computes which
post-processing
- * passes are required. Throws an [[org.apache.spark.sql.AnalysisException]]
for
- * unsupported or contradictory combinations (currently: `computeUpdates`
with
- * surfaced carry-overs but no carry-over removal).
- */
- private def evaluateRequirements(
- changelog: Changelog,
- context: ChangelogContext): PostProcessingRequirements = {
- val requiresCarryOverRemoval =
- context.deduplicationMode() != ChangelogContext.DeduplicationMode.NONE &&
- changelog.containsCarryoverRows()
- val requiresUpdateDetection =
- context.computeUpdates() && changelog.representsUpdateAsDeleteAndInsert()
- val requiresNetChanges =
- context.deduplicationMode() ==
ChangelogContext.DeduplicationMode.NET_CHANGES &&
- changelog.containsIntermediateChanges()
-
- // If carry-overs are surfaced and update detection is enabled without
carry-over
- // removal, carry-overs would be falsely classified as updates, leading to
wrong
- // results. Hence we throw.
- if (requiresUpdateDetection &&
- changelog.containsCarryoverRows() &&
- context.deduplicationMode() ==
ChangelogContext.DeduplicationMode.NONE) {
- throw QueryCompilationErrors.cdcUpdateDetectionRequiresCarryOverRemoval(
- changelog.name())
- }
-
- PostProcessingRequirements(
- requiresCarryOverRemoval, requiresUpdateDetection, requiresNetChanges)
- }
-
//
---------------------------------------------------------------------------
// Row Level Post Processing (Update Detection & Carry-over Removal)
//
---------------------------------------------------------------------------
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnresolveRelationsInTransaction.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnresolveRelationsInTransaction.scala
index b0497897bbac..c41ef2119a1a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnresolveRelationsInTransaction.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnresolveRelationsInTransaction.scala
@@ -25,7 +25,7 @@ import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
/**
* When a transaction is active, converts resolved [[DataSourceV2Relation]]
nodes back to
- * [[V2TableReference]] placeholders for all relations loaded by a catalog
with the same
+ * [[V2Reference]] placeholders for all relations loaded by a catalog with the
same
* name as the transaction catalog.
*
* This forces re-resolution of those relations against the transaction's
catalog, which
@@ -61,7 +61,7 @@ class UnresolveRelationsInTransaction(val catalogManager:
CatalogManager)
// cached/pre-analyzed DataFrames or temp views) are still rewritten.
plan.transformWithSubqueries {
case r: DataSourceV2Relation if isLoadedFromCatalog(r, catalog) =>
- V2TableReference.createForTransaction(r)
+ V2Reference.createForTransaction(r)
}
}
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/V2Reference.scala
similarity index 71%
rename from
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V2TableReference.scala
rename to
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V2Reference.scala
index 2d4ef18b5f60..7e864bedebb7 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/V2Reference.scala
@@ -19,11 +19,11 @@ package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.analysis.V2TableReference.Context
+import org.apache.spark.sql.catalyst.analysis.V2Reference.Context
+import org.apache.spark.sql.catalyst.analysis.V2Reference.TemporaryViewContext
+import org.apache.spark.sql.catalyst.analysis.V2Reference.TransactionContext
+import org.apache.spark.sql.catalyst.analysis.V2Reference.WriteTargetContext
import org.apache.spark.sql.catalyst.analysis.V2TableReference.TableInfo
-import
org.apache.spark.sql.catalyst.analysis.V2TableReference.TemporaryViewContext
-import
org.apache.spark.sql.catalyst.analysis.V2TableReference.TransactionContext
-import
org.apache.spark.sql.catalyst.analysis.V2TableReference.WriteTargetContext
import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.catalyst.plans.logical.LeafNode
import org.apache.spark.sql.catalyst.plans.logical.Statistics
@@ -37,40 +37,43 @@ import org.apache.spark.sql.connector.catalog.Table
import org.apache.spark.sql.connector.catalog.TableCatalog
import org.apache.spark.sql.connector.catalog.V2TableUtil
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import
org.apache.spark.sql.util.SchemaValidationMode.{ALLOW_NEW_TOP_LEVEL_FIELDS,
PROHIBIT_CHANGES}
import org.apache.spark.util.ArrayImplicits._
/**
- * A reference to a V2 table.
+ * A reference to a V2 table or changelog.
*
* References are placeholders for the latest table metadata and are replaced
with actual table
* versions during analysis, allowing Spark to reload tables with up-to-date
metadata. The newly
* loaded table metadata is validated against the original metadata depending
on the context.
- * For instance, temporary views with fully resolved logical plans don't allow
schema changes
- * in underlying tables.
+ * For instance, temporary views with fully resolved logical plans retain
their captured schema.
*/
-private[sql] case class V2TableReference private(
- catalog: TableCatalog,
- identifier: Identifier,
- options: CaseInsensitiveStringMap,
- info: TableInfo,
- output: Seq[AttributeReference],
- context: Context)
+private[sql] sealed abstract class V2Reference
extends LeafNode with MultiInstanceRelation with NamedRelation {
- override def name: String = V2TableUtil.toQualifiedName(catalog, identifier)
+ def catalog: TableCatalog
- override def newInstance(): V2TableReference = {
- copy(output = output.map(_.newInstance()))
- }
+ def identifier: Identifier
+
+ def options: CaseInsensitiveStringMap
+
+ def info: TableInfo
+
+ override def output: Seq[AttributeReference]
+
+ def context: Context
+
+ override def newInstance(): V2Reference
+
+ override def name: String = V2TableUtil.toQualifiedName(catalog, identifier)
override def computeStats(): Statistics = Statistics.DUMMY
override def simpleString(maxFields: Int): String = {
val outputString = truncatedString(output, "[", ", ", "]", maxFields)
- s"TableReference$outputString $name"
+ s"${nodeName.stripPrefix("V2")}$outputString $name"
}
def toRelation(table: Table): DataSourceV2Relation = {
@@ -78,12 +81,47 @@ private[sql] case class V2TableReference private(
}
}
+/** A reference that reloads ordinary table metadata. */
+private[sql] case class V2TableReference private[analysis](
+ catalog: TableCatalog,
+ identifier: Identifier,
+ options: CaseInsensitiveStringMap,
+ info: TableInfo,
+ output: Seq[AttributeReference],
+ context: Context) extends V2Reference {
+
+ override def newInstance(): V2TableReference = {
+ copy(output = output.map(_.newInstance()))
+ }
+}
+
private[sql] object V2TableReference {
case class TableInfo(
tableId: Option[String],
columns: Seq[Column],
metadataColumns: Seq[MetadataColumn])
+}
+
+/**
+ * A reference that retains the requested changelog context and captured
post-processing metadata.
+ * Re-resolution reloads the changelog while preserving the analyzed CDC
operators and schema.
+ */
+private[sql] case class V2ChangelogReference private[analysis](
+ catalog: TableCatalog,
+ identifier: Identifier,
+ options: CaseInsensitiveStringMap,
+ changelog: ChangelogTable,
+ info: TableInfo,
+ output: Seq[AttributeReference],
+ context: Context) extends V2Reference {
+
+ override def newInstance(): V2ChangelogReference = {
+ copy(output = output.map(_.newInstance()))
+ }
+}
+
+private[sql] object V2Reference {
sealed trait Context {
/** Whether re-resolution may reuse the per-query relation cache. */
@@ -111,39 +149,42 @@ private[sql] object V2TableReference {
val sharedCacheable = false
}
- def createForTempView(relation: DataSourceV2Relation, viewName:
Seq[String]): V2TableReference = {
+ def createForTempView(relation: DataSourceV2Relation, viewName:
Seq[String]): V2Reference = {
create(relation, TemporaryViewContext(viewName))
}
- // V2TableReference nodes in the transaction context are produced by
+ // V2Reference nodes in the transaction context are produced by
// UnresolveRelationsInTransaction which unresolves already resolved
relations.
- def createForTransaction(relation: DataSourceV2Relation): V2TableReference =
{
+ def createForTransaction(relation: DataSourceV2Relation): V2Reference = {
create(relation, TransactionContext)
}
- def createForWriteTarget(relation: DataSourceV2Relation): V2TableReference =
{
+ def createForWriteTarget(relation: DataSourceV2Relation): V2Reference = {
create(relation, WriteTargetContext)
}
- private def create(relation: DataSourceV2Relation, context: Context):
V2TableReference = {
- val ref = V2TableReference(
- relation.catalog.get.asTableCatalog,
- relation.identifier.get,
- relation.options,
- TableInfo(
- tableId = Option(relation.table.id),
- columns = relation.table.columns.toImmutableArraySeq,
- metadataColumns = V2TableUtil.extractMetadataColumns(relation)),
- relation.output,
- context)
+ private def create(relation: DataSourceV2Relation, context: Context):
V2Reference = {
+ val catalog = relation.catalog.get.asTableCatalog
+ val identifier = relation.identifier.get
+ val info = TableInfo(
+ tableId = Option(relation.table.id),
+ columns = relation.table.columns.toImmutableArraySeq,
+ metadataColumns = V2TableUtil.extractMetadataColumns(relation))
+ val ref = relation.table match {
+ case changelog: ChangelogTable =>
+ V2ChangelogReference(catalog, identifier, relation.options, changelog,
info,
+ relation.output, context)
+ case _ =>
+ V2TableReference(catalog, identifier, relation.options, info,
relation.output, context)
+ }
ref.copyTagsFrom(relation)
ref
}
}
-private[sql] object V2TableReferenceUtils extends SQLConfHelper {
+private[sql] object V2ReferenceUtils extends SQLConfHelper {
- def validateLoadedTable(table: Table, ref: V2TableReference): Unit = {
+ def validateLoadedTable(table: Table, ref: V2Reference): Unit = {
ref.context match {
case ctx: TemporaryViewContext =>
validateLoadedTableInTempView(table, ref, ctx)
@@ -154,7 +195,7 @@ private[sql] object V2TableReferenceUtils extends
SQLConfHelper {
}
}
- private def validateNoChanges(table: Table, ref: V2TableReference): Unit = {
+ private def validateNoChanges(table: Table, ref: V2Reference): Unit = {
// Make sure the table was not dropped and recreated.
ref.info.tableId.foreach(V2TableUtil.validateTableId(ref.name, _, table))
@@ -183,7 +224,7 @@ private[sql] object V2TableReferenceUtils extends
SQLConfHelper {
private def validateLoadedTableInTempView(
table: Table,
- ref: V2TableReference,
+ ref: V2Reference,
ctx: TemporaryViewContext): Unit = {
val tableName = ref.identifier.toQualifiedNameParts(ref.catalog)
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 39dc1b853e05..de170375effb 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
@@ -505,11 +505,26 @@ private[sql] object CatalogV2Util {
def extractTableStateOptions(
catalog: CatalogPlugin,
options: CaseInsensitiveStringMap): CaseInsensitiveStringMap = {
- val stateKeys = catalog.asTableCatalog.tableStateOptionKeys.asScala
- .map(_.toLowerCase(Locale.ROOT))
- .toSet
+ extractStateOptions(catalog.asTableCatalog.tableStateOptionKeys, options)
+ }
+
+ /**
+ * Extracts the options passed to `loadChangelog`. Together with the
changelog context, these
+ * options identify changelog state independently of the complete options
used to plan each scan.
+ */
+ def extractChangelogStateOptions(
+ catalog: CatalogPlugin,
+ options: CaseInsensitiveStringMap): CaseInsensitiveStringMap = {
+ extractStateOptions(catalog.asTableCatalog.changelogStateOptionKeys,
options)
+ }
+
+ private def extractStateOptions(
+ keys: util.Set[String],
+ options: CaseInsensitiveStringMap): CaseInsensitiveStringMap = {
+ val stateKeys = keys.asScala.map(_.toLowerCase(Locale.ROOT)).toSet
+ // Preserve the effective value even if differently cased aliases have
different values.
val projected = options.asCaseSensitiveMap().asScala.collect {
- case (key, value) if stateKeys.contains(key.toLowerCase(Locale.ROOT)) =>
key -> value
+ case (key, _) if stateKeys.contains(key.toLowerCase(Locale.ROOT)) => key
-> options.get(key)
}.toMap
new CaseInsensitiveStringMap(projected.asJava)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/LookupCatalog.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/LookupCatalog.scala
index 14c066373032..9cf9205f41a4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/LookupCatalog.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/LookupCatalog.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.connector.catalog
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases,
UnresolvedIdentifier, UnresolvedRelation, V2TableReference}
+import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases,
UnresolvedIdentifier, UnresolvedRelation, V2Reference}
import org.apache.spark.sql.catalyst.plans.logical.TransactionalWrite
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
@@ -173,7 +173,7 @@ private[sql] trait LookupCatalog extends Logging {
Some(c)
case UnresolvedIdentifier(CatalogAndIdentifier(c:
TransactionalCatalogPlugin, _), _) =>
Some(c)
- case ref: V2TableReference =>
+ case ref: V2Reference =>
ref.catalog match {
case c: TransactionalCatalogPlugin => Some(c)
case _ => None
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 0f536cbeb303..5d73e53fba9d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -2378,6 +2378,16 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
"currentTableId" -> currentTableId))
}
+ def changelogChangedAfterAnalysis(
+ tableName: String,
+ changedProperties: Seq[String]): Throwable = {
+ new AnalysisException(
+ errorClass =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.CHANGELOG_METADATA_MISMATCH",
+ messageParameters = Map(
+ "tableName" -> toSQLId(tableName),
+ "changedProperties" -> changedProperties.mkString(", ")))
+ }
+
def columnsChangedAfterAnalysis(
tableName: String,
errors: Seq[String]): Throwable = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala
index ec45f1f37317..35496830289f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTable.scala
@@ -19,7 +19,8 @@ package org.apache.spark.sql.execution.datasources.v2
import java.util.{EnumSet => JEnumSet, Set => JSet}
-import org.apache.spark.sql.connector.catalog.{Changelog, ChangelogContext,
Column, SupportsRead, Table, TableCapability}
+import org.apache.spark.sql.connector.catalog.{Changelog, ChangelogContext,
Column, Identifier, SupportsRead, Table, TableCapability, TableCatalog}
+import org.apache.spark.sql.connector.catalog.ChangelogRange.{TimestampRange,
UnboundedRange, VersionRange}
import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ,
MICRO_BATCH_READ}
import org.apache.spark.sql.connector.read.ScanBuilder
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -38,9 +39,37 @@ case class ChangelogTable(
changelogContext: ChangelogContext,
resolved: Boolean = false) extends Table with SupportsRead {
- // Validate that the connector returned a schema with the required CDC
metadata columns
- // and correct types.
+ // Capture the metadata used to build the analyzer's CDC post-processing
operators.
+ private val containsCarryoverRows = changelog.containsCarryoverRows()
+ private val containsIntermediateChanges =
changelog.containsIntermediateChanges()
+ private val representsUpdateAsDeleteAndInsert =
changelog.representsUpdateAsDeleteAndInsert()
+
+ private[sql] val requiresCarryoverRemoval =
+ changelogContext.deduplicationMode() !=
ChangelogContext.DeduplicationMode.NONE &&
+ containsCarryoverRows
+ private[sql] val requiresUpdateDetection =
+ changelogContext.computeUpdates() && representsUpdateAsDeleteAndInsert
+ private[sql] val requiresNetChangeCollapse =
+ changelogContext.deduplicationMode() ==
ChangelogContext.DeduplicationMode.NET_CHANGES &&
+ containsIntermediateChanges
+
+ // Validate the schema and option combinations before capturing the
referenced fields.
ChangelogTable.validateSchema(changelog)
+ if (requiresUpdateDetection && containsCarryoverRows &&
!requiresCarryoverRemoval) {
+ throw
QueryCompilationErrors.cdcUpdateDetectionRequiresCarryOverRemoval(name)
+ }
+
+ private val rowId = if (
+ requiresCarryoverRemoval || requiresUpdateDetection ||
requiresNetChangeCollapse) {
+ changelog.rowId().toVector.map(_.fieldNames().toVector)
+ } else {
+ Vector.empty
+ }
+ private val rowVersion = if (requiresCarryoverRemoval) {
+ Option(changelog.rowVersion()).map(_.fieldNames().toVector)
+ } else {
+ None
+ }
override def name: String = changelog.name
@@ -51,10 +80,50 @@ case class ChangelogTable(
}
override def capabilities: JSet[TableCapability] = JEnumSet.of(BATCH_READ,
MICRO_BATCH_READ)
+
+ private[sql] def isBounded: Boolean = changelogContext.range() match {
+ case range: VersionRange => range.endingVersion().isPresent
+ case range: TimestampRange => range.endingTimestamp().isPresent
+ case _: UnboundedRange => false
+ }
+
+ /** Checks that refreshing the changelog preserves the already analyzed CDC
rewrites. */
+ def validateRefresh(current: ChangelogTable): Unit = {
+ if (resolved) {
+ val changedProperties = Seq(
+ "containsCarryoverRows" ->
+ (requiresCarryoverRemoval != current.requiresCarryoverRemoval),
+ "containsIntermediateChanges" ->
+ (requiresNetChangeCollapse != current.requiresNetChangeCollapse),
+ "representsUpdateAsDeleteAndInsert" ->
+ (requiresUpdateDetection != current.requiresUpdateDetection),
+ "rowId" -> (rowId != current.rowId),
+ "rowVersion" -> (rowVersion != current.rowVersion)).collect {
+ case (property, true) => property
+ }
+ if (changedProperties.nonEmpty) {
+ throw QueryCompilationErrors.changelogChangedAfterAnalysis(name,
changedProperties)
+ }
+ }
+ }
}
object ChangelogTable {
+ def load(
+ catalog: TableCatalog,
+ ident: Identifier,
+ context: ChangelogContext,
+ stateOptions: CaseInsensitiveStringMap): ChangelogTable = {
+ val changelog = try {
+ catalog.loadChangelog(ident, context, stateOptions)
+ } catch {
+ case _: UnsupportedOperationException =>
+ throw QueryCompilationErrors.cdcNotSupportedError(catalog.name())
+ }
+ ChangelogTable(changelog, context)
+ }
+
private[v2] def validateSchema(cl: Changelog): Unit = {
val byName = cl.columns.map(c => c.name -> c).toMap
def check(name: String, expected: DataType*): Unit = {
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 c078504ea33a..3e176d014934 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
@@ -120,7 +120,7 @@ class TableLookupCacheSuite extends AnalysisTest with
Matchers {
}
}
- test("nested view analysis shares both query-scoped caches") {
+ test("nested view analysis shares query-scoped caches") {
AnalysisContext.withNewAnalysisContext {
val outer = AnalysisContext.get
val viewDesc = CatalogTable(
@@ -133,6 +133,7 @@ class TableLookupCacheSuite extends AnalysisTest with
Matchers {
AnalysisContext.withAnalysisContext(viewDesc) {
assert(AnalysisContext.get.relationCache eq outer.relationCache)
assert(AnalysisContext.get.tableCache eq outer.tableCache)
+ assert(AnalysisContext.get.changelogCache eq outer.changelogCache)
}
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2ReferenceSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2ReferenceSuite.scala
new file mode 100644
index 000000000000..9205b882ad50
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2ReferenceSuite.scala
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import java.util
+
+import org.apache.spark.sql.AnalysisException
+import
org.apache.spark.sql.catalyst.analysis.V2Reference.{TemporaryViewContext,
TransactionContext, WriteTargetContext}
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Statistics}
+import org.apache.spark.sql.connector.catalog.{ChangelogContext, Column,
Identifier, InMemoryChangelog, InMemoryTable, InMemoryTableCatalog, Table}
+import
org.apache.spark.sql.connector.catalog.ChangelogContext.DeduplicationMode
+import org.apache.spark.sql.connector.catalog.ChangelogRange.UnboundedRange
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
+import org.apache.spark.sql.types.LongType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class V2ReferenceSuite extends PlanTest {
+
+ private val catalog = new InMemoryTableCatalog
+ catalog.initialize("catalog", CaseInsensitiveStringMap.empty())
+ private val identifier = Identifier.of(Array("ns"), "table")
+ private val options = new CaseInsensitiveStringMap(util.Map.of("branch",
"main"))
+ private val columns = Array(Column.create("id", LongType))
+ private val changelogContext = new ChangelogContext(
+ new UnboundedRange(), DeduplicationMode.NONE, false)
+
+ private def table(isChangelog: Boolean, dataColumns: Array[Column] =
columns): Table = {
+ if (isChangelog) {
+ ChangelogTable(
+ new InMemoryChangelog("changes", dataColumns, Seq.empty),
+ changelogContext,
+ resolved = true)
+ } else {
+ new InMemoryTable("table", dataColumns, Array.empty, util.Map.of(), id =
"table-id")
+ }
+ }
+
+ private def relation(table: Table): DataSourceV2Relation = {
+ DataSourceV2Relation.create(table, Some(catalog), Some(identifier),
options)
+ }
+
+ gridTest("reference factories preserve captured metadata and choose the
reference type")(
+ for {
+ isChangelog <- Seq(false, true)
+ context <- Seq(TemporaryViewContext(Seq("view")), TransactionContext,
WriteTargetContext)
+ } yield (isChangelog, context)) { case (isChangelog, context) =>
+ val original = relation(table(isChangelog))
+ original.setTagValue(LogicalPlan.PLAN_ID_TAG, 42L)
+ val ref = context match {
+ case TemporaryViewContext(viewName) =>
V2Reference.createForTempView(original, viewName)
+ case TransactionContext => V2Reference.createForTransaction(original)
+ case WriteTargetContext => V2Reference.createForWriteTarget(original)
+ }
+
+ assert(ref.isInstanceOf[V2ChangelogReference] == isChangelog)
+ assert(ref.isInstanceOf[V2TableReference] != isChangelog)
+ assert(ref.catalog eq catalog)
+ assert(ref.identifier == identifier)
+ assert(ref.options eq options)
+ assert(ref.context == context)
+ assert(ref.info.tableId == Option(original.table.id()))
+ assert(ref.info.columns == original.table.columns().toSeq)
+ assert(ref.output == original.output)
+ assert(ref.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(42L))
+ assert(ref.name == original.name)
+ assert(ref.computeStats() == Statistics.DUMMY)
+ assert(ref.simpleString(10).startsWith(
+ if (isChangelog) "ChangelogReference[" else "TableReference["))
+ ref match {
+ case changelogRef: V2ChangelogReference =>
+ assert(changelogRef.changelog eq original.table)
+ assert(changelogRef.changelog.changelogContext eq changelogContext)
+ assert(changelogRef.changelog.resolved)
+ case _: V2TableReference =>
+ }
+
+ val fresh = ref.newInstance()
+ assert(fresh.getClass == ref.getClass)
+ assert(fresh.context == context)
+ assert(fresh.info == ref.info)
+ assert(fresh.options eq options)
+ assert(fresh.output.map(_.name) == ref.output.map(_.name))
+ assert(fresh.output.zip(ref.output).forall { case (a, b) => a.exprId !=
b.exprId })
+ comparePlans(ref.toRelation(original.table), original)
+ }
+
+ gridTest("both reference types validate captured schemas for temporary
views")(
+ Seq(false, true)) { isChangelog =>
+ val original = relation(table(isChangelog))
+ val ref = V2Reference.createForTempView(original, Seq("view"))
+ V2ReferenceUtils.validateLoadedTable(
+ table(isChangelog, columns :+ Column.create("added", LongType)), ref)
+ checkError(
+ intercept[AnalysisException] {
+ V2ReferenceUtils.validateLoadedTable(table(isChangelog, Array.empty),
ref)
+ },
+ condition = "INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
+ parameters = Map(
+ "viewName" -> "`view`",
+ "tableName" -> "`catalog`.`ns`.`table`",
+ "colType" -> "data",
+ "errors" -> "- `id` BIGINT has been removed"))
+ }
+
+ gridTest("both reference types reject changed schemas in transactions")(
+ Seq(false, true)) { isChangelog =>
+ val ref = V2Reference.createForTransaction(relation(table(isChangelog)))
+ checkError(
+ intercept[AnalysisException] {
+ V2ReferenceUtils.validateLoadedTable(table(isChangelog, Array.empty),
ref)
+ },
+ condition = "INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ parameters = Map(
+ "tableName" -> "`catalog`.`ns`.`table`",
+ "errors" -> "- `id` BIGINT has been removed"))
+ }
+}
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 7c2de5e69890..acbc2aefe212 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
@@ -327,6 +327,91 @@ class CatalogV2UtilSuite extends SparkFunSuite {
assert(stateOptions.isEmpty)
}
+ gridTest("state option projection preserves effective values for
duplicate-case keys")(
+ Seq("table", "changelog")) { kind =>
+ val catalog = catalogWithStateOptions(java.util.Set.of("SnApShOt"))
+ when(catalog.changelogStateOptionKeys()).thenCallRealMethod()
+ val entries = Seq("snapshot" -> "lower-value", "SNAPSHOT" -> "upper-value")
+ Seq(entries, entries.reverse).foreach { orderedEntries =>
+ val original = new java.util.LinkedHashMap[String, String]()
+ orderedEntries.foreach { case (key, value) => original.put(key, value) }
+ original.put("split-size", "5")
+ val options = new CaseInsensitiveStringMap(original)
+
+ val projected = if (kind == "table") {
+ CatalogV2Util.extractTableStateOptions(catalog, options)
+ } else {
+ CatalogV2Util.extractChangelogStateOptions(catalog, options)
+ }
+
+ assert(options.get("snapshot") == orderedEntries.last._2)
+ assert(projected.size() == 1)
+ assert(projected.get("snapshot") == options.get("snapshot"))
+ assert(projected.asCaseSensitiveMap().containsKey("snapshot"))
+ assert(projected.asCaseSensitiveMap().containsKey("SNAPSHOT"))
+ }
+ }
+
+ test("extractChangelogStateOptions defaults to table state keys") {
+ val catalog = catalogWithStateOptions(java.util.Set.of("BrAnCh"))
+ when(catalog.changelogStateOptionKeys()).thenCallRealMethod()
+ val options = new CaseInsensitiveStringMap(java.util.Map.of(
+ "BRANCH", "Main", "split-size", "5", "startingVersion", "1"))
+
+ val stateOptions = CatalogV2Util.extractChangelogStateOptions(catalog,
options)
+
+ assert(stateOptions == new
CaseInsensitiveStringMap(java.util.Map.of("branch", "Main")))
+ assert(stateOptions.asCaseSensitiveMap().containsKey("BRANCH"))
+ assert(options.size() == 3)
+ }
+
+ test("extractChangelogStateOptions allows independent changelog state keys")
{
+ val catalog = catalogWithStateOptions(java.util.Set.of("branch"))
+
when(catalog.changelogStateOptionKeys()).thenReturn(java.util.Set.of("schemaVersion"))
+ val options = new CaseInsensitiveStringMap(java.util.Map.of(
+ "branch", "Main", "SCHEMAVERSION", "10", "split-size", "5"))
+
+ val changelogOptions = CatalogV2Util.extractChangelogStateOptions(catalog,
options)
+ val tableOptions = CatalogV2Util.extractTableStateOptions(catalog, options)
+
+ assert(changelogOptions ==
+ new CaseInsensitiveStringMap(java.util.Map.of("schemaVersion", "10")))
+ assert(tableOptions == new
CaseInsensitiveStringMap(java.util.Map.of("branch", "Main")))
+ }
+
+ test("extractChangelogStateOptions allows an empty override of table state
keys") {
+ val catalog = catalogWithStateOptions(java.util.Set.of("branch"))
+
when(catalog.changelogStateOptionKeys()).thenReturn(java.util.Set.of[String]())
+ val options = new CaseInsensitiveStringMap(java.util.Map.of("branch",
"Main"))
+
+ assert(CatalogV2Util.extractChangelogStateOptions(catalog,
options).isEmpty)
+ }
+
+ test("extractChangelogStateOptions returns no options by default") {
+ val catalog = mock(classOf[TableCatalog])
+ when(catalog.tableStateOptionKeys()).thenCallRealMethod()
+ when(catalog.changelogStateOptionKeys()).thenCallRealMethod()
+ val options = new CaseInsensitiveStringMap(
+ java.util.Map.of("branch", "Main", "split-size", "5"))
+
+ assert(CatalogV2Util.extractChangelogStateOptions(catalog,
options).isEmpty)
+ }
+
+ test("extractChangelogStateOptions compares keys case-insensitively and
values sensitively") {
+ val catalog = mock(classOf[TableCatalog])
+
when(catalog.changelogStateOptionKeys()).thenReturn(java.util.Set.of("BrAnCh"))
+ val lowerCaseKey = CatalogV2Util.extractChangelogStateOptions(
+ catalog, new CaseInsensitiveStringMap(java.util.Map.of("branch",
"Main")))
+ val upperCaseKey = CatalogV2Util.extractChangelogStateOptions(
+ catalog, new CaseInsensitiveStringMap(java.util.Map.of("BRANCH",
"Main")))
+ val lowerCaseValue = CatalogV2Util.extractChangelogStateOptions(
+ catalog, new CaseInsensitiveStringMap(java.util.Map.of("branch",
"main")))
+
+ assert(lowerCaseKey == upperCaseKey)
+ assert(lowerCaseKey.hashCode() == upperCaseKey.hashCode())
+ assert(lowerCaseKey != lowerCaseValue)
+ }
+
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/InMemoryChangelogCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala
index 0c1def1ac55c..1fab462c6362 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala
@@ -20,8 +20,6 @@ package org.apache.spark.sql.connector.catalog
import scala.collection.mutable
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
-import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.connector.catalog.ChangelogRange.{TimestampRange,
UnboundedRange, VersionRange}
import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
import org.apache.spark.sql.connector.read._
@@ -40,13 +38,16 @@ class InMemoryChangelogCatalog extends InMemoryCatalog {
private val changeData: mutable.Map[String,
mutable.ArrayBuffer[InternalRow]] =
mutable.Map.empty
- // Stores the most recent ChangelogContext and options passed to
loadChangelog(), so tests
- // can verify that the parser/DataFrame API correctly constructed and
forwarded them.
- private var _lastChangelogContext: Option[ChangelogContext] = None
- def lastChangelogContext: Option[ChangelogContext] = _lastChangelogContext
+ private val _loadChangelogCalls =
+ mutable.ArrayBuffer.empty[(ChangelogContext, CaseInsensitiveStringMap)]
+ def loadChangelogCalls: Seq[(ChangelogContext, CaseInsensitiveStringMap)] =
+ _loadChangelogCalls.toSeq
+ def resetLoadChangelogCalls(): Unit = _loadChangelogCalls.clear()
+ def lastChangelogContext: Option[ChangelogContext] =
_loadChangelogCalls.lastOption.map(_._1)
+ def lastOptions: Option[CaseInsensitiveStringMap] =
_loadChangelogCalls.lastOption.map(_._2)
- private var _lastOptions: Option[CaseInsensitiveStringMap] = None
- def lastOptions: Option[CaseInsensitiveStringMap] = _lastOptions
+ private var _lastScanOptions: Option[CaseInsensitiveStringMap] = None
+ def lastScanOptions: Option[CaseInsensitiveStringMap] = _lastScanOptions
// Per-table overrides for Changelog properties (carry-over rows,
intermediate changes,
// update representation, row identity). Tests can set these to exercise
post-processing.
@@ -68,12 +69,9 @@ class InMemoryChangelogCatalog extends InMemoryCatalog {
ident: Identifier,
changelogContext: ChangelogContext,
options: CaseInsensitiveStringMap): Changelog = {
- _lastChangelogContext = Some(changelogContext)
- _lastOptions = Some(options)
- if (!tableExists(ident)) {
- throw new NoSuchTableException(ident.asMultipartIdentifier)
- }
- val table = loadTable(ident)
+ _loadChangelogCalls += ((changelogContext, options))
+ // Access the connector's own metadata without invoking the ordinary
table-loading API.
+ val table = liveTable(ident)
val allRows = changeData.getOrElse(
ident.toString, mutable.ArrayBuffer.empty)
val numDataCols = table.columns.length
@@ -82,7 +80,8 @@ class InMemoryChangelogCatalog extends InMemoryCatalog {
val filtered = filterByRange(allRows.toSeq, commitVersionIdx,
changelogContext.range())
val props = changelogProperties.getOrElse(ident.toString,
ChangelogProperties())
new InMemoryChangelog(
- table.name + "_changelog", table.columns, filtered, props)
+ table.name + "_changelog", table.columns, filtered, props,
+ scanOptions => _lastScanOptions = Some(scanOptions))
}
/**
@@ -172,17 +171,29 @@ class InMemoryChangelog(
tableName: String,
dataColumns: Array[Column],
changeRows: Seq[InternalRow],
- properties: ChangelogProperties = ChangelogProperties()) extends Changelog
{
+ properties: ChangelogProperties = ChangelogProperties(),
+ onScan: CaseInsensitiveStringMap => Unit = _ => ()) extends Changelog {
private val cdcColumns: Array[Column] = dataColumns ++ Array(
Column.create("_change_type", StringType),
Column.create("_commit_version", LongType),
Column.create("_commit_timestamp", TimestampType,
properties.commitTimestampNullable))
+ private val rows = changeRows.iterator.map(_.copy()).toVector
+ // Independent loads can share cached results only when their captured read
states match.
+ private val readIdentity = (tableName, dataColumns.toVector, rows,
properties)
+
override def name(): String = tableName
override def columns(): Array[Column] = cdcColumns
+ override def equals(other: Any): Boolean = other match {
+ case that: InMemoryChangelog => getClass == that.getClass && readIdentity
== that.readIdentity
+ case _ => false
+ }
+
+ override def hashCode(): Int = readIdentity.hashCode()
+
override def containsCarryoverRows(): Boolean =
properties.containsCarryoverRows
override def containsIntermediateChanges(): Boolean =
properties.containsIntermediateChanges
@@ -205,7 +216,8 @@ class InMemoryChangelog(
override def newScanBuilder(
options: CaseInsensitiveStringMap): ScanBuilder = {
- new InMemoryChangelogScanBuilder(readSchema, changeRows)
+ onScan(options)
+ new InMemoryChangelogScanBuilder(readSchema, rows)
}
def readSchema: StructType = {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTableSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTableSuite.scala
new file mode 100644
index 000000000000..96147f96e02f
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ChangelogTableSuite.scala
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import java.util.Optional
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{
+ ChangelogContext, ChangelogProperties, ChangelogRange, Column, Identifier,
InMemoryChangelog,
+ InMemoryChangelogCatalog}
+import
org.apache.spark.sql.connector.catalog.ChangelogContext.DeduplicationMode
+import org.apache.spark.sql.connector.catalog.ChangelogRange.{TimestampRange,
UnboundedRange, VersionRange}
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.types.LongType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
+
+class ChangelogTableSuite extends SparkFunSuite {
+
+ private val dataColumns = Seq("id", "other_id", "row_version",
"other_version")
+ .map(name => Column.create(name, LongType, false)).toArray
+ private val properties = ChangelogProperties(
+ containsCarryoverRows = true,
+ containsIntermediateChanges = true,
+ representsUpdateAsDeleteAndInsert = true,
+ rowIdNames = Seq("id"),
+ rowVersionName = Some("row_version"))
+
+ private def table(
+ props: ChangelogProperties = properties,
+ mode: DeduplicationMode = DeduplicationMode.NET_CHANGES,
+ computeUpdates: Boolean = true,
+ resolved: Boolean = true,
+ range: ChangelogRange = new UnboundedRange()): ChangelogTable = {
+ ChangelogTable(
+ new InMemoryChangelog("changes", dataColumns, Seq.empty, props),
+ new ChangelogContext(range, mode, computeUpdates),
+ resolved)
+ }
+
+ private def checkMetadataChange(
+ captured: ChangelogTable,
+ current: ChangelogTable,
+ changedProperties: String): Unit = {
+ checkError(
+ intercept[AnalysisException] { captured.validateRefresh(current) },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.CHANGELOG_METADATA_MISMATCH",
+ parameters = Map(
+ "tableName" -> "`changes`",
+ "changedProperties" -> changedProperties))
+ }
+
+ gridTest("changelog range is bounded only when it has an ending bound")(Seq(
+ new VersionRange("1", Optional.of("2"), true, true) -> true,
+ new VersionRange("1", Optional.empty[String](), true, true) -> false,
+ new TimestampRange(1L, Optional.of[java.lang.Long](2L), true, true) ->
true,
+ new TimestampRange(1L, Optional.empty[java.lang.Long](), true, true) ->
false,
+ new UnboundedRange() -> false)) { case (range, expected) =>
+ assert(table(range = range).isBounded == expected)
+ }
+
+ Seq(
+ properties.copy(containsCarryoverRows = false) -> "containsCarryoverRows,
rowVersion",
+ properties.copy(containsIntermediateChanges = false) ->
"containsIntermediateChanges",
+ properties.copy(representsUpdateAsDeleteAndInsert = false) ->
+ "representsUpdateAsDeleteAndInsert",
+ properties.copy(rowIdNames = Seq("other_id")) -> "rowId",
+ properties.copy(rowVersionName = Some("other_version")) -> "rowVersion"
+ ).foreach { case (changed, changedProperties) =>
+ test(s"resolved changelog refresh rejects changed $changedProperties") {
+ checkMetadataChange(table(), table(changed), changedProperties)
+ }
+ }
+
+ test("changelog refresh compares metadata and references by value") {
+ table().validateRefresh(table())
+ }
+
+ test("raw changelog refresh ignores metadata unused by post-processing") {
+ val changed = properties.copy(
+ containsCarryoverRows = false,
+ containsIntermediateChanges = false,
+ representsUpdateAsDeleteAndInsert = false,
+ rowIdNames = Seq("other_id"),
+ rowVersionName = Some("other_version"))
+ table(mode = DeduplicationMode.NONE, computeUpdates =
false).validateRefresh(
+ table(changed, mode = DeduplicationMode.NONE, computeUpdates = false))
+ }
+
+ test("carryover-only changelog refresh ignores unused intermediate and
update metadata") {
+ val changed = properties.copy(
+ containsIntermediateChanges = false,
+ representsUpdateAsDeleteAndInsert = false)
+ table(mode = DeduplicationMode.DROP_CARRYOVERS, computeUpdates =
false).validateRefresh(
+ table(
+ changed,
+ mode = DeduplicationMode.DROP_CARRYOVERS,
+ computeUpdates = false))
+ }
+
+ gridTest("changelog construction rejects invalid post-processing options")(
+ Seq(false, true)) { resolved =>
+ checkError(
+ intercept[AnalysisException] {
+ table(mode = DeduplicationMode.NONE, computeUpdates = true, resolved =
resolved)
+ },
+ condition =
"INVALID_CDC_OPTION.UPDATE_DETECTION_REQUIRES_CARRY_OVER_REMOVAL",
+ parameters = Map("changelogName" -> "changes"))
+ }
+
+ test("update-only changelog refresh ignores unused row-version references") {
+ val updateOnly = properties.copy(
+ containsCarryoverRows = false,
+ containsIntermediateChanges = false)
+ table(updateOnly).validateRefresh(
+ table(updateOnly.copy(rowVersionName = Some("other_version"))))
+ }
+
+ test("changelog refresh preserves a snapshot of mutable row reference
paths") {
+ val mutablePath = Array("id")
+ val changelog = new InMemoryChangelog("changes", dataColumns, Seq.empty,
properties) {
+ override def rowId(): Array[NamedReference] = Array(new NamedReference {
+ override def fieldNames(): Array[String] = mutablePath
+ })
+ }
+ val context = new ChangelogContext(
+ new UnboundedRange(), DeduplicationMode.NET_CHANGES, true)
+ val captured = ChangelogTable(changelog, context, resolved = true)
+ mutablePath(0) = "other_id"
+ checkMetadataChange(captured, ChangelogTable(changelog, context), "rowId")
+ }
+
+ test("unresolved changelog refresh permits metadata changes before
post-processing") {
+ table(resolved = false).validateRefresh(
+ table(properties.copy(rowIdNames = Seq("other_id"))))
+ }
+
+ test("changelog relations compare by connector equality, context,
identifier, and options") {
+ val catalog = new InMemoryChangelogCatalog
+ catalog.initialize("catalog", CaseInsensitiveStringMap.empty())
+ val identifier = Identifier.of(Array("ns"), "table")
+ val options = new CaseInsensitiveStringMap(Map("state" -> "base").asJava)
+
+ def relation(
+ changelogTable: ChangelogTable = table(),
+ ident: Identifier = identifier,
+ relationOptions: CaseInsensitiveStringMap = options):
DataSourceV2Relation = {
+ DataSourceV2Relation.create(
+ changelogTable, Some(catalog), Some(ident), relationOptions)
+ }
+
+ val base = relation()
+ val equivalent = relation()
+ assert(!(base.table.asInstanceOf[ChangelogTable].changelog eq
+ equivalent.table.asInstanceOf[ChangelogTable].changelog))
+ assert(base.sameResult(equivalent))
+ assert(base.semanticHash() == equivalent.semanticHash())
+ assert(!base.sameResult(
+ relation(table(mode = DeduplicationMode.DROP_CARRYOVERS))))
+ assert(!base.sameResult(relation(table(computeUpdates = false))))
+ assert(!base.sameResult(relation(table(
+ range = new VersionRange("1", Optional.of("2"), true, true)))))
+ assert(!base.sameResult(
+ relation(ident = Identifier.of(Array("ns"), "other_table"))))
+ assert(!base.sameResult(relation(
+ relationOptions = new CaseInsensitiveStringMap(Map("state" ->
"other").asJava))))
+
+ val row = InternalRow(1L, 2L, 3L, 4L, UTF8String.fromString("insert"), 5L,
6L)
+ val changedData = new InMemoryChangelog("changes", dataColumns, Seq(row),
properties)
+ assert(!base.sameResult(relation(table().copy(changelog = changedData))))
+ }
+
+ test("changelog relations preserve connector identity equality") {
+ val catalog = new InMemoryChangelogCatalog
+ catalog.initialize("catalog", CaseInsensitiveStringMap.empty())
+ val identifier = Identifier.of(Array("ns"), "table")
+
+ def identityTable(): ChangelogTable = {
+ val changelog = new InMemoryChangelog("changes", dataColumns, Seq.empty,
properties) {
+ override def equals(other: Any): Boolean = other match {
+ case ref: AnyRef => this eq ref
+ case _ => false
+ }
+
+ override def hashCode(): Int = System.identityHashCode(this)
+ }
+ table().copy(changelog = changelog)
+ }
+
+ def relation(changelogTable: ChangelogTable): DataSourceV2Relation = {
+ DataSourceV2Relation.create(changelogTable, Some(catalog),
Some(identifier))
+ }
+
+ val captured = identityTable()
+ val valueCompared = table().changelog
+ assert(valueCompared != captured.changelog)
+ assert(captured.changelog != valueCompared)
+ val base = relation(captured)
+ val sameConnector = relation(captured.copy())
+ assert(base.sameResult(sameConnector))
+ assert(base.semanticHash() == sameConnector.semanticHash())
+ assert(!base.sameResult(relation(identityTable())))
+ }
+}
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 139216884435..9f4ee12b8cec 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
@@ -44,7 +44,7 @@ import org.apache.spark.sql.execution.datasources.{
HadoopFsRelation,
LogicalRelation,
LogicalRelationWithTable}
-import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2Relation, ExtractV2CatalogAndIdentifier, ExtractV2Table, FileTable,
V2TableRefreshUtil}
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
ChangelogTable, DataSourceV2Relation, ExtractV2CatalogAndIdentifier,
ExtractV2Table, FileTable, V2TableRefreshUtil}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -430,7 +430,8 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
private def tryRefreshPlan(spark: SparkSession, plan: LogicalPlan):
Option[LogicalPlan] = {
try {
EliminateSubqueryAliases(plan) match {
- case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if
r.timeTravelSpec.isEmpty =>
+ case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
+ if r.timeTravelSpec.isEmpty &&
!r.table.isInstanceOf[ChangelogTable] =>
val table = CatalogV2Util.getTable(catalog, ident, options =
r.options)
if (r.table.id == table.id) {
Some(DataSourceV2Relation.create(table, Some(catalog),
Some(ident), r.options))
@@ -455,9 +456,12 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
resolver: Resolver): Option[LogicalPlan] = {
val name = ident.toQualifiedNameParts(catalog)
val cachedRelations = findCachedRelations(name, resolver)
+ // Changelog loads have their own context and state options, so they
cannot supply an
+ // ordinary table's metadata even when the catalog and identifier match.
val cachedRelation = cachedRelations.collectFirst {
case r: DataSourceV2Relation
- if r.catalog.contains(catalog) && r.identifier.contains(ident) &&
+ if !r.table.isInstanceOf[ChangelogTable] &&
+ r.catalog.contains(catalog) && r.identifier.contains(ident) &&
tableId.forall(_ == r.table.id) &&
CatalogV2Util.extractTableStateOptions(catalog, r.options) ==
stateOptions =>
r
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index 4338bd367fe0..bee5e97afc4f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -27,7 +27,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.catalyst.{CapturesConfig, SQLConfHelper,
TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.{AnalysisContext,
GlobalTempView, LocalTempView, ResolvedIdentifier, ResolvedPersistentView,
SchemaEvolution, SchemaUnsupported, ViewSchemaMode, ViewType}
-import org.apache.spark.sql.catalyst.analysis.V2TableReference
+import org.apache.spark.sql.catalyst.analysis.V2Reference
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType, TemporaryViewRelation}
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
SubqueryExpression, VariableReference}
import org.apache.spark.sql.catalyst.plans.logical.{AlterViewAs,
AnalysisOnlyCommand, CreateTempView, CreateView, CTEInChildren, CTERelationDef,
LogicalPlan, Project, View, WithCTE}
@@ -36,7 +36,7 @@ import
org.apache.spark.sql.classic.ClassicConversions.castToImpl
import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin,
Identifier, ViewCatalog}
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.{IdentifierHelper,
NamespaceHelper}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
import org.apache.spark.sql.internal.StaticSQLConf
import org.apache.spark.sql.types.{MetadataBuilder, StructType}
import org.apache.spark.sql.util.SchemaUtils
@@ -758,7 +758,10 @@ object ViewHelper extends SQLConfHelper with Logging with
CapturesConfig {
plan transform {
case r: DataSourceV2Relation
if r.catalog.isDefined && r.identifier.isDefined &&
r.timeTravelSpec.isEmpty =>
- V2TableReference.createForTempView(r, viewName.nameParts)
+ r.table match {
+ case changelog: ChangelogTable if changelog.isBounded => r
+ case _ => V2Reference.createForTempView(r, viewName.nameParts)
+ }
}
}
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 da5f930795e7..ff0ca5bf5a4e 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
@@ -21,6 +21,7 @@ import scala.collection.mutable
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.ChangelogCacheKey
import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan}
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{Identifier, Table,
TableCatalog, V2TableUtil}
@@ -43,8 +44,8 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper
with Logging {
* - Data columns: Verifies captured columns align with the current schema
* - Metadata columns: Checks metadata column consistency
*
- * Tables with time travel specifications are skipped as they reference a
specific point
- * in time and don't have to be refreshed.
+ * Tables with time travel specifications and changelogs with an explicit
ending bound are
+ * skipped because they reference captured historical data.
*
* Schema validation mode depends on the underlying plan. Commands, for
instance,
* prohibit any schema changes while queries permit adding columns.
@@ -70,8 +71,8 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper
with Logging {
* - Data columns: Verifies captured columns align with the current schema
* - Metadata columns: Checks metadata column consistency
*
- * Tables with time travel specifications are skipped as they reference a
specific point
- * in time and don't have to be refreshed.
+ * Tables with time travel specifications and changelogs with an explicit
ending bound are
+ * skipped because they reference captured historical data.
*
* @param spark the currently active Spark session
* @param plan the logical plan to refresh
@@ -85,21 +86,34 @@ private[sql] object V2TableRefreshUtil extends
SQLConfHelper with Logging {
versionedOnly: Boolean,
schemaValidationMode: SchemaValidationMode): LogicalPlan = {
val currentTables = mutable.HashMap.empty[CurrentTableKey, Table]
+ val currentChangelogs = mutable.HashMap.empty[ChangelogCacheKey,
ChangelogTable]
plan transformWithSubqueries {
+ case r @ DataSourceV2Relation(changelog: ChangelogTable, _, _, _, _, _)
+ if changelog.isBounded => r
case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
if (r.isVersioned || !versionedOnly) && r.timeTravelSpec.isEmpty =>
- 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, stateOptions)
match {
- case Some(cached) =>
- logDebug(s"Refreshing table metadata for $tableName using shared
relation cache")
- cached.table
- case _ =>
- logDebug(s"Refreshing table metadata for $tableName using
catalog")
- CatalogV2Util.getTableWithStateOptions(catalog, ident,
stateOptions)
- }
- })
+ val currentTable = r.table match {
+ case captured: ChangelogTable =>
+ val stateOptions =
CatalogV2Util.extractChangelogStateOptions(catalog, r.options)
+ val key = ChangelogCacheKey(catalog, ident,
captured.changelogContext, stateOptions)
+ val current = currentChangelogs.getOrElseUpdate(key,
+ ChangelogTable.load(catalog, ident, captured.changelogContext,
stateOptions))
+ captured.validateRefresh(current)
+ current.copy(resolved = captured.resolved)
+ case _ =>
+ val stateOptions = CatalogV2Util.extractTableStateOptions(catalog,
r.options)
+ currentTables.getOrElseUpdate((catalog, ident, stateOptions), {
+ val tableName = V2TableUtil.toQualifiedName(catalog, ident)
+ lookupCachedRelation(spark, catalog, ident, r.table,
stateOptions) match {
+ case Some(cached) =>
+ logDebug(s"Refreshing table metadata for $tableName using
shared relation cache")
+ cached.table
+ case _ =>
+ logDebug(s"Refreshing table metadata for $tableName using
catalog")
+ CatalogV2Util.getTableWithStateOptions(catalog, ident,
stateOptions)
+ }
+ })
+ }
validateTableIdentity(currentTable, r)
validateDataColumns(currentTable, r, schemaValidationMode)
validateMetadataColumns(currentTable, r, schemaValidationMode)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
index 8c70efcb8615..4bb6124d37da 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
@@ -29,7 +29,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.{SparkException, SparkIllegalArgumentException,
SparkIllegalStateException}
import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.sql.catalyst.analysis.{ResolveDeduplicate,
V2TableReference}
+import org.apache.spark.sql.catalyst.analysis.{ResolveDeduplicate, V2Reference}
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
CurrentBatchTimestamp, CurrentDate, CurrentTimestamp, CurrentTimestampNanos,
FileSourceMetadataAttribute, LocalTimestamp, LocalTimestampNanos}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Deduplicate,
DeduplicateWithinWatermark, Distinct, FlatMapGroupsInPandasWithState,
FlatMapGroupsWithState, GlobalLimit, Join, LeafNode, LocalRelation,
LogicalPlan, Project, StreamSourceAwareLogicalPlan, TransformWithState,
TransformWithStateInPySpark}
@@ -430,7 +430,7 @@ class MicroBatchExecution(
val catalogManager =
sparkSessionForStream.sessionState.catalogManager
val streamingCatalog = catalogManager.catalog(catalog.name)
val v2Relation = DataSourceV2Relation.create(s,
Some(streamingCatalog), Some(ident))
- V2TableReference.createForWriteTarget(v2Relation)
+ V2Reference.createForWriteTarget(v2Relation)
case Some((catalog, ident)) =>
DataSourceV2Relation.create(s, Some(catalog), Some(ident))
case None => DataSourceV2Relation.create(s, None, None)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
index 082be2ac22c8..b615857b80f9 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
@@ -19,18 +19,22 @@ package org.apache.spark.sql.connector
import java.util.Collections
-import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.ResolveChangelogTable
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Window}
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.classic.Dataset
import org.apache.spark.sql.connector.catalog._
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.connector.catalog.ChangelogRange
import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference, Transform}
import org.apache.spark.sql.connector.read.ScanBuilder
-import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation, V2TableRefreshUtil}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{ArrayType, IntegerType, LongType, MapType,
StringType, StructField, StructType, TimestampType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
/**
* Tests for the CDC (Change Data Capture) analyzer resolution path:
@@ -40,25 +44,36 @@ class ChangelogResolutionSuite extends SharedSparkSession {
private val cdcCatalogName = "cdc_catalog"
private val noCdcCatalogName = "no_cdc_catalog"
+ private val ident = Identifier.of(Array.empty, "test_table")
+
+ private def cdcCatalog: ChangelogStateOptionsCatalog = {
+ spark.sessionState.catalogManager.catalog(cdcCatalogName)
+ .asInstanceOf[ChangelogStateOptionsCatalog]
+ }
override def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set(s"spark.sql.catalog.$cdcCatalogName",
- classOf[InMemoryChangelogCatalog].getName)
+ classOf[ChangelogStateOptionsCatalog].getName)
+ spark.conf.set(s"spark.sql.catalog.$cdcCatalogName.tableStateOptionKeys",
"branch")
spark.conf.set(s"spark.sql.catalog.$noCdcCatalogName",
classOf[InMemoryTableCatalog].getName)
}
override def afterAll(): Unit = {
spark.conf.unset(s"spark.sql.catalog.$cdcCatalogName")
+ spark.conf.unset(s"spark.sql.catalog.$cdcCatalogName.tableStateOptionKeys")
spark.conf.unset(s"spark.sql.catalog.$noCdcCatalogName")
super.afterAll()
}
override def beforeEach(): Unit = {
super.beforeEach()
+ cdcCatalog.changelogKeys = None
+ cdcCatalog.rejectCurrentTableLoads = false
+ cdcCatalog.clearChangeRows(ident)
+ cdcCatalog.setChangelogProperties(ident, ChangelogProperties())
val catalog =
spark.sessionState.catalogManager.catalog(cdcCatalogName).asTableCatalog
- val ident = Identifier.of(Array.empty, "test_table")
if (catalog.tableExists(ident)) {
catalog.dropTable(ident)
}
@@ -208,63 +223,336 @@ class ChangelogResolutionSuite extends
SharedSparkSession {
assert(range.endingVersion().get() == "5")
}
- test("user-defined options are forwarded to loadChangelog") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
-
- spark.read
- .option("startingVersion", "1")
- .option("customOption", "customValue")
- .changes(s"$cdcCatalogName.test_table")
-
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
- assert(opts.get.get("startingVersion") == "1")
+ gridTest("CDC resolution filters state options and retains scan options")(
+ for (streaming <- Seq(false, true); useSql <- Seq(false, true)) yield
(streaming, useSql)) {
+ case (streaming, useSql) =>
+ val cat = cdcCatalog
+ cat.resetLoadChangelogCalls()
+ val tableName = s"$cdcCatalogName.test_table"
+ val df = if (useSql) {
+ val prefix = if (streaming) "STREAM " else ""
+ sql(s"SELECT * FROM $prefix$tableName CHANGES FROM VERSION 1 " +
+ "WITH ('BrAnCh' = 'Main', 'customOption' = 'customValue')")
+ } else if (streaming) {
+ spark.readStream.option("startingVersion", "1").option("BrAnCh",
"Main")
+ .option("customOption", "customValue").changes(tableName)
+ } else {
+ spark.read.option("startingVersion", "1").option("BrAnCh", "Main")
+ .option("customOption", "customValue").changes(tableName)
+ }
+ val options = df.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.options
+ case r: StreamingRelationV2 => r.extraOptions
+ }.get
+ assert(cat.loadChangelogCalls.size == 1)
+ assert(cat.lastOptions.get.size() == 1)
+ assert(cat.lastOptions.get.get("branch") == "Main")
+ assert(options.get("branch") == "Main")
+ assert(options.get("customOption") == "customValue")
+ if (!useSql) assert(options.get("startingVersion") == "1")
+ if (!streaming) {
+ df.queryExecution.optimizedPlan
+ assert(cat.lastScanOptions.contains(options))
+ }
+ }
+
+ gridTest("CDC state option keys can inherit, override, or disable table
state keys")(
+ Seq("default", "override", "empty")) { keyMode =>
+ val cat = cdcCatalog
+ cat.changelogKeys = keyMode match {
+ case "override" => Some(java.util.Set.of("ReF"))
+ case "empty" => Some(java.util.Set.of[String]())
+ case _ => None
+ }
+ spark.read.option("startingVersion", "1").option("BRANCH", "Main")
+ .option("ref", "Dev").option("scanOption", "10")
+ .changes(s"$cdcCatalogName.test_table").queryExecution.analyzed
+
+ val expected = keyMode match {
+ case "default" => new
CaseInsensitiveStringMap(Collections.singletonMap("branch", "Main"))
+ case "override" => new
CaseInsensitiveStringMap(Collections.singletonMap("ref", "Dev"))
+ case _ => CaseInsensitiveStringMap.empty()
+ }
+ assert(cat.lastOptions.contains(expected))
}
- test("user-defined options are forwarded to loadChangelog - SQL WITH
clause") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
+ private def changeRow(id: Long, version: Long): InternalRow = InternalRow(
+ id, UTF8String.fromString(s"data-$id"),
UTF8String.fromString(Changelog.CHANGE_TYPE_INSERT),
+ version, version * 1000000L)
- sql(s"SELECT * FROM $cdcCatalogName.test_table CHANGES FROM VERSION 1 " +
- "WITH ('customOption' = 'customValue')").queryExecution.analyzed
-
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
+ test("CDC metadata loads are shared across scan options but isolate contexts
and state values") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L), changeRow(2L, 2L)))
+ cat.resetLoadChangelogCalls()
+ val tableName = s"$cdcCatalogName.test_table"
+ val df = sql(
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('BrAnCh' = 'Main', 'split-size' = '5') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('branch' = 'Main', 'split-size' = '9') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 2 " +
+ "WITH ('branch' = 'Main') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('branch' = 'main') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('branch' = 'Main', 'deduplicationMode' = 'none')")
+ val relations = df.queryExecution.analyzed.collect { case r:
DataSourceV2Relation => r }
+ val changelogs = relations.map(_.table.asInstanceOf[ChangelogTable])
+ assert(changelogs.head.changelog eq changelogs(1).changelog)
+ assert(changelogs.head.changelog ne changelogs(2).changelog)
+ assert(changelogs.head.changelog ne changelogs(3).changelog)
+ assert(changelogs.head.changelog ne changelogs(4).changelog)
+ assert(relations.take(2).map(_.options.get("split-size")) == Seq("5", "9"))
+ assert(cat.loadChangelogCalls.size == 4)
+ QueryTest.checkAnswer(
+ df, Seq.fill(4)(Row(1L)) ++ Seq.fill(5)(Row(2L)), checkToRDD = false)
+
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val refreshed = V2TableRefreshUtil.refresh(spark,
df.queryExecution.analyzed)
+ assert(cat.loadChangelogCalls.size == 4)
+ assert(cat.loadTableCalls.isEmpty)
+ val refreshedChanges = refreshed.collect {
+ case r: DataSourceV2Relation => r.table.asInstanceOf[ChangelogTable]
+ }
+ assert(refreshedChanges.head.changelog eq refreshedChanges(1).changelog)
+ assert(refreshedChanges.map(_.changelogContext) ==
changelogs.map(_.changelogContext))
+ assert(refreshedChanges.forall(_.resolved))
+ }
+
+ test("ordinary and CDC metadata caches remain separate") {
+ val tableName = s"$cdcCatalogName.test_table"
+ sql(s"INSERT INTO $tableName VALUES (100, 'current')")
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ cat.resetLoadTableCalls()
+ cat.resetLoadChangelogCalls()
+ val df = sql(s"SELECT id FROM $tableName UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 TO VERSION 10")
+ val relations = df.queryExecution.analyzed.collect { case r:
DataSourceV2Relation => r }
+ assert(relations.count(_.table.isInstanceOf[ChangelogTable]) == 1)
+ assert(cat.loadTableCalls.size == 1)
+ assert(cat.loadChangelogCalls.size == 1)
+ QueryTest.checkAnswer(df, Seq(Row(100L), Row(1L)), checkToRDD = false)
+ }
+
+ test("refreshing cached CDC data preserves CDC reads and ordinary table
cache isolation") {
+ val tableName = s"$cdcCatalogName.test_table"
+ val cdcQuery = s"SELECT * FROM $tableName CHANGES FROM VERSION 1"
+ sql(s"INSERT INTO $tableName VALUES (100, 'current')")
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ val cached = sql(cdcQuery).cache()
+ val cacheManager = spark.sharedState.cacheManager
+ try {
+ checkAnswer(cached.select("id"), Seq(Row(1L)))
+ assert(cacheManager.lookupCachedData(spark.table(tableName)).isEmpty)
+ checkAnswer(spark.table(tableName).select("id"), Seq(Row(100L)))
+
+ cat.addChangeRows(ident, Seq(changeRow(2L, 2L)))
+ spark.catalog.refreshTable(tableName)
+
+ assert(cacheManager.numCachedEntries == 1)
+ assert(cacheManager.lookupCachedData(spark.table(tableName)).isEmpty)
+ checkAnswer(spark.table(tableName).select("id"), Seq(Row(100L)))
+
+ assert(cacheManager.lookupCachedData(cached).isEmpty)
+
+ val fresh = sql(cdcQuery)
+ assert(cacheManager.lookupCachedData(fresh).isDefined)
+ checkAnswer(fresh.select("id"), Seq(Row(1L), Row(2L)))
+
+ val differentRange = sql(
+ s"SELECT * FROM $tableName CHANGES FROM VERSION 2")
+ assert(cacheManager.lookupCachedData(differentRange).isEmpty)
+ } finally {
+ spark.catalog.clearCache()
+ }
}
- test("user-defined options are forwarded to loadChangelog -
DataStreamReader") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
-
- spark.readStream
- .option("startingVersion", "1")
- .option("customOption", "customValue")
- .changes(s"$cdcCatalogName.test_table")
- .queryExecution.analyzed
-
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
- assert(opts.get.get("startingVersion") == "1")
+ test("bounded CDC reads do not load the latest ordinary table") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L), changeRow(10L, 10L),
changeRow(100L, 100L)))
+ cat.rejectCurrentTableLoads = true
+ cat.resetLoadTableCalls()
+ cat.resetLoadChangelogCalls()
+ val df = sql(
+ s"SELECT id FROM $cdcCatalogName.test_table CHANGES FROM VERSION 1 TO
VERSION 10")
+ QueryTest.checkAnswer(df, Seq(Row(1L), Row(10L)), checkToRDD = false)
+ assert(cat.loadTableCalls.isEmpty)
+ assert(cat.loadChangelogCalls.size == 1)
+ val refreshed = V2TableRefreshUtil.refresh(spark,
df.queryExecution.analyzed)
+ assert(refreshed.fastEquals(df.queryExecution.analyzed))
+ assert(cat.loadTableCalls.isEmpty)
+ assert(cat.loadChangelogCalls.size == 1)
+ }
+
+ gridTest("CDC refresh reloads only open-ended ranges")(
+ Seq[(ChangelogRange, Boolean)](
+ new ChangelogRange.VersionRange(
+ "1", java.util.Optional.of("10"), true, true) -> true,
+ new ChangelogRange.TimestampRange(
+ 1000000L, java.util.Optional.of(java.lang.Long.valueOf(10000000L)),
true, true) -> true,
+ new ChangelogRange.VersionRange(
+ "1", java.util.Optional.empty[String](), true, true) -> false,
+ new ChangelogRange.TimestampRange(
+ 1000000L, java.util.Optional.empty[java.lang.Long](), true, true) ->
false,
+ new ChangelogRange.UnboundedRange() -> false)) { case (range, bounded)
=>
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ val context = new ChangelogContext(
+ range, ChangelogContext.DeduplicationMode.DROP_CARRYOVERS, false)
+ val stateOptions = new
CaseInsensitiveStringMap(Collections.singletonMap("branch", "Main"))
+ val scanOptions = new java.util.HashMap[String,
String](stateOptions.asCaseSensitiveMap())
+ scanOptions.put("split-size", "5")
+ val captured = ChangelogTable(
+ cat.loadChangelog(ident, context, stateOptions), context, resolved =
true)
+ val original = DataSourceV2Relation.create(
+ captured, Some(cat), Some(ident), new
CaseInsensitiveStringMap(scanOptions))
+ cat.addChangeRows(ident, Seq(changeRow(2L, 2L)))
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+
+ val refreshed = V2TableRefreshUtil.refresh(spark, original)
+ .asInstanceOf[DataSourceV2Relation]
+ assert(refreshed.options == original.options)
+ assert(cat.loadTableCalls.isEmpty)
+ if (bounded) {
+ assert(refreshed eq original)
+ assert(refreshed.table eq captured)
+ assert(cat.loadChangelogCalls.isEmpty)
+ } else {
+ assert(refreshed.table ne captured)
+ assert(cat.loadChangelogCalls == Seq((context, stateOptions)))
+ val current = refreshed.table.asInstanceOf[ChangelogTable]
+ assert(current.changelogContext == context)
+ assert(current.resolved)
+ }
+ val expected = if (bounded) Seq(Row(1L)) else Seq(Row(1L), Row(2L))
+ QueryTest.checkAnswer(
+ Dataset.ofRows(spark, refreshed).select("id"), expected, checkToRDD =
false)
+ }
+
+ test("versioned-only refresh leaves an unversioned changelog captured") {
+ val cat = cdcCatalog
+ val analyzed = spark.read.option("startingVersion", "1")
+ .changes(s"$cdcCatalogName.test_table").queryExecution.analyzed
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val refreshed = V2TableRefreshUtil.refresh(spark, analyzed, versionedOnly
= true)
+ assert(refreshed.fastEquals(analyzed))
+ assert(cat.loadChangelogCalls.isEmpty)
+ assert(cat.loadTableCalls.isEmpty)
+ }
+
+ test("CDC refresh uses CDC state keys and validates the captured schema") {
+ val cat = cdcCatalog
+ cat.changelogKeys = Some(java.util.Set.of("ReF"))
+ val analyzed = spark.read.option("startingVersion", "1")
+ .option("branch", "Main").option("ref", "History").option("split-size",
"5")
+ .changes(s"$cdcCatalogName.test_table").queryExecution.analyzed
+ val context = cat.lastChangelogContext.get
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val refreshed = V2TableRefreshUtil.refresh(spark, analyzed)
+ assert(cat.loadTableCalls.isEmpty)
+ assert(cat.loadChangelogCalls.size == 1)
+ assert(cat.lastChangelogContext.contains(context))
+ assert(cat.lastOptions.get.size() == 1)
+ assert(cat.lastOptions.get.get("ref") == "History")
+ val relation = refreshed.collectFirst { case r: DataSourceV2Relation => r
}.get
+ assert(relation.options.get("split-size") == "5")
+ assert(relation.table.asInstanceOf[ChangelogTable].resolved)
+
+ cat.alterTable(ident, TableChange.deleteColumn(Array("data"), false))
+ checkError(
+ intercept[AnalysisException] { V2TableRefreshUtil.refresh(spark,
analyzed) },
+ condition = "INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ parameters = Map(
+ "tableName" -> s"`$cdcCatalogName`.`test_table`",
+ "errors" -> "- `data` STRING has been removed"))
+ }
+
+ test("CDC temp views preserve distinct contexts when metadata is reloaded") {
+ withTempView("cdc_first", "cdc_second") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L), changeRow(2L, 2L)))
+ val tableName = s"$cdcCatalogName.test_table"
+ sql(s"SELECT * FROM $tableName CHANGES FROM VERSION 1")
+ .createOrReplaceTempView("cdc_first")
+ sql(s"SELECT * FROM $tableName CHANGES FROM VERSION 2")
+ .createOrReplaceTempView("cdc_second")
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val df = sql("SELECT id FROM cdc_first UNION ALL SELECT id FROM
cdc_second " +
+ "UNION ALL SELECT id FROM cdc_first")
+ val changelogs = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation => r.table.asInstanceOf[ChangelogTable]
+ }
+ assert(changelogs.map(_.changelogContext).distinct.size == 2)
+ assert(changelogs.head.changelog eq changelogs.last.changelog)
+ assert(changelogs.forall(_.resolved))
+ assert(cat.loadChangelogCalls.size == 2)
+ assert(cat.loadTableCalls.isEmpty)
+ QueryTest.checkAnswer(
+ df, Seq(Row(1L), Row(1L), Row(2L), Row(2L), Row(2L)), checkToRDD =
false)
+ }
}
- test("user-defined options are forwarded to loadChangelog - streaming SQL") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
-
- sql(s"SELECT * FROM STREAM $cdcCatalogName.test_table CHANGES FROM VERSION
1 " +
- "WITH ('customOption' = 'customValue')").queryExecution.analyzed
+ gridTest("bounded CDC temp views preserve captured reads")(
+ Seq("VERSION 1 TO VERSION 10",
+ "TIMESTAMP '1970-01-01 00:00:01' TO TIMESTAMP '1970-01-01 00:00:10'"))
{ range =>
+ withTempView("bounded_cdc") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ val changes = sql(s"SELECT * FROM $cdcCatalogName.test_table CHANGES
FROM $range")
+ val captured = changes.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.get
+ changes.createOrReplaceTempView("bounded_cdc")
+ cat.addChangeRows(ident, Seq(changeRow(2L, 2L)))
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+
+ val fromView = spark.table("bounded_cdc")
+ val current = fromView.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.get
+ assert(current eq captured)
+ assert(cat.loadChangelogCalls.isEmpty)
+ assert(cat.loadTableCalls.isEmpty)
+ QueryTest.checkAnswer(fromView.select("id"), Seq(Row(1L)), checkToRDD =
false)
+ }
+ }
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
+ test("CDC refresh preserves post-processing and rejects changed row-version
metadata") {
+ val tableIdent = recreatePostProcessingTable()
+ val cat = cdcCatalog
+ val properties = ChangelogProperties(
+ containsCarryoverRows = true,
+ rowIdNames = Seq("id"),
+ rowVersionName = Some("row_commit_version"))
+ cat.setChangelogProperties(tableIdent, properties)
+ cat.addChangeRows(tableIdent, Seq(
+ InternalRow(1L, 1L, UTF8String.fromString(Changelog.CHANGE_TYPE_DELETE),
1L, 1000000L),
+ InternalRow(1L, 1L, UTF8String.fromString(Changelog.CHANGE_TYPE_INSERT),
1L, 1000000L)))
+ val analyzed = spark.read.option("startingVersion", "1")
+
.changes(s"$cdcCatalogName.test_table").select("id").queryExecution.analyzed
+ assert(analyzed.exists(_.isInstanceOf[Window]))
+ cat.addChangeRows(tableIdent, Seq(
+ InternalRow(2L, 2L, UTF8String.fromString(Changelog.CHANGE_TYPE_INSERT),
2L, 2000000L)))
+ val refreshed = V2TableRefreshUtil.refresh(spark, analyzed)
+ assert(ResolveChangelogTable(refreshed).fastEquals(refreshed))
+ QueryTest.checkAnswer(Dataset.ofRows(spark, refreshed), Seq(Row(2L)),
checkToRDD = false)
+
+ cat.setChangelogProperties(tableIdent,
+ properties.copy(rowVersionName = Some("_commit_version")))
+ checkError(
+ intercept[AnalysisException] { V2TableRefreshUtil.refresh(spark,
analyzed) },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.CHANGELOG_METADATA_MISMATCH",
+ parameters = Map(
+ "tableName" -> s"`$cdcCatalogName`.`test_table_changelog`",
+ "changedProperties" -> "rowVersion"))
}
//
===========================================================================
@@ -550,6 +838,21 @@ class ChangelogResolutionSuite extends SharedSparkSession {
}
+/** Supports configurable CDC keys and detects accidental ordinary-table loads
in CDC paths. */
+class ChangelogStateOptionsCatalog extends InMemoryChangelogCatalog {
+ var changelogKeys: Option[java.util.Set[String]] = None
+ var rejectCurrentTableLoads: Boolean = false
+
+ override def changelogStateOptionKeys(): java.util.Set[String] = {
+ changelogKeys.getOrElse(tableStateOptionKeys())
+ }
+
+ override def loadTable(ident: Identifier): Table = {
+ assert(!rejectCurrentTableLoads, "CDC must not load the current ordinary
table")
+ super.loadTable(ident)
+ }
+}
+
/**
* Test-only [[Changelog]] implementation that returns a hand-crafted schema.
Used to
* exercise [[ChangelogTable]]'s schema validation without going through a
real catalog.
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 8b7fd2bb80e3..42ff027431ca 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
@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.analysis.{
RelationCache,
RelationResolution,
UnresolvedRelation,
- V2TableReference}
+ V2Reference}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
import org.apache.spark.sql.connector.catalog.{
@@ -1634,11 +1634,11 @@ class DataSourceV2OptionSuite extends
DatasourceV2SQLBase {
.analyzed
.collectFirst { case r: DataSourceV2Relation => r }
.getOrElse(fail("expected a v2 relation"))
- val readRef = V2TableReference.createForTransaction(original)
- val otherReadRef = V2TableReference.createForTransaction(original.copy(
+ val readRef = V2Reference.createForTransaction(original)
+ val otherReadRef = V2Reference.createForTransaction(original.copy(
options = new CaseInsensitiveStringMap(
java.util.Map.of("snapshot", "s1", "split-size", "9"))))
- val writeRef = V2TableReference.createForWriteTarget(original)
+ val writeRef = V2Reference.createForWriteTarget(original)
var sharedRelationCacheLookups = 0
val sharedRelationCache: RelationCache = (_, _, _, _, _) => {
sharedRelationCacheLookups += 1
@@ -1710,8 +1710,8 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
.analyzed
.collectFirst { case r: DataSourceV2Relation => r }
.getOrElse(fail("expected a v2 relation"))
- val initialRef = V2TableReference.createForTempView(cached,
Seq("state_view"))
- val otherOptionsRef = V2TableReference.createForTempView(
+ val initialRef = V2Reference.createForTempView(cached, Seq("state_view"))
+ val otherOptionsRef = V2Reference.createForTempView(
cached.copy(options = new CaseInsensitiveStringMap(
java.util.Map.of("snapshot", "s1", "split-size", "9"))),
Seq("state_view"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
