You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by jt...@apache.org on 2026/07/24 23:43:18 UTC
(spark) branch master updated: [SPARK-58320][SDP] Build SCD2 AutoCDC auxiliary table spec
This is an automated email from the ASF dual-hosted git repository.
jose-torres pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6bef01e1603b [SPARK-58320][SDP] Build SCD2 AutoCDC auxiliary table spec
6bef01e1603b is described below
commit 6bef01e1603b3d71d31300e9d07461f6993be5f7
Author: Andreas Neumann <an...@databricks.com>
AuthorDate: Fri Jul 24 16:43:05 2026 -0700
[SPARK-58320][SDP] Build SCD2 AutoCDC auxiliary table spec
### What changes were proposed in this pull request?
Implement `buildScd2AuxiliaryTableSpecFor` in `AutoCdcAuxiliaryTable`, which was previously an internal-error stub (`"SCD2 auxiliary table derivation is not yet implemented."`).
Unlike the SCD1 auxiliary table -- which stores only the key columns plus the CDC metadata column -- the SCD2 auxiliary table holds full hidden rows (tombstones and coalesced no-op upserts that may later be promoted into the visible target). Its schema is therefore the entire SCD2 target row schema plus the aux-only `__spark_autocdc_deleted_by_batch_id` logical-delete marker (a nullable `Long`). This matches what `Scd2BatchProcessor.findAffectedRowsFromAuxiliaryTable` reads and what [...]
`DatasetManager` materialization/evolution is schema-driven and SCD-type-agnostic, so it creates and evolves the SCD2 auxiliary table from this spec with no further change.
Also widens `Scd2BatchProcessor.deletedByBatchIdColName` to `private[pipelines]` so the graph package can name the marker column.
This is one sub-task of enabling SCD2 AutoCDC end to end (under [SPARK-56249](https://issues.apache.org/jira/browse/SPARK-56249)), following [SPARK-58319](https://issues.apache.org/jira/browse/SPARK-58319) (SCD2 target schema derivation). It is dormant until the switch-flip sub-task ([SPARK-58321](https://issues.apache.org/jira/browse/SPARK-58321)) enables SCD2 flows: the flow planner and construction still reject SCD2 today, so no SCD2 auxiliary table is materialized yet.
### Why are the changes needed?
Deriving the SCD2 auxiliary-table spec is a prerequisite for SCD2 support. The pipeline's dataset materialization builds and
schema-evolves the auxiliary table from this spec, and the SCD2 reconciliation engine (already landed in SPARK-57378) reads/writes exactly this shape, so the auxiliary table cannot be created correctly until the spec produces the right schema, properties, and drift metadata.
### Does this PR introduce _any_ user-facing change?
No. SCD2 AutoCDC flows are still not constructible/runnable (rejected by the flow planner), so no SCD2 auxiliary table is created and there is no change to released or user-facing behavior. This only makes the spec derivable internally.
### How was this patch tested?
New `AutoCdcScd2AuxiliaryTableSpecSuite` resolves a real SCD2 AutoCDC graph and asserts the derived auxiliary-table spec:
- the schema equals the full SCD2 target row schema plus the marker column;
- the marker is a nullable `Long` named `__spark_autocdc_deleted_by_batch_id`, appended last;
- the framework columns (`__START_AT`, `__END_AT`, `_cdc_metadata`) are retained from the target row schema;
- the spec records `ScdType.Type2`, the key column names (property + `expectedKeyFields`), and the correct identifiers;
- composite keys are all carried through.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 4.8
Closes #57514 from anew/spark-58320-build-scd2-aux-table-spec.
Authored-by: Andreas Neumann <an...@databricks.com>
Signed-off-by: Jose Torres <jt...@apache.org>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 2 +-
.../pipelines/graph/AutoCdcAuxiliaryTable.scala | 82 ++++++++++-
.../graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala | 158 +++++++++++++++++++++
3 files changed, 235 insertions(+), 7 deletions(-)
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
index cfcafe03f68d..157d2d5788b6 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
@@ -1362,7 +1362,7 @@ object Scd2BatchProcessor {
* idempotency, should a microbatch fail between a MERGE executed against the auxiliary
* table and the MERGE executed against the target table.
*/
- private[autocdc] val deletedByBatchIdColName: String =
+ private[pipelines] val deletedByBatchIdColName: String =
s"${AutoCdcReservedNames.prefix}deleted_by_batch_id"
/**
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
index 6fcf2fc4cdab..9f3dd2dba0d1 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
@@ -27,8 +27,8 @@ import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Table => CatalogTable, TableCatalog}
-import org.apache.spark.sql.pipelines.autocdc.{AutoCdcReservedNames, ScdType}
-import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.pipelines.autocdc.{AutoCdcReservedNames, Scd2BatchProcessor, ScdType}
+import org.apache.spark.sql.types.{LongType, StructField, StructType}
/**
* Helpers to construct and validate an AutoCDC flow's auxiliary table within the context of a
@@ -106,10 +106,10 @@ object AutoCdcAuxiliaryTable {
inputAutoCdcFlow
)
case ScdType.Type2 =>
- // SCD2 auxiliary derivation lands with SCD2 support. AutoCdcMergeFlow rejects SCD2 at
- // construction today, so a resolved SCD2 flow cannot exist and this branch is unreachable.
- throw SparkException.internalError(
- "SCD2 auxiliary table derivation is not yet implemented."
+ buildScd2AuxiliaryTableSpecFor(
+ targetTable,
+ targetTableSchema,
+ inputAutoCdcFlow
)
}
}
@@ -178,6 +178,76 @@ object AutoCdcAuxiliaryTable {
)
}
+ /**
+ * Build the SCD2 auxiliary table spec given the AutoCdc flow's declared keys and the target
+ * table it writes to.
+ *
+ * Unlike the SCD1 auxiliary table (which stores only keys + CDC metadata), the SCD2 auxiliary
+ * table stores full hidden rows -- tombstones and coalesced no-op upserts that may later be
+ * promoted into the visible target -- so its schema is the entire SCD2 target row schema plus
+ * the aux-only [[Scd2BatchProcessor.deletedByBatchIdColName]] logical-delete marker. This
+ * matches what [[Scd2BatchProcessor.findAffectedRowsFromAuxiliaryTable]] reads and the merges
+ * write: "[[Scd2BatchProcessor.deletedByBatchIdColName]] in addition to all of the columns in
+ * the target table".
+ *
+ * @param targetTable the dataset that owns the SCD2 auxiliary table
+ * @param targetTableSchema the AutoCDC target's evolved schema as of the latest pipeline run
+ * (the union of all flows writing to the target after schema
+ * evolution), which already carries the SCD2 framework columns
+ * (`__START_AT`, `__END_AT`, and the CDC metadata column)
+ * @param inputAutoCdcFlow the AutoCDC flow writing to `targetTable`
+ * @return the SCD2 auxiliary-table spec
+ */
+ private def buildScd2AuxiliaryTableSpecFor(
+ targetTable: Table,
+ targetTableSchema: StructType,
+ inputAutoCdcFlow: AutoCdcMergeFlow
+ ): AuxiliaryTableSpec = {
+ val scd2AuxiliaryTableIdentifier = identifier(targetTable.identifier)
+
+ val resolver = inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver
+ val autoCdcKeyColumnNames = inputAutoCdcFlow.changeArgs.keys.map(_.name)
+
+ // Resolve the key fields from the (evolved) target schema, exactly as SCD1 does, so the
+ // recorded key names/types used for drift detection come from the same source of truth.
+ val keyFields = autoCdcKeyColumnNames.map { keyColumnName =>
+ findFieldInTargetSchema(
+ targetTableSchema = targetTableSchema,
+ targetTableIdentifier = targetTable.identifier,
+ autoCdcFlowIdentifier = inputAutoCdcFlow.identifier,
+ fieldName = keyColumnName,
+ resolver = resolver
+ )
+ }
+
+ // The SCD2 auxiliary table holds full rows in the SCD2 target row schema, plus the aux-only
+ // logical-delete marker column appended last. The marker holds the batchId whose MERGE
+ // logically deleted a row (null on live rows), so it is a nullable Long.
+ val deletedByBatchIdField =
+ StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
+ val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+ deletedByBatchIdField)
+
+ val scd2AuxiliaryTableProperties =
+ // Record which SCD strategy this auxiliary table serves so downstream readers can identify it
+ // without inspecting the schema.
+ Map(scdTypePropertyKey -> ScdType.Type2.label) ++
+ // Persist the AutoCDC key column names as a JSON list; immutable post-creation (full-refresh
+ // is the only way to change it).
+ Map(keyColumnNamesProperty -> serializeKeyColumnNames(keyFields.map(_.name))) ++
+ // Inherit the target's format so MERGE semantics line up. When unspecified, omit the provider
+ // so the catalog falls back to its default.
+ targetTable.format.map(TableCatalog.PROP_PROVIDER -> _)
+
+ AutoCdcAuxiliaryTableSpec(
+ identifier = scd2AuxiliaryTableIdentifier,
+ schema = scd2AuxiliaryTableSchema,
+ properties = scd2AuxiliaryTableProperties,
+ targetTableIdentifier = targetTable.identifier,
+ expectedKeyFields = keyFields,
+ expectedScdType = ScdType.Type2
+ )
+ }
+
/**
* Resolve the [[StructField]] named `fieldName` in `targetTableSchema` (the AutoCDC target's
* evolved schema). The key columns and the CDC metadata column are always present in that schema,
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
new file mode 100644
index 000000000000..d86f3ba89bdd
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
@@ -0,0 +1,158 @@
+/*
+ * 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.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ChangeArgs,
+ ColumnSelection,
+ Scd2BatchProcessor,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{PipelineTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{LongType, StructField, StructType}
+
+/**
+ * Tests for [[AutoCdcAuxiliaryTable.buildAuxiliaryTableSpecFor]] on the SCD2 branch
+ * (`buildScd2AuxiliaryTableSpecFor`), exercised through the graph the same way production does:
+ * a resolved SCD2 AutoCDC flow yields an [[AutoCdcAuxiliaryTableSpec]] via
+ * [[DataflowGraph.auxiliaryTableSpecs]].
+ *
+ * The SCD2 auxiliary table stores full hidden rows (tombstones and coalesced no-op upserts), so
+ * its schema is the entire SCD2 target row schema plus the aux-only
+ * [[Scd2BatchProcessor.deletedByBatchIdColName]] logical-delete marker -- richer than the SCD1
+ * auxiliary table, which stores only keys + CDC metadata.
+ */
+class AutoCdcScd2AuxiliaryTableSpecSuite extends PipelineTest with SharedSparkSession {
+
+ private def targetIdentifier = fullyQualifiedIdentifier("target")
+
+ /** Source change feed with data columns `(id, name, version)`. */
+ private def sourceDf = {
+ val session = spark
+ import session.implicits._
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+ stream.toDF().toDF("id", "name", "version")
+ }
+
+ /** Resolve a graph with a single SCD2 AUTO CDC flow into `target` and return its aux spec. */
+ private def scd2AuxSpec(
+ keys: Seq[String] = Seq("id"),
+ columnSelection: Option[ColumnSelection] = None): AutoCdcAuxiliaryTableSpec = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ ctx.registerTable("target")
+ ctx.registerFlow(AutoCdcFlow(
+ identifier = targetIdentifier,
+ destinationIdentifier = targetIdentifier,
+ func = ctx.dfFlowFunc(sourceDf),
+ queryContext = QueryContext(
+ currentCatalog = catalogInPipelineSpec,
+ currentDatabase = databaseInPipelineSpec),
+ origin = QueryOrigin.empty,
+ changeArgs = ChangeArgs(
+ keys = keys.map(k => UnqualifiedColumnName(Seq(k))),
+ sequencing = functions.col("version"),
+ columnSelection = columnSelection,
+ deleteCondition = None,
+ storedAsScdType = ScdType.Type2)))
+ val graph = ctx.resolveToDataflowGraph()
+ graph.auxiliaryTableSpecs(targetIdentifier).asInstanceOf[AutoCdcAuxiliaryTableSpec]
+ }
+
+ /** The SCD2 target (inferred) schema for the default single-flow graph. */
+ private def scd2TargetSchema(): StructType = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ ctx.registerTable("target")
+ ctx.registerFlow(AutoCdcFlow(
+ identifier = targetIdentifier,
+ destinationIdentifier = targetIdentifier,
+ func = ctx.dfFlowFunc(sourceDf),
+ queryContext = QueryContext(
+ currentCatalog = catalogInPipelineSpec,
+ currentDatabase = databaseInPipelineSpec),
+ origin = QueryOrigin.empty,
+ changeArgs = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName(Seq("id"))),
+ sequencing = functions.col("version"),
+ columnSelection = None,
+ deleteCondition = None,
+ storedAsScdType = ScdType.Type2)))
+ ctx.resolveToDataflowGraph().inferredSchema(targetIdentifier)
+ }
+
+ test("SCD2 aux schema is the full target schema plus the deleted-by-batch-id marker") {
+ val targetSchema = scd2TargetSchema()
+ val spec = scd2AuxSpec()
+
+ // The target schema already carries the SCD2 framework columns (__START_AT / __END_AT /
+ // _cdc_metadata); the aux schema is exactly that, with the logical-delete marker appended.
+ val expected = StructType(
+ targetSchema.fields :+
+ StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
+ )
+ assert(spec.schema == expected)
+ }
+
+ test("SCD2 aux schema ends with a nullable long deleted-by-batch-id column") {
+ val spec = scd2AuxSpec()
+ val marker = spec.schema.fields.last
+ assert(marker.name == Scd2BatchProcessor.deletedByBatchIdColName)
+ assert(marker.dataType == LongType)
+ assert(marker.nullable, "the logical-delete marker must be nullable (null on live rows)")
+ }
+
+ test("SCD2 aux schema retains the framework columns from the target row schema") {
+ val fieldNames = scd2AuxSpec().schema.fieldNames.toSeq
+ Seq(
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.deletedByBatchIdColName
+ ).foreach { c =>
+ assert(fieldNames.contains(c), s"expected $c in aux schema $fieldNames")
+ }
+ }
+
+ test("SCD2 aux spec records the SCD2 type, key names, and matching identifier") {
+ val spec = scd2AuxSpec()
+ assert(spec.expectedScdType == ScdType.Type2)
+ assert(spec.properties(AutoCdcAuxiliaryTable.scdTypePropertyKey) == ScdType.Type2.label)
+ assert(spec.expectedKeyFields.map(_.name) == Seq("id"))
+ assert(
+ AutoCdcAuxiliaryTable.parseKeyColumnNames(
+ spec.properties(AutoCdcAuxiliaryTable.keyColumnNamesProperty)).contains(Seq("id")))
+ assert(spec.identifier == AutoCdcAuxiliaryTable.identifier(targetIdentifier))
+ assert(spec.targetTableIdentifier == targetIdentifier)
+ }
+
+ test("SCD2 aux spec carries all declared key fields for a composite key") {
+ val spec = scd2AuxSpec(keys = Seq("id", "name"))
+ assert(spec.expectedKeyFields.map(_.name) == Seq("id", "name"))
+ assert(
+ AutoCdcAuxiliaryTable.parseKeyColumnNames(
+ spec.properties(AutoCdcAuxiliaryTable.keyColumnNamesProperty)).contains(Seq("id", "name")))
+ // Both keys survive into the aux schema.
+ assert(spec.schema.fieldNames.contains("id"))
+ assert(spec.schema.fieldNames.contains("name"))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org