You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2026/09/30 05:03:41 UTC
(spark) branch branch-4.3 updated: [SPARK-59841][4.3][SQL] Align changelog state options and metadata reuse
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 <ge...@apache.org>
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 <ge...@apache.org>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org