szehon-ho commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3306150690
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
Review Comment:
nit: naming -- Spark production traits with this shape (abstract `protected
def` collaboration points + concrete `protected def` helpers) consistently use
the `Base` suffix: `DataSourceV2ScanExecBase`, `ShowCreateTableCommandBase`,
`DescribeCommandBase`, `WriteFilesExecBase`, `WindowExecBase`, etc. The `Mixin`
suffix is almost entirely confined to test utilities (`StorageRootMixin`,
`SparkErrorTestMixin`, ...). Consider `AutoCdcMergeWriteBase` for consistency.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
Review Comment:
Side-effect in trait body is fragile -- suggest moving this call into the
concrete `Scd1MergeStreamingWrite.startStream()`.
Running `requireDestinationSupportsRowLevelOps()` at trait-init time
dereferences abstract members (`spark`, `destination`) during the linearized
constructor of any class mixing this trait in. If a subclass implements those
as `val`s (including constructor `val` params declared *after* the trait in
linearization order), they are not initialized yet when the trait body runs --
classic Scala trait-init-order NPE.
None of the comparable Spark `*Base` traits do this; they only declare
helpers and let the concrete class invoke them at the right time. Calling
`requireDestinationSupportsRowLevelOps()` at the top of `startStream()` (or in
an explicit `init()` hook) gives the same eager-failure semantics without the
construction-order coupling.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
Review Comment:
Concern: `CREATE TABLE IF NOT EXISTS` is a silent no-op against a stale aux
schema.
After dropping `validateNoAutoCdcKeyDrift` in `f17c7d0`, this helper is the
only thing that runs against a previously-created aux table. If a prior run
created the aux table with a wider key set (e.g. `(region, id, _cdc_metadata)`)
and the flow now declares `keys = Seq("id")`, the existing aux table persists,
the MERGE joins only on `id`, and tombstones from `(region=X, id=1)` apply to
every event with `id=1` regardless of `region` -- silent semantic corruption.
Two options worth considering:
1. Cheapest: branch on `catalog.tableExists` and, when the table exists,
load it and assert that its column set equals `auxiliaryTableSchema`'s column
set. Much lighter than the removed `numKeyColumns` validation, but enough to
catch shrink/swap.
2. Alternatively, keep current behavior but document in the SDP programming
guide that **changing AutoCDC keys on an existing target requires a full
refresh**, so users know how to recover.
Also a minor perf nit: today this issues `CREATE TABLE IF NOT EXISTS` on
every pipeline start; gating on `tableExists` would skip the redundant SQL
after the first run.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
+ val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+ // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
+ // target's format is unspecified (None), omit the USING clause and fall
back to the
+ // session's default source provider.
+ val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
+ spark.sql(
+ s"""CREATE TABLE IF NOT EXISTS
+ |${auxIdent.quotedString}
+ |(${auxiliaryTableSchema.toDDL}) $usingClause""".stripMargin
+ )
+ auxIdent
+ }
+
+ /**
+ * Validate that the target table's underlying connector implements
+ * [[SupportsRowLevelOperations]], which is the V2 connector contract for
MERGE/UPDATE/DELETE
+ * with rewrite - all operations that the AutoCDC transformation executes.
+ */
+ private def requireDestinationSupportsRowLevelOps(): Unit = {
+ val (catalog, v2Identifier) = resolveTableCatalog(spark,
destination.identifier)
+ val destinationTable = catalog.loadTable(v2Identifier)
+
+ if (!destinationTable.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> destination.identifier.quotedString,
+ "format" -> destination.format.getOrElse("<session default>")
+ )
+ )
+ }
+ }
+
+ private def resolveTableCatalog(
+ spark: SparkSession,
+ ident: TableIdentifier): (TableCatalog, Identifier) = {
+ val catalogManager = spark.sessionState.catalogManager
+ val catalog = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ .asInstanceOf[TableCatalog]
+ val namespace = ident.database.getOrElse(
+ throw SparkException.internalError(
+ s"Cannot resolve table identifier ${ident.quotedString}: namespace is
unspecified."
+ )
+ )
+ (catalog, Identifier.of(Array(namespace), ident.table))
+ }
+}
+
+/**
+ * A [[StreamingFlowExecution]] that applies a CDC event stream to a target
[[Table]] via
+ * SCD Type 1 MERGE semantics.
+ */
+class Scd1MergeStreamingWrite(
+ val identifier: TableIdentifier,
+ val flow: AutoCdcMergeFlow,
+ val graph: DataflowGraph,
+ val updateContext: PipelineUpdateContext,
+ val checkpointPath: String,
+ val trigger: Trigger,
+ val destination: Table,
+ val sqlConf: Map[String, String]
+) extends StreamingFlowExecution with AutoCdcMergeWriteMixin {
+
+ override def getOrigin: QueryOrigin = flow.origin
+
+ override protected def changeArgs: ChangeArgs = flow.changeArgs
+
+ /**
+ * Resolved Spark [[DataType]] of the sequencing expression.
+ */
+ private def sequencingType: DataType =
Review Comment:
nit: this can be a `lazy val`. `flow.df` and `flow.changeArgs.sequencing`
are fixed for the stream's lifetime, so the schema lookup does not need to
repeat on every access. A `lazy val` better captures intent and avoids
re-evaluation each time `Scd1BatchProcessor` is constructed.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -303,6 +303,25 @@ object DatasetManager extends Logging {
context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
}
+ if (isFullRefresh) {
+ // On full refresh, drop the AutoCDC auxiliary state associated with
this table (if any) so
+ // that stale delete-tracking data and table properties are not carried
forward into the new
+ // table generation. We unconditionally issue the DROP for every
fully-refreshed target; for
+ // non-AutoCDC tables this is a no-op because
[[AutoCdcAuxiliaryTable.identifier]] derives
+ // its name from [[AutoCdcReservedNames.prefix]], which is reserved
across AutoCDC and
+ // therefore cannot collide with a user-managed table.
+
+ // Intentionally DROP and not TRUNCATE for two reasons; First, the
auxiliary table may
+ // contain table properties that represent stateful information (ex. SCD
key count) that
Review Comment:
Stale comment -- `numKeyColumns` / "SCD key count" properties no longer
exist.
Commit `f17c7d0` removed the `numKeyColumns` aux table property and the
entire `parseRecordedNumKeyColumns` path. The first reason for "DROP, not
TRUNCATE" no longer applies -- the aux table today only carries tombstone
*rows*, not stateful properties. Suggest reducing to just the schema-evolution
reason, e.g.:
```scala
// Intentionally DROP and not TRUNCATE: the auxiliary table is internal and
not part of the
// dataflow graph, so it does not go through schema evolution like other
tables. On a full
// refresh we drop it so its schema can be recomputed from the (possibly
evolved) target.
```
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
+ val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+ // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
+ // target's format is unspecified (None), omit the USING clause and fall
back to the
+ // session's default source provider.
+ val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
+ spark.sql(
+ s"""CREATE TABLE IF NOT EXISTS
+ |${auxIdent.quotedString}
+ |(${auxiliaryTableSchema.toDDL}) $usingClause""".stripMargin
+ )
+ auxIdent
+ }
+
+ /**
+ * Validate that the target table's underlying connector implements
+ * [[SupportsRowLevelOperations]], which is the V2 connector contract for
MERGE/UPDATE/DELETE
+ * with rewrite - all operations that the AutoCDC transformation executes.
+ */
+ private def requireDestinationSupportsRowLevelOps(): Unit = {
+ val (catalog, v2Identifier) = resolveTableCatalog(spark,
destination.identifier)
+ val destinationTable = catalog.loadTable(v2Identifier)
+
+ if (!destinationTable.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> destination.identifier.quotedString,
+ "format" -> destination.format.getOrElse("<session default>")
+ )
+ )
+ }
+ }
+
+ private def resolveTableCatalog(
+ spark: SparkSession,
+ ident: TableIdentifier): (TableCatalog, Identifier) = {
+ val catalogManager = spark.sessionState.catalogManager
+ val catalog = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ .asInstanceOf[TableCatalog]
+ val namespace = ident.database.getOrElse(
+ throw SparkException.internalError(
+ s"Cannot resolve table identifier ${ident.quotedString}: namespace is
unspecified."
+ )
+ )
+ (catalog, Identifier.of(Array(namespace), ident.table))
+ }
+}
+
+/**
+ * A [[StreamingFlowExecution]] that applies a CDC event stream to a target
[[Table]] via
+ * SCD Type 1 MERGE semantics.
+ */
+class Scd1MergeStreamingWrite(
+ val identifier: TableIdentifier,
+ val flow: AutoCdcMergeFlow,
+ val graph: DataflowGraph,
+ val updateContext: PipelineUpdateContext,
+ val checkpointPath: String,
+ val trigger: Trigger,
+ val destination: Table,
+ val sqlConf: Map[String, String]
+) extends StreamingFlowExecution with AutoCdcMergeWriteMixin {
+
+ override def getOrigin: QueryOrigin = flow.origin
+
+ override protected def changeArgs: ChangeArgs = flow.changeArgs
+
+ /**
+ * Resolved Spark [[DataType]] of the sequencing expression.
+ */
+ private def sequencingType: DataType =
+ flow.df.select(flow.changeArgs.sequencing).schema.head.dataType
+
+ override def startStream(): StreamingQuery = {
+ val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
+ val auxiliaryTableIdentifier = createAuxiliaryTableIfNotExists(spark =
updateContext.spark)
+
+ sourceChangeDataFeed.writeStream
+ .queryName(displayName)
+ .option("checkpointLocation", checkpointPath)
+ .trigger(trigger)
+ .foreachBatch((batch: Dataset[Row], batchId: Long) => {
+ val foreachBatchHandler = Scd1ForeachBatchHandler(
+ batchProcessor = Scd1BatchProcessor(
+ changeArgs = flow.changeArgs,
+ resolvedSequencingType = sequencingType
+ ),
+ auxiliaryTableIdentifier = auxiliaryTableIdentifier,
+ targetTableIdentifier = destination.identifier
+ )
+ foreachBatchHandler.execute(batch, batchId)
+ })
Review Comment:
nit: hoist `Scd1BatchProcessor` (and `Scd1ForeachBatchHandler` if its
identifiers are also stable) out of the `foreachBatch` closure.
`Scd1BatchProcessor(changeArgs = flow.changeArgs, resolvedSequencingType =
sequencingType)` is immutable per stream. Constructing a fresh one inside
`foreachBatch` allocates per microbatch with no behavioral benefit. Building
both once outside the closure and capturing them would be cleaner and shaves a
small per-batch cost.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala:
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end tests that exercise interactions between separate AutoCDC
pipelines (i.e.
+ * distinct [[DataflowGraph]] / [[TestPipelineUpdateContext]] invocations)
sharing the same
+ * v2 catalog. These complement the single-pipeline AutoCDC suites by
validating the
+ * boundary semantics between independently-deployed pipelines.
+ *
+ * Each test constructs two graphs and runs them sequentially. In real
deployments these
+ * could be two different pipeline definitions writing into the same
metastore; the tests
+ * here verify that AutoCDC's per-target catalog state (target table,
auxiliary table,
+ * schema invariants) behaves correctly across these pipeline boundaries.
+ */
+class AutoCdcScd1MultiPipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ test("two AutoCDC pipelines targeting separate tables maintain independent
target and " +
+ "auxiliary tables") {
+ val session = spark
+ import session.implicits._
+
+ // Two distinct target tables created up-front.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_a " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_b " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ // Pipeline #1 only knows about `t_a`. Its auxiliary table
cat.ns1.__auxiliary_..._t_a
Review Comment:
nit: stale aux-table name in the comment.
The aux table is now named via `AutoCdcReservedNames.prefix` +
`aux_state_<target>`, so the actual identifier here is
`cat.ns1.__spark_autocdc_aux_state_t_a`. Suggest updating the comment to match
(or just reference `AutoCdcAuxiliaryTable.identifier(...)` and let the reader
resolve it).
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,708 @@
+/*
+ * 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.pipelines.graph
+
+import java.sql.Timestamp
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.pipelines.autocdc.{
+ ColumnSelection,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests covering AutoCDC's interaction with schema evolution across pipeline
runs. The
+ * suite documents the supported additive cases (new top-level columns, new
nested fields
+ * in array-of-struct, broadening / narrowing column selection) and the cases
that fail
+ * loudly today (subtractive nested evolution, type-incompatible changes,
case-only
+ * renames).
+ *
+ * These behaviors are largely inherited from the lower layers
(`SchemaMergingUtils` for
+ * schema merge, the v2 writer's column-resolution layer for nested-field
handling) rather
+ * than implemented in AutoCDC itself; the tests here serve as the contract
for AutoCDC's
+ * observable behavior on top of those layers.
+ */
+class AutoCdcScd1SchemaEvolutionSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ test("a nullable non-key column merges correctly with mixed NULL and
non-NULL values") {
+ val session = spark
+ import session.implicits._
+
+ // Single MemoryStream with `email` as nullable from the start. Run #1
emits a row with
+ // a NULL email; run #2 emits an upsert with a non-NULL email.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Option[String], Long)]
+ def buildCtx(): TestGraphRegistrationContext = new
TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "email",
"version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ // Run #1: insert with NULL email.
+ stream.addData((1, "alice", None, 1L))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", null, 1L, cdcMeta(None, Some(1L))))
+ )
+
+ // Run #2: upsert with non-NULL email at higher seq replaces the row.
+ stream.addData((1, "alice2", Some("[email protected]"), 2L))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice2", "[email protected]", 2L, cdcMeta(None, Some(2L))))
+ )
+ }
+
+ test("widening a non-key column's type between runs fails with " +
+ "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
+ val session = spark
+ import session.implicits._
+
+ // Changing a non-key column's type between pipeline runs is rejected by
+ // `SchemaMergingUtils` with CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE even when
the new type
+ // is strictly wider. Users must full-refresh the target to change column
types.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, age INT, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Int, Long)]
+ stream1.addData((1, 30, 1L))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream1.toDF().toDF("id", "age", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ runPipeline(ctx1)
+
+ // Run #2: widen `age` from Int to Long.
+ val stream2 = MemoryStream[(Int, Long, Long)]
+ stream2.addData((1, 31L, 2L))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("id", "age", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE",
+ sqlState = Some("42825"),
+ // `left` is the persisted (run #1) INT type; `right` is run #2's
widened BIGINT.
+ parameters = Map(
+ "left" -> "\"INT\"",
+ "right" -> "\"BIGINT\""
+ )
+ )
+ }
+
+ test("narrowing a non-key column's type between runs fails with " +
+ "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
+ val session = spark
+ import session.implicits._
+
+ // Mirror image of the widening test above: changing a non-key column's
type between
+ // pipeline runs is rejected by SchemaMergingUtils with
CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE
+ // even when the new type is strictly narrower.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, payload BIGINT, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long, Long)]
+ stream1.addData((1, 100L, 1L))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream1.toDF().toDF("id", "payload", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ runPipeline(ctx1)
+
+ // Run #2: narrow `payload` from Long (BIGINT) to Int (INT).
+ val stream2 = MemoryStream[(Int, Int, Long)]
+ stream2.addData((1, 5, 2L))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("id", "payload", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE",
+ sqlState = Some("42825"),
+ // `left` is the persisted (run #1) BIGINT type; `right` is run #2's
narrowed INT.
+ parameters = Map(
+ "left" -> "\"BIGINT\"",
+ "right" -> "\"INT\""
+ )
+ )
+ }
+
+ test("a new top-level nullable column appearing in the source DF between
runs is " +
+ "added to the target") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ // Single MemoryStream of (id, name, email, version) shared across runs so
the streaming
+ // checkpoint can resume cleanly. Run #1's flow drops `email` so the
source's resolved DF
+ // schema is 3 columns; run #2 keeps all 4. The MemoryStream's underlying
tuple schema is
+ // unchanged (only the downstream projection differs), so the source
identity that the
+ // OffsetSeqLog records is stable across runs.
+ val stream = MemoryStream[(Int, String, Option[String], Long)]
+ def buildCtx(includeEmail: Boolean): TestGraphRegistrationContext =
+ new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ val sourceDf = stream.toDF().toDF("id", "name", "email", "version")
+ val projectedDf = if (includeEmail) sourceDf else
sourceDf.drop("email")
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(projectedDf),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ // Run #1: source projects (id, name, version). Target schema is unchanged.
+ stream.addData((1, "alice", None, 1L))
+ runPipeline(buildCtx(includeEmail = false))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, cdcMeta(None, Some(1L))))
+ )
+
+ // Run #2: source projects (id, name, email, version). mergeSchemas
appends `email` to
+ // the target (StructType.merge keeps the left schema's order and appends
right-only
+ // fields); existing rows get NULL for the new column.
+ stream.addData((2, "bob", Some("[email protected]"), 2L))
+ runPipeline(buildCtx(includeEmail = true))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, cdcMeta(None, Some(1L)), null),
+ Row(2, "bob", 2L, cdcMeta(None, Some(2L)), "[email protected]")
+ )
+ )
+ }
+
+ test("broadening the column selection between runs adds the newly-included
column to " +
+ "the target") {
+ val session = spark
+ import session.implicits._
+
+ // Source DF schema is fixed at (id, name, email, version) across both
runs. Only the
+ // `columnSelection` knob differs: run #1 includes (id, name, version);
run #2 selects
+ // None (= all source columns). mergeSchemas adds `email` to the target
via the same
+ // generic SDP path as the new-source-column case, but driven by the
+ // [[ColumnSelection]] knob rather than the source DF's own schema.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, String, Long)]
+ def buildCtx(selection: Option[ColumnSelection]):
TestGraphRegistrationContext =
+ new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "email",
"version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ columnSelection = selection
+ ))
+ }
+
+ // Run #1: only (id, name, version) selected; `email` is dropped before
the MERGE.
+ stream.addData((1, "alice", "ignored", 1L))
+ runPipeline(buildCtx(selection = Some(ColumnSelection.IncludeColumns(
+ Seq("id", "name", "version").map(UnqualifiedColumnName(_))
+ ))))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, cdcMeta(None, Some(1L))))
+ )
+
+ // Run #2: broaden to no selection. mergeSchemas adds `email`; existing
rows get NULL,
+ // new rows get the actual value.
+ stream.addData((2, "bob", "[email protected]", 2L))
+ runPipeline(buildCtx(selection = None))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, cdcMeta(None, Some(1L)), null),
+ Row(2, "bob", 2L, cdcMeta(None, Some(2L)), "[email protected]")
+ )
+ )
+ }
+
+ test("narrowing the column selection between runs preserves the dropped
column on " +
+ "existing rows and leaves it NULL on new rows") {
+ val session = spark
+ import session.implicits._
+
+ // Validates the additive-only column-selection contract on the narrowing
side:
+ // tightening `columnSelection` between runs leaves the dropped column in
place at the
+ // schema level (SDP's `SchemaMergingUtils.mergeSchemas` is a union, never
a subtraction).
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, String, Long)]
+ def buildCtx(selection: Option[ColumnSelection]):
TestGraphRegistrationContext =
+ new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "email",
"version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ columnSelection = selection
+ ))
+ }
+
+ // Run #1: include all columns; populate `email` for key=1.
+ stream.addData((1, "alice", "[email protected]", 1L))
+ runPipeline(buildCtx(selection = None))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", "[email protected]", 1L, cdcMeta(None, Some(1L))))
+ )
+
+ // Run #2: narrow the selection to drop `email`. The merge omits `email`
from both
+ // INSERT and UPDATE assignment maps; key=1's `email` is preserved at
"[email protected]" while
+ // key=2 is inserted with `email = NULL`.
+ stream.addData((2, "bob", "ignored", 2L))
+ runPipeline(buildCtx(selection = Some(ColumnSelection.IncludeColumns(
+ Seq("id", "name", "version").map(UnqualifiedColumnName(_))
+ ))))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", "[email protected]", 1L, cdcMeta(None, Some(1L))),
+ Row(2, "bob", null, 2L, cdcMeta(None, Some(2L)))
+ )
+ )
+ }
+
+ test("a top-level column dropped from the source DF between runs is
preserved on " +
+ "existing rows and left NULL on new rows") {
+ val session = spark
+ import session.implicits._
+
+ // Symmetric to the new-source-column case (which exercises the source DF
*gaining* a
+ // column). Validates that the additive-only column-selection contract
holds when the
+ // narrowing is driven by the source DF's own schema shrinking, rather
than by a
+ // tightening [[ChangeArgs.columnSelection]].
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ // Same `MemoryStream[(Int, String, Option[String], Long)]` shape across
runs; runs
+ // differ in whether `email` is kept in the projected source DF.
+ val stream = MemoryStream[(Int, String, Option[String], Long)]
+ def buildCtx(includeEmail: Boolean): TestGraphRegistrationContext =
+ new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ val sourceDf = stream.toDF().toDF("id", "name", "email", "version")
+ val projectedDf = if (includeEmail) sourceDf else
sourceDf.drop("email")
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(projectedDf),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ // Run #1: wide source DF (id, name, email, version). mergeSchemas appends
`email` to
+ // the target.
+ stream.addData((1, "alice", Some("[email protected]"), 1L))
+ runPipeline(buildCtx(includeEmail = true))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, cdcMeta(None, Some(1L)), "[email protected]"))
+ )
+
+ // Run #2: source DF drops `email` upstream of the flow. Target still has
`email`
+ // (`StructType.merge` is additive-only); the merge omits `email` from
both INSERT and
+ // UPDATE assignment maps. Key=1's `email` is preserved at "[email protected]";
key=2 is inserted
+ // with `email = NULL`.
+ stream.addData((2, "bob", None, 2L))
+ runPipeline(buildCtx(includeEmail = false))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, cdcMeta(None, Some(1L)), "[email protected]"),
+ Row(2, "bob", 2L, cdcMeta(None, Some(2L)), null)
+ )
+ )
+ }
+
+ test("dropping a nested struct field between runs fails with
INCOMPATIBLE_DATA_FOR_TABLE") {
+ val session = spark
+ import session.implicits._
+
+ // The v2 writer's column-resolution layer requires every nested target
field to be
+ // present in the microbatch DF. When run #2's source projection drops
`b.c`, the merge
+ // fails with INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA. Users who want
to drop a
+ // nested field between runs must full-refresh the target.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(key INT NOT NULL, version BIGINT NOT NULL, " +
+ s"value STRUCT<a:INT,b:STRUCT<c:INT,d:INT>>, $cdcMetadataDdl)"
+ )
+
+ // Stream is (key, version, a, b_c, b_d). Each run reshapes into different
`value`
+ // shapes; the underlying tuple shape is unchanged so the streaming
source's identity
+ // is stable across runs.
+ val stream = MemoryStream[(Int, Long, Int, Int, Int)]
+ def buildCtx(includeC: Boolean): TestGraphRegistrationContext =
+ new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ val src = stream.toDF().toDF("key", "version", "a", "b_c", "b_d")
+ val inner = if (includeC) {
+ functions.struct(functions.col("b_c").as("c"),
functions.col("b_d").as("d"))
+ } else {
+ functions.struct(functions.col("b_d").as("d"))
+ }
+ val projected = src.select(
+ functions.col("key"),
+ functions.col("version"),
+ functions.struct(functions.col("a"), inner.as("b")).as("value")
+ )
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(projected),
+ keys = Seq("key"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ stream.addData((1, 1L, 1, 1, 1), (2, 1L, 2, 2, 2))
+ runPipeline(buildCtx(includeC = true))
+
+ // Run #2 drops b.c. The v2 writer rejects the merge because it cannot
find data for
+ // the target's `value.b.c` column.
+ stream.addData((1, 2L, 10, 99, 10), (3, 1L, 3, 99, 3))
+ val ex = intercept[RuntimeException] { runPipeline(buildCtx(includeC =
false)) }
+ val all = Iterator(ex) ++ ex.getSuppressed.iterator
+ assert(
+ all.exists(t => Option(t.getMessage).exists(m =>
+ m.contains("INCOMPATIBLE_DATA_FOR_TABLE") && m.contains("value") &&
m.contains("b") &&
+ m.contains("c"))),
+ s"Expected INCOMPATIBLE_DATA_FOR_TABLE failure for value.b.c, got:
${ex.getMessage}"
+ )
Review Comment:
nit: standardize on `checkErrorInPipelineFailure`.
This case (and the symmetric `array<struct>` drop case at L566-571, the
`AMBIGUOUS_REFERENCE` case at L609-614, and the
`CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE` case at L701-706) assert via
`message.contains(...)` while the rest of the suite uses
`checkErrorInPipelineFailure(condition = ..., parameters = ...)`. Substring
matching is brittle and will not catch error-class regressions. Worth aligning
these four with the rest of the suite for consistency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
