This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 469fc2eee035 [SPARK-59841][SQL] Align changelog state options and
metadata reuse
469fc2eee035 is described below
commit 469fc2eee035d22e99ce68e73cefed06496a3c5e
Author: Gengliang Wang <[email protected]>
AuthorDate: Tue Sep 29 20:30:50 2026 -0700
[SPARK-59841][SQL] Align changelog state options and metadata reuse
### What changes were proposed in this pull request?
Add `TableCatalog.changelogStateOptionKeys()`, defaulting to
`tableStateOptionKeys()`, and pass only these declared options to the
catalog-level `loadChangelog` API. Keep all options on each relation for scan
planning. Reuse changelog metadata within an analysis or refresh pass by
`(catalog, identifier, ChangelogContext, changelog state options)`.
Use separate `V2TableReference` and `V2ChangelogReference` nodes under a
common `V2Reference` when re-resolving metadata. `ChangelogTable` captures the
post-processing requirements, validates option combinations at construction,
and checks compatibility when reloading an analyzed CDC plan. Plan comparison
relies on connector-provided `equals` and `hashCode`.
Preserve bounded CDC reads during refresh and DataFrame temporary-view
access. Refresh open-ended CDC reads through `loadChangelog`, retaining their
context and options. Transaction references continue to re-resolve through the
transaction catalog.
### Why are the changes needed?
[SPARK-59841](https://issues.apache.org/jira/browse/SPARK-59841):
`loadTable` uses declared state options, while `loadChangelog` receives all
options and does not share the same metadata-reuse semantics. Reads that differ
only in scan options should reuse CDC metadata; different ranges or state
options must remain separate.
Direct catalog-level CDC loading lets a connector read changes from
versions 1 through 10 when the latest version is 100 without Spark first
loading the current table snapshot.
### Does this PR introduce _any_ user-facing change?
Yes. `loadChangelog` now receives only options declared by
`changelogStateOptionKeys()`. Existing CDC connectors must declare any options
required during metadata loading and obtain Spark-parsed range and processing
parameters from `ChangelogContext`. Scan-only options remain available to
`Changelog.newScanBuilder`.
The new method defaults to the table state keys. An override supplies the
complete CDC state-key set, including any table keys the connector still needs.
The new API is marked `since 4.3.0`.
### How was this patch tested?
Passed 323 tests across the focused Catalyst, CDC, table-option, and
transaction suites below.
Focused reruns of `ChangelogTableSuite` (20 tests) and
`ChangelogResolutionSuite` (49 tests) also
passed. Java Checkstyle and Scala style checks for Catalyst and SQL
main/test sources passed,
along with whitespace and added-source ASCII checks.
The local Maven mirror rejects `at.yawk.lz4:lz4-java:1.12.0`, and public
Maven mirrors are unreachable from this environment. Local tests use the cached
1.11.3 artifact through the temporary SBT setting below; repository dependency
declarations are unchanged:
```text
set ThisBuild / dependencyOverrides += "at.yawk.lz4" % "lz4-java" % "1.11.3"
catalyst/testOnly *V2ReferenceSuite *ChangelogTableSuite
*CatalogV2UtilSuite *TableLookupCacheSuite
sql/testOnly *ChangelogResolutionSuite *ChangelogEndToEndSuite
*ResolveChangelogTable*Suite *DataSourceV2OptionSuite
*AppendDataTransactionSuite *StreamingTransactionSuite
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-6)
Closes #59119 from gengliangwang/codex/spark-59841-changelog-state-options.
Authored-by: Gengliang Wang <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
(cherry picked from commit a32c06b2cfe0275b5e6643810dbff8c91b74e5c6)
Signed-off-by: Gengliang Wang <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 5 +
.../spark/sql/connector/catalog/Changelog.java | 5 +
.../spark/sql/connector/catalog/TableCatalog.java | 33 +-
.../spark/sql/catalyst/analysis/Analyzer.scala | 30 +-
.../sql/catalyst/analysis/ChangelogCacheKey.scala | 31 ++
.../sql/catalyst/analysis/CheckAnalysis.scala | 4 +-
.../sql/catalyst/analysis/RelationResolution.scala | 59 ++-
.../catalyst/analysis/ResolveChangelogTable.scala | 69 +---
.../analysis/UnresolveRelationsInTransaction.scala | 4 +-
.../{V2TableReference.scala => V2Reference.scala} | 119 ++++--
.../sql/connector/catalog/CatalogV2Util.scala | 23 +-
.../sql/connector/catalog/LookupCatalog.scala | 4 +-
.../spark/sql/errors/QueryCompilationErrors.scala | 10 +
.../execution/datasources/v2/ChangelogTable.scala | 75 +++-
.../catalyst/analysis/TableLookupCacheSuite.scala | 3 +-
.../sql/catalyst/analysis/V2ReferenceSuite.scala | 133 +++++++
.../sql/connector/catalog/CatalogV2UtilSuite.scala | 85 +++++
.../catalog/InMemoryChangelogCatalog.scala | 46 ++-
.../datasources/v2/ChangelogTableSuite.scala | 223 +++++++++++
.../apache/spark/sql/execution/CacheManager.scala | 10 +-
.../apache/spark/sql/execution/command/views.scala | 9 +-
.../datasources/v2/V2TableRefreshUtil.scala | 46 ++-
.../streaming/runtime/MicroBatchExecution.scala | 4 +-
.../sql/connector/ChangelogResolutionSuite.scala | 411 ++++++++++++++++++---
.../sql/connector/DataSourceV2OptionSuite.scala | 12 +-
25 files changed, 1205 insertions(+), 248 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 4eb3244b26d2..3b2d7d1ebd25 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -3330,6 +3330,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 366d173d9399..85e4a5e07368 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)`.
@@ -268,6 +271,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,
@@ -1228,7 +1232,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 =>
@@ -1238,7 +1242,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)
@@ -1246,12 +1250,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)
}
}
@@ -1268,7 +1272,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)
@@ -1279,11 +1283,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
@@ -1309,7 +1313,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)
@@ -2756,13 +2760,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 3b73eeb5c17c..3b5a89a6c074 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
@@ -488,9 +488,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 6ad8b7a6acbe..ef3d3a9a12bd 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
@@ -2372,6 +2372,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 b3bf3b8ab46d..39abf9a58621 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 a8285f6e2dd9..fadce66839e1 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.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.types.{MetadataBuilder, StructType}
import org.apache.spark.sql.util.SchemaUtils
@@ -800,7 +800,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 2f9bed533c25..d52c77602dbb 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
@@ -21,6 +21,7 @@ import scala.collection.mutable
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.ChangelogCacheKey
import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan}
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{Identifier, Table,
TableCatalog, V2TableUtil}
@@ -43,8 +44,8 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper
with Logging {
* - Data columns: Verifies captured columns align with the current schema
* - Metadata columns: Checks metadata column consistency
*
- * Tables with time travel specifications are skipped as they reference a
specific point
- * in time and don't have to be refreshed.
+ * Tables with time travel specifications and changelogs with an explicit
ending bound are
+ * skipped because they reference captured historical data.
*
* Schema validation mode depends on the underlying plan. Commands, for
instance,
* prohibit any schema changes while queries permit adding columns.
@@ -70,8 +71,8 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper
with Logging {
* - Data columns: Verifies captured columns align with the current schema
* - Metadata columns: Checks metadata column consistency
*
- * Tables with time travel specifications are skipped as they reference a
specific point
- * in time and don't have to be refreshed.
+ * Tables with time travel specifications and changelogs with an explicit
ending bound are
+ * skipped because they reference captured historical data.
*
* @param spark the currently active Spark session
* @param plan the logical plan to refresh
@@ -85,21 +86,34 @@ private[sql] object V2TableRefreshUtil extends
SQLConfHelper with Logging {
versionedOnly: Boolean,
schemaValidationMode: SchemaValidationMode): LogicalPlan = {
val currentTables = mutable.HashMap.empty[CurrentTableKey, Table]
+ val currentChangelogs = mutable.HashMap.empty[ChangelogCacheKey,
ChangelogTable]
plan transformWithSubqueries {
+ case r @ DataSourceV2Relation(changelog: ChangelogTable, _, _, _, _, _)
+ if changelog.isBounded => r
case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
if (r.isVersioned || !versionedOnly) && r.timeTravelSpec.isEmpty =>
- val stateOptions = CatalogV2Util.extractTableStateOptions(catalog,
r.options)
- val currentTable = currentTables.getOrElseUpdate((catalog, ident,
stateOptions), {
- val tableName = V2TableUtil.toQualifiedName(catalog, ident)
- lookupCachedRelation(spark, catalog, ident, r.table, stateOptions)
match {
- case Some(cached) =>
- logDebug(s"Refreshing table metadata for $tableName using shared
relation cache")
- cached.table
- case _ =>
- logDebug(s"Refreshing table metadata for $tableName using
catalog")
- CatalogV2Util.getTableWithStateOptions(catalog, ident,
stateOptions)
- }
- })
+ val currentTable = r.table match {
+ case captured: ChangelogTable =>
+ val stateOptions =
CatalogV2Util.extractChangelogStateOptions(catalog, r.options)
+ val key = ChangelogCacheKey(catalog, ident,
captured.changelogContext, stateOptions)
+ val current = currentChangelogs.getOrElseUpdate(key,
+ ChangelogTable.load(catalog, ident, captured.changelogContext,
stateOptions))
+ captured.validateRefresh(current)
+ current.copy(resolved = captured.resolved)
+ case _ =>
+ val stateOptions = CatalogV2Util.extractTableStateOptions(catalog,
r.options)
+ currentTables.getOrElseUpdate((catalog, ident, stateOptions), {
+ val tableName = V2TableUtil.toQualifiedName(catalog, ident)
+ lookupCachedRelation(spark, catalog, ident, r.table,
stateOptions) match {
+ case Some(cached) =>
+ logDebug(s"Refreshing table metadata for $tableName using
shared relation cache")
+ cached.table
+ case _ =>
+ logDebug(s"Refreshing table metadata for $tableName using
catalog")
+ CatalogV2Util.getTableWithStateOptions(catalog, ident,
stateOptions)
+ }
+ })
+ }
validateTableIdentity(currentTable, r)
validateDataColumns(currentTable, r, schemaValidationMode)
validateMetadataColumns(currentTable, r, schemaValidationMode)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
index 8c70efcb8615..4bb6124d37da 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
@@ -29,7 +29,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.{SparkException, SparkIllegalArgumentException,
SparkIllegalStateException}
import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.sql.catalyst.analysis.{ResolveDeduplicate,
V2TableReference}
+import org.apache.spark.sql.catalyst.analysis.{ResolveDeduplicate, V2Reference}
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
CurrentBatchTimestamp, CurrentDate, CurrentTimestamp, CurrentTimestampNanos,
FileSourceMetadataAttribute, LocalTimestamp, LocalTimestampNanos}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Deduplicate,
DeduplicateWithinWatermark, Distinct, FlatMapGroupsInPandasWithState,
FlatMapGroupsWithState, GlobalLimit, Join, LeafNode, LocalRelation,
LogicalPlan, Project, StreamSourceAwareLogicalPlan, TransformWithState,
TransformWithStateInPySpark}
@@ -430,7 +430,7 @@ class MicroBatchExecution(
val catalogManager =
sparkSessionForStream.sessionState.catalogManager
val streamingCatalog = catalogManager.catalog(catalog.name)
val v2Relation = DataSourceV2Relation.create(s,
Some(streamingCatalog), Some(ident))
- V2TableReference.createForWriteTarget(v2Relation)
+ V2Reference.createForWriteTarget(v2Relation)
case Some((catalog, ident)) =>
DataSourceV2Relation.create(s, Some(catalog), Some(ident))
case None => DataSourceV2Relation.create(s, None, None)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
index 082be2ac22c8..b615857b80f9 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
@@ -19,18 +19,22 @@ package org.apache.spark.sql.connector
import java.util.Collections
-import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.ResolveChangelogTable
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Window}
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.classic.Dataset
import org.apache.spark.sql.connector.catalog._
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.connector.catalog.ChangelogRange
import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference, Transform}
import org.apache.spark.sql.connector.read.ScanBuilder
-import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation, V2TableRefreshUtil}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{ArrayType, IntegerType, LongType, MapType,
StringType, StructField, StructType, TimestampType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
/**
* Tests for the CDC (Change Data Capture) analyzer resolution path:
@@ -40,25 +44,36 @@ class ChangelogResolutionSuite extends SharedSparkSession {
private val cdcCatalogName = "cdc_catalog"
private val noCdcCatalogName = "no_cdc_catalog"
+ private val ident = Identifier.of(Array.empty, "test_table")
+
+ private def cdcCatalog: ChangelogStateOptionsCatalog = {
+ spark.sessionState.catalogManager.catalog(cdcCatalogName)
+ .asInstanceOf[ChangelogStateOptionsCatalog]
+ }
override def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set(s"spark.sql.catalog.$cdcCatalogName",
- classOf[InMemoryChangelogCatalog].getName)
+ classOf[ChangelogStateOptionsCatalog].getName)
+ spark.conf.set(s"spark.sql.catalog.$cdcCatalogName.tableStateOptionKeys",
"branch")
spark.conf.set(s"spark.sql.catalog.$noCdcCatalogName",
classOf[InMemoryTableCatalog].getName)
}
override def afterAll(): Unit = {
spark.conf.unset(s"spark.sql.catalog.$cdcCatalogName")
+ spark.conf.unset(s"spark.sql.catalog.$cdcCatalogName.tableStateOptionKeys")
spark.conf.unset(s"spark.sql.catalog.$noCdcCatalogName")
super.afterAll()
}
override def beforeEach(): Unit = {
super.beforeEach()
+ cdcCatalog.changelogKeys = None
+ cdcCatalog.rejectCurrentTableLoads = false
+ cdcCatalog.clearChangeRows(ident)
+ cdcCatalog.setChangelogProperties(ident, ChangelogProperties())
val catalog =
spark.sessionState.catalogManager.catalog(cdcCatalogName).asTableCatalog
- val ident = Identifier.of(Array.empty, "test_table")
if (catalog.tableExists(ident)) {
catalog.dropTable(ident)
}
@@ -208,63 +223,336 @@ class ChangelogResolutionSuite extends
SharedSparkSession {
assert(range.endingVersion().get() == "5")
}
- test("user-defined options are forwarded to loadChangelog") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
-
- spark.read
- .option("startingVersion", "1")
- .option("customOption", "customValue")
- .changes(s"$cdcCatalogName.test_table")
-
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
- assert(opts.get.get("startingVersion") == "1")
+ gridTest("CDC resolution filters state options and retains scan options")(
+ for (streaming <- Seq(false, true); useSql <- Seq(false, true)) yield
(streaming, useSql)) {
+ case (streaming, useSql) =>
+ val cat = cdcCatalog
+ cat.resetLoadChangelogCalls()
+ val tableName = s"$cdcCatalogName.test_table"
+ val df = if (useSql) {
+ val prefix = if (streaming) "STREAM " else ""
+ sql(s"SELECT * FROM $prefix$tableName CHANGES FROM VERSION 1 " +
+ "WITH ('BrAnCh' = 'Main', 'customOption' = 'customValue')")
+ } else if (streaming) {
+ spark.readStream.option("startingVersion", "1").option("BrAnCh",
"Main")
+ .option("customOption", "customValue").changes(tableName)
+ } else {
+ spark.read.option("startingVersion", "1").option("BrAnCh", "Main")
+ .option("customOption", "customValue").changes(tableName)
+ }
+ val options = df.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.options
+ case r: StreamingRelationV2 => r.extraOptions
+ }.get
+ assert(cat.loadChangelogCalls.size == 1)
+ assert(cat.lastOptions.get.size() == 1)
+ assert(cat.lastOptions.get.get("branch") == "Main")
+ assert(options.get("branch") == "Main")
+ assert(options.get("customOption") == "customValue")
+ if (!useSql) assert(options.get("startingVersion") == "1")
+ if (!streaming) {
+ df.queryExecution.optimizedPlan
+ assert(cat.lastScanOptions.contains(options))
+ }
+ }
+
+ gridTest("CDC state option keys can inherit, override, or disable table
state keys")(
+ Seq("default", "override", "empty")) { keyMode =>
+ val cat = cdcCatalog
+ cat.changelogKeys = keyMode match {
+ case "override" => Some(java.util.Set.of("ReF"))
+ case "empty" => Some(java.util.Set.of[String]())
+ case _ => None
+ }
+ spark.read.option("startingVersion", "1").option("BRANCH", "Main")
+ .option("ref", "Dev").option("scanOption", "10")
+ .changes(s"$cdcCatalogName.test_table").queryExecution.analyzed
+
+ val expected = keyMode match {
+ case "default" => new
CaseInsensitiveStringMap(Collections.singletonMap("branch", "Main"))
+ case "override" => new
CaseInsensitiveStringMap(Collections.singletonMap("ref", "Dev"))
+ case _ => CaseInsensitiveStringMap.empty()
+ }
+ assert(cat.lastOptions.contains(expected))
}
- test("user-defined options are forwarded to loadChangelog - SQL WITH
clause") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
+ private def changeRow(id: Long, version: Long): InternalRow = InternalRow(
+ id, UTF8String.fromString(s"data-$id"),
UTF8String.fromString(Changelog.CHANGE_TYPE_INSERT),
+ version, version * 1000000L)
- sql(s"SELECT * FROM $cdcCatalogName.test_table CHANGES FROM VERSION 1 " +
- "WITH ('customOption' = 'customValue')").queryExecution.analyzed
-
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
+ test("CDC metadata loads are shared across scan options but isolate contexts
and state values") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L), changeRow(2L, 2L)))
+ cat.resetLoadChangelogCalls()
+ val tableName = s"$cdcCatalogName.test_table"
+ val df = sql(
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('BrAnCh' = 'Main', 'split-size' = '5') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('branch' = 'Main', 'split-size' = '9') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 2 " +
+ "WITH ('branch' = 'Main') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('branch' = 'main') UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 " +
+ "WITH ('branch' = 'Main', 'deduplicationMode' = 'none')")
+ val relations = df.queryExecution.analyzed.collect { case r:
DataSourceV2Relation => r }
+ val changelogs = relations.map(_.table.asInstanceOf[ChangelogTable])
+ assert(changelogs.head.changelog eq changelogs(1).changelog)
+ assert(changelogs.head.changelog ne changelogs(2).changelog)
+ assert(changelogs.head.changelog ne changelogs(3).changelog)
+ assert(changelogs.head.changelog ne changelogs(4).changelog)
+ assert(relations.take(2).map(_.options.get("split-size")) == Seq("5", "9"))
+ assert(cat.loadChangelogCalls.size == 4)
+ QueryTest.checkAnswer(
+ df, Seq.fill(4)(Row(1L)) ++ Seq.fill(5)(Row(2L)), checkToRDD = false)
+
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val refreshed = V2TableRefreshUtil.refresh(spark,
df.queryExecution.analyzed)
+ assert(cat.loadChangelogCalls.size == 4)
+ assert(cat.loadTableCalls.isEmpty)
+ val refreshedChanges = refreshed.collect {
+ case r: DataSourceV2Relation => r.table.asInstanceOf[ChangelogTable]
+ }
+ assert(refreshedChanges.head.changelog eq refreshedChanges(1).changelog)
+ assert(refreshedChanges.map(_.changelogContext) ==
changelogs.map(_.changelogContext))
+ assert(refreshedChanges.forall(_.resolved))
+ }
+
+ test("ordinary and CDC metadata caches remain separate") {
+ val tableName = s"$cdcCatalogName.test_table"
+ sql(s"INSERT INTO $tableName VALUES (100, 'current')")
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ cat.resetLoadTableCalls()
+ cat.resetLoadChangelogCalls()
+ val df = sql(s"SELECT id FROM $tableName UNION ALL " +
+ s"SELECT id FROM $tableName CHANGES FROM VERSION 1 TO VERSION 10")
+ val relations = df.queryExecution.analyzed.collect { case r:
DataSourceV2Relation => r }
+ assert(relations.count(_.table.isInstanceOf[ChangelogTable]) == 1)
+ assert(cat.loadTableCalls.size == 1)
+ assert(cat.loadChangelogCalls.size == 1)
+ QueryTest.checkAnswer(df, Seq(Row(100L), Row(1L)), checkToRDD = false)
+ }
+
+ test("refreshing cached CDC data preserves CDC reads and ordinary table
cache isolation") {
+ val tableName = s"$cdcCatalogName.test_table"
+ val cdcQuery = s"SELECT * FROM $tableName CHANGES FROM VERSION 1"
+ sql(s"INSERT INTO $tableName VALUES (100, 'current')")
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ val cached = sql(cdcQuery).cache()
+ val cacheManager = spark.sharedState.cacheManager
+ try {
+ checkAnswer(cached.select("id"), Seq(Row(1L)))
+ assert(cacheManager.lookupCachedData(spark.table(tableName)).isEmpty)
+ checkAnswer(spark.table(tableName).select("id"), Seq(Row(100L)))
+
+ cat.addChangeRows(ident, Seq(changeRow(2L, 2L)))
+ spark.catalog.refreshTable(tableName)
+
+ assert(cacheManager.numCachedEntries == 1)
+ assert(cacheManager.lookupCachedData(spark.table(tableName)).isEmpty)
+ checkAnswer(spark.table(tableName).select("id"), Seq(Row(100L)))
+
+ assert(cacheManager.lookupCachedData(cached).isEmpty)
+
+ val fresh = sql(cdcQuery)
+ assert(cacheManager.lookupCachedData(fresh).isDefined)
+ checkAnswer(fresh.select("id"), Seq(Row(1L), Row(2L)))
+
+ val differentRange = sql(
+ s"SELECT * FROM $tableName CHANGES FROM VERSION 2")
+ assert(cacheManager.lookupCachedData(differentRange).isEmpty)
+ } finally {
+ spark.catalog.clearCache()
+ }
}
- test("user-defined options are forwarded to loadChangelog -
DataStreamReader") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
-
- spark.readStream
- .option("startingVersion", "1")
- .option("customOption", "customValue")
- .changes(s"$cdcCatalogName.test_table")
- .queryExecution.analyzed
-
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
- assert(opts.get.get("startingVersion") == "1")
+ test("bounded CDC reads do not load the latest ordinary table") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L), changeRow(10L, 10L),
changeRow(100L, 100L)))
+ cat.rejectCurrentTableLoads = true
+ cat.resetLoadTableCalls()
+ cat.resetLoadChangelogCalls()
+ val df = sql(
+ s"SELECT id FROM $cdcCatalogName.test_table CHANGES FROM VERSION 1 TO
VERSION 10")
+ QueryTest.checkAnswer(df, Seq(Row(1L), Row(10L)), checkToRDD = false)
+ assert(cat.loadTableCalls.isEmpty)
+ assert(cat.loadChangelogCalls.size == 1)
+ val refreshed = V2TableRefreshUtil.refresh(spark,
df.queryExecution.analyzed)
+ assert(refreshed.fastEquals(df.queryExecution.analyzed))
+ assert(cat.loadTableCalls.isEmpty)
+ assert(cat.loadChangelogCalls.size == 1)
+ }
+
+ gridTest("CDC refresh reloads only open-ended ranges")(
+ Seq[(ChangelogRange, Boolean)](
+ new ChangelogRange.VersionRange(
+ "1", java.util.Optional.of("10"), true, true) -> true,
+ new ChangelogRange.TimestampRange(
+ 1000000L, java.util.Optional.of(java.lang.Long.valueOf(10000000L)),
true, true) -> true,
+ new ChangelogRange.VersionRange(
+ "1", java.util.Optional.empty[String](), true, true) -> false,
+ new ChangelogRange.TimestampRange(
+ 1000000L, java.util.Optional.empty[java.lang.Long](), true, true) ->
false,
+ new ChangelogRange.UnboundedRange() -> false)) { case (range, bounded)
=>
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ val context = new ChangelogContext(
+ range, ChangelogContext.DeduplicationMode.DROP_CARRYOVERS, false)
+ val stateOptions = new
CaseInsensitiveStringMap(Collections.singletonMap("branch", "Main"))
+ val scanOptions = new java.util.HashMap[String,
String](stateOptions.asCaseSensitiveMap())
+ scanOptions.put("split-size", "5")
+ val captured = ChangelogTable(
+ cat.loadChangelog(ident, context, stateOptions), context, resolved =
true)
+ val original = DataSourceV2Relation.create(
+ captured, Some(cat), Some(ident), new
CaseInsensitiveStringMap(scanOptions))
+ cat.addChangeRows(ident, Seq(changeRow(2L, 2L)))
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+
+ val refreshed = V2TableRefreshUtil.refresh(spark, original)
+ .asInstanceOf[DataSourceV2Relation]
+ assert(refreshed.options == original.options)
+ assert(cat.loadTableCalls.isEmpty)
+ if (bounded) {
+ assert(refreshed eq original)
+ assert(refreshed.table eq captured)
+ assert(cat.loadChangelogCalls.isEmpty)
+ } else {
+ assert(refreshed.table ne captured)
+ assert(cat.loadChangelogCalls == Seq((context, stateOptions)))
+ val current = refreshed.table.asInstanceOf[ChangelogTable]
+ assert(current.changelogContext == context)
+ assert(current.resolved)
+ }
+ val expected = if (bounded) Seq(Row(1L)) else Seq(Row(1L), Row(2L))
+ QueryTest.checkAnswer(
+ Dataset.ofRows(spark, refreshed).select("id"), expected, checkToRDD =
false)
+ }
+
+ test("versioned-only refresh leaves an unversioned changelog captured") {
+ val cat = cdcCatalog
+ val analyzed = spark.read.option("startingVersion", "1")
+ .changes(s"$cdcCatalogName.test_table").queryExecution.analyzed
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val refreshed = V2TableRefreshUtil.refresh(spark, analyzed, versionedOnly
= true)
+ assert(refreshed.fastEquals(analyzed))
+ assert(cat.loadChangelogCalls.isEmpty)
+ assert(cat.loadTableCalls.isEmpty)
+ }
+
+ test("CDC refresh uses CDC state keys and validates the captured schema") {
+ val cat = cdcCatalog
+ cat.changelogKeys = Some(java.util.Set.of("ReF"))
+ val analyzed = spark.read.option("startingVersion", "1")
+ .option("branch", "Main").option("ref", "History").option("split-size",
"5")
+ .changes(s"$cdcCatalogName.test_table").queryExecution.analyzed
+ val context = cat.lastChangelogContext.get
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val refreshed = V2TableRefreshUtil.refresh(spark, analyzed)
+ assert(cat.loadTableCalls.isEmpty)
+ assert(cat.loadChangelogCalls.size == 1)
+ assert(cat.lastChangelogContext.contains(context))
+ assert(cat.lastOptions.get.size() == 1)
+ assert(cat.lastOptions.get.get("ref") == "History")
+ val relation = refreshed.collectFirst { case r: DataSourceV2Relation => r
}.get
+ assert(relation.options.get("split-size") == "5")
+ assert(relation.table.asInstanceOf[ChangelogTable].resolved)
+
+ cat.alterTable(ident, TableChange.deleteColumn(Array("data"), false))
+ checkError(
+ intercept[AnalysisException] { V2TableRefreshUtil.refresh(spark,
analyzed) },
+ condition = "INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ parameters = Map(
+ "tableName" -> s"`$cdcCatalogName`.`test_table`",
+ "errors" -> "- `data` STRING has been removed"))
+ }
+
+ test("CDC temp views preserve distinct contexts when metadata is reloaded") {
+ withTempView("cdc_first", "cdc_second") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L), changeRow(2L, 2L)))
+ val tableName = s"$cdcCatalogName.test_table"
+ sql(s"SELECT * FROM $tableName CHANGES FROM VERSION 1")
+ .createOrReplaceTempView("cdc_first")
+ sql(s"SELECT * FROM $tableName CHANGES FROM VERSION 2")
+ .createOrReplaceTempView("cdc_second")
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+ val df = sql("SELECT id FROM cdc_first UNION ALL SELECT id FROM
cdc_second " +
+ "UNION ALL SELECT id FROM cdc_first")
+ val changelogs = df.queryExecution.analyzed.collect {
+ case r: DataSourceV2Relation => r.table.asInstanceOf[ChangelogTable]
+ }
+ assert(changelogs.map(_.changelogContext).distinct.size == 2)
+ assert(changelogs.head.changelog eq changelogs.last.changelog)
+ assert(changelogs.forall(_.resolved))
+ assert(cat.loadChangelogCalls.size == 2)
+ assert(cat.loadTableCalls.isEmpty)
+ QueryTest.checkAnswer(
+ df, Seq(Row(1L), Row(1L), Row(2L), Row(2L), Row(2L)), checkToRDD =
false)
+ }
}
- test("user-defined options are forwarded to loadChangelog - streaming SQL") {
- val cat = spark.sessionState.catalogManager
- .catalog(cdcCatalogName)
- .asInstanceOf[InMemoryChangelogCatalog]
-
- sql(s"SELECT * FROM STREAM $cdcCatalogName.test_table CHANGES FROM VERSION
1 " +
- "WITH ('customOption' = 'customValue')").queryExecution.analyzed
+ gridTest("bounded CDC temp views preserve captured reads")(
+ Seq("VERSION 1 TO VERSION 10",
+ "TIMESTAMP '1970-01-01 00:00:01' TO TIMESTAMP '1970-01-01 00:00:10'"))
{ range =>
+ withTempView("bounded_cdc") {
+ val cat = cdcCatalog
+ cat.addChangeRows(ident, Seq(changeRow(1L, 1L)))
+ val changes = sql(s"SELECT * FROM $cdcCatalogName.test_table CHANGES
FROM $range")
+ val captured = changes.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.get
+ changes.createOrReplaceTempView("bounded_cdc")
+ cat.addChangeRows(ident, Seq(changeRow(2L, 2L)))
+ cat.resetLoadChangelogCalls()
+ cat.resetLoadTableCalls()
+
+ val fromView = spark.table("bounded_cdc")
+ val current = fromView.queryExecution.analyzed.collectFirst {
+ case r: DataSourceV2Relation => r.table
+ }.get
+ assert(current eq captured)
+ assert(cat.loadChangelogCalls.isEmpty)
+ assert(cat.loadTableCalls.isEmpty)
+ QueryTest.checkAnswer(fromView.select("id"), Seq(Row(1L)), checkToRDD =
false)
+ }
+ }
- val opts = cat.lastOptions
- assert(opts.isDefined)
- assert(opts.get.get("customOption") == "customValue")
+ test("CDC refresh preserves post-processing and rejects changed row-version
metadata") {
+ val tableIdent = recreatePostProcessingTable()
+ val cat = cdcCatalog
+ val properties = ChangelogProperties(
+ containsCarryoverRows = true,
+ rowIdNames = Seq("id"),
+ rowVersionName = Some("row_commit_version"))
+ cat.setChangelogProperties(tableIdent, properties)
+ cat.addChangeRows(tableIdent, Seq(
+ InternalRow(1L, 1L, UTF8String.fromString(Changelog.CHANGE_TYPE_DELETE),
1L, 1000000L),
+ InternalRow(1L, 1L, UTF8String.fromString(Changelog.CHANGE_TYPE_INSERT),
1L, 1000000L)))
+ val analyzed = spark.read.option("startingVersion", "1")
+
.changes(s"$cdcCatalogName.test_table").select("id").queryExecution.analyzed
+ assert(analyzed.exists(_.isInstanceOf[Window]))
+ cat.addChangeRows(tableIdent, Seq(
+ InternalRow(2L, 2L, UTF8String.fromString(Changelog.CHANGE_TYPE_INSERT),
2L, 2000000L)))
+ val refreshed = V2TableRefreshUtil.refresh(spark, analyzed)
+ assert(ResolveChangelogTable(refreshed).fastEquals(refreshed))
+ QueryTest.checkAnswer(Dataset.ofRows(spark, refreshed), Seq(Row(2L)),
checkToRDD = false)
+
+ cat.setChangelogProperties(tableIdent,
+ properties.copy(rowVersionName = Some("_commit_version")))
+ checkError(
+ intercept[AnalysisException] { V2TableRefreshUtil.refresh(spark,
analyzed) },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.CHANGELOG_METADATA_MISMATCH",
+ parameters = Map(
+ "tableName" -> s"`$cdcCatalogName`.`test_table_changelog`",
+ "changedProperties" -> "rowVersion"))
}
//
===========================================================================
@@ -550,6 +838,21 @@ class ChangelogResolutionSuite extends SharedSparkSession {
}
+/** Supports configurable CDC keys and detects accidental ordinary-table loads
in CDC paths. */
+class ChangelogStateOptionsCatalog extends InMemoryChangelogCatalog {
+ var changelogKeys: Option[java.util.Set[String]] = None
+ var rejectCurrentTableLoads: Boolean = false
+
+ override def changelogStateOptionKeys(): java.util.Set[String] = {
+ changelogKeys.getOrElse(tableStateOptionKeys())
+ }
+
+ override def loadTable(ident: Identifier): Table = {
+ assert(!rejectCurrentTableLoads, "CDC must not load the current ordinary
table")
+ super.loadTable(ident)
+ }
+}
+
/**
* Test-only [[Changelog]] implementation that returns a hand-crafted schema.
Used to
* exercise [[ChangelogTable]]'s schema validation without going through a
real catalog.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
index 8b7fd2bb80e3..42ff027431ca 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.analysis.{
RelationCache,
RelationResolution,
UnresolvedRelation,
- V2TableReference}
+ V2Reference}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
import org.apache.spark.sql.connector.catalog.{
@@ -1634,11 +1634,11 @@ class DataSourceV2OptionSuite extends
DatasourceV2SQLBase {
.analyzed
.collectFirst { case r: DataSourceV2Relation => r }
.getOrElse(fail("expected a v2 relation"))
- val readRef = V2TableReference.createForTransaction(original)
- val otherReadRef = V2TableReference.createForTransaction(original.copy(
+ val readRef = V2Reference.createForTransaction(original)
+ val otherReadRef = V2Reference.createForTransaction(original.copy(
options = new CaseInsensitiveStringMap(
java.util.Map.of("snapshot", "s1", "split-size", "9"))))
- val writeRef = V2TableReference.createForWriteTarget(original)
+ val writeRef = V2Reference.createForWriteTarget(original)
var sharedRelationCacheLookups = 0
val sharedRelationCache: RelationCache = (_, _, _, _, _) => {
sharedRelationCacheLookups += 1
@@ -1710,8 +1710,8 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase
{
.analyzed
.collectFirst { case r: DataSourceV2Relation => r }
.getOrElse(fail("expected a v2 relation"))
- val initialRef = V2TableReference.createForTempView(cached,
Seq("state_view"))
- val otherOptionsRef = V2TableReference.createForTempView(
+ val initialRef = V2Reference.createForTempView(cached, Seq("state_view"))
+ val otherOptionsRef = V2Reference.createForTempView(
cached.copy(options = new CaseInsensitiveStringMap(
java.util.Map.of("snapshot", "s1", "split-size", "9"))),
Seq("state_view"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
