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 938d5cd72f44 [SPARK-58319][SDP] Derive SCD2 AutoCDC target table schema
938d5cd72f44 is described below
commit 938d5cd72f44b701ff8872bccd7c689b22bc6974
Author: Andreas Neumann <[email protected]>
AuthorDate: Fri Jul 24 13:25:26 2026 -0700
[SPARK-58319][SDP] Derive SCD2 AutoCDC target table schema
Title: [SPARK-58319][SDP] Derive SCD2 AutoCDC target table schema
Body:
### What changes were proposed in this pull request?
Implement the `ScdType.Type2` branch of `AutoCdcMergeFlow.schema` and
`load` in `Flow.scala`, which previously threw`AUTOCDC_SCD2_NOT_SUPPORTED`. The
SCD2 target schema is the user-selected columns followed by the SCD2 framework
columns, in the exact order and nullability that
`Scd2BatchProcessor.preprocessMicrobatch` projects and the merges persist:
- `__START_AT` (typed as the sequencing type, nullable)
- `__END_AT` (typed as the sequencing type, nullable)
- `_cdc_metadata` (the SCD2 operational metadata struct, non-null)
To build this from the `graph` package, the visibility of a few
`Scd2BatchProcessor` members is widened: `startAtColName` / `endAtColName`
(`private[autocdc]` -> `private[pipelines]`) and `constructCdcMetadataCol`
(`private` -> `private[pipelines]`).`cdcMetadataColSchema` was already
`private[pipelines]`.
This is one subtask of enabling SCD2 AutoCDC end to end (under
[SPARK-56249](https://issues.apache.org/jira/browse/SPARK-56249)). After this
change, SCD2 AutoCDC flows are **constructible** (their output schema
resolves), but they still **cannot run**: the auxiliary-table derivation
(`AutoCdcAuxiliaryTable`) and the flow planner (`FlowPlanner`) still reject
SCD2. Those remaining gates and the streaming-write wiring are handled in
follow-up subtasks (SPARK-58320, SPARK-58321).
### Why are the changes needed?
Deriving the SCD2 target schema is a prerequisite for SCD2 support: the
pipeline's downstream dataset materialization and schema evolution consume
`AutoCdcMergeFlow.schema`, so an SCD2 flow cannot be registered or its target
table created until this branch produces the correct augmented schema. The
`Scd2BatchProcessor` reconciliation engine that produces rows in this exact
shape already landed (SPARK-57378); this connects the flow's declared schema
to it.
### Does this PR introduce _any_ user-facing change?
No. SCD2 AutoCDC flows remain unusable end to end (still gated by the
auxiliary-table and planner checks), so there is no change to released or
user-facing behavior. This only makes the flow's schema resolvable internally.
### How was this patch tested?
New unit tests in `AutoCdcFlowSuite` (replacing the obsolete "rejects
SCD2 at construction" test):
- `schema` appends the SCD2 framework columns in the correct
order/nullability when no column selection is set;
- a `columnSelection` narrows only the user-data portion, before the
framework columns;
- the framework columns and the metadata record-start-at field pick up a
non-Long resolved sequencing type;
- `load()`'s empty-dataframe schema matches `schema` for SCD2.
### Was this patch authored or co-authored using generative AI tooling?
Generated-By: Claude Opus 4.8
Closes #57496 from anew/spark-58319-derive-scd2-target-schema.
Authored-by: Andreas Neumann <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 6 +-
.../apache/spark/sql/pipelines/graph/Flow.scala | 36 +++-
.../sql/pipelines/autocdc/AutoCdcFlowSuite.scala | 230 +++++++++++++++++++--
3 files changed, 248 insertions(+), 24 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 34bd8a5d2929..cfcafe03f68d 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
@@ -1379,7 +1379,7 @@ object Scd2BatchProcessor {
* The invariant in both tables is: startAtColName <=
recordStartAtFieldName. If an event was
* generated at time X, it is active by time X, or earlier if it is not a
run head.
*/
- private[autocdc] val startAtColName: String = "__START_AT"
+ private[pipelines] val startAtColName: String = "__START_AT"
/**
* What this column represents depends on which AutoCDC artifact table it is
read from.
@@ -1393,7 +1393,7 @@ object Scd2BatchProcessor {
* Else this row is a coalesced no-op row that is part of an upsert run,
and by
* convention the value will always be null.
*/
- private[autocdc] val endAtColName: String = "__END_AT"
+ private[pipelines] val endAtColName: String = "__END_AT"
/**
* Column names reserved by AutoCDC that will be projected onto the
microbatch and
@@ -1519,7 +1519,7 @@ object Scd2BatchProcessor {
* Construct the CDC metadata struct column for SCD2 rows, following the
exact schema and
* field ordering defined by [[cdcMetadataColSchema]].
*/
- private def constructCdcMetadataCol(
+ private[pipelines] def constructCdcMetadataCol(
recordStartAt: Column,
sequencingType: DataType
): Column = {
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
index dd4d1556afbf..31444f0c15f3 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
@@ -30,6 +30,7 @@ import org.apache.spark.sql.pipelines.autocdc.{
ChangeArgs,
ColumnSelection,
Scd1BatchProcessor,
+ Scd2BatchProcessor,
ScdType
}
import org.apache.spark.sql.types.{DataType, StructField, StructType}
@@ -294,9 +295,21 @@ class AutoCdcMergeFlow(
)
)
case ScdType.Type2 =>
- throw new AnalysisException(
- errorClass = "AUTOCDC_SCD2_NOT_SUPPORTED",
- messageParameters = Map.empty
+ // SCD2 produces a target table with all the user-selected output
columns followed by the
+ // SCD2 framework columns (in this exact order and nullability, matching
what
+ // Scd2BatchProcessor.preprocessMicrobatch projects and the merges
persist): the visible
+ // interval bounds __START_AT / __END_AT typed as the sequencing type,
then the operational
+ // CDC metadata column.
+ StructType(
+ userSelectedSchema.fields ++ Seq(
+ StructField(Scd2BatchProcessor.startAtColName, sequencingType,
nullable = true),
+ StructField(Scd2BatchProcessor.endAtColName, sequencingType,
nullable = true),
+ StructField(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(sequencingType),
+ nullable = false
+ )
+ )
)
}
@@ -338,10 +351,19 @@ class AutoCdcMergeFlow(
df.select(userSelectedCols :+ emptyCdcMetadataCol: _*)
case ScdType.Type2 =>
- throw new AnalysisException(
- errorClass = "AUTOCDC_SCD2_NOT_SUPPORTED",
- messageParameters = Map.empty
- )
+ val userSelectedCols: Seq[Column] =
userSelectedSchema.fieldNames.toSeq.map(F.col)
+ // Mirror the SCD2 augmented schema: null interval bounds (cast to the
sequencing type)
+ // and an empty CDC metadata struct, matching
AutoCdcMergeFlow.schema's Type2 branch.
+ val emptyStartAtCol: Column =
+
F.lit(null).cast(sequencingType).as(Scd2BatchProcessor.startAtColName)
+ val emptyEndAtCol: Column =
+ F.lit(null).cast(sequencingType).as(Scd2BatchProcessor.endAtColName)
+ val emptyCdcMetadataCol: Column =
Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = F.lit(null),
+ sequencingType = sequencingType
+ ).as(AutoCdcReservedNames.cdcMetadataColName)
+
+ df.select(userSelectedCols ++ Seq(emptyStartAtCol, emptyEndAtCol,
emptyCdcMetadataCol): _*)
}
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
index 32374f8ecb04..98c31a3020ef 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
@@ -288,21 +288,201 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
assert(first eq second, "schema should be cached as a val and return the
same instance")
}
- test("AutoCdcMergeFlow rejects SCD2 at construction with
AUTOCDC_SCD2_NOT_SUPPORTED") {
- // Constructing the flow forces the resolved schema, which is unsupported
for SCD2 today.
- // Failing eagerly (rather than deferring to the first downstream `schema`
read) is the
- // intended UX -- pipeline graph analysis should not be able to register
an SCD2 AutoCDC
- // flow at all.
- checkError(
- exception = intercept[AnalysisException] {
- newAutoCdcMergeFlow(
- sourceDf = threeColumnSourceDf(),
- storedAsScdType = ScdType.Type2
+ test(
+ "AutoCdcMergeFlow.schema appends the SCD2 framework columns when no
columnSelection is set"
+ ) {
+ // SCD2 appends __START_AT / __END_AT (typed as the sequencing type,
nullable) followed by the
+ // SCD2 _cdc_metadata struct (non-null), in this exact order -- matching
the canonical target
+ // schema that Scd2BatchProcessor.preprocessMicrobatch projects and the
merges persist.
+ val resolvedFlow = newAutoCdcMergeFlow(
+ sourceDf = threeColumnSourceDf(),
+ storedAsScdType = ScdType.Type2
+ )
+
+ val expected = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("name", StringType)
+ .add("seq", LongType)
+ .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true)
+ .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true)
+ .add(
+ StructField(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType),
+ nullable = false
)
- },
- condition = "AUTOCDC_SCD2_NOT_SUPPORTED",
- sqlState = "0A000",
- parameters = Map.empty
+ )
+ assert(resolvedFlow.schema == expected)
+ }
+
+ test("AutoCdcMergeFlow.schema applies an IncludeColumns selection before the
SCD2 " +
+ "framework columns") {
+ // An IncludeColumns selection narrows only the user-data portion; the
framework columns are
+ // always appended and cannot be selected away.
+ val resolvedFlow = newAutoCdcMergeFlow(
+ sourceDf = threeColumnSourceDf(),
+ storedAsScdType = ScdType.Type2,
+ columnSelection = Some(
+ ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("id"), UnqualifiedColumnName("seq"))
+ )
+ )
+ )
+
+ assert(
+ resolvedFlow.schema.fieldNames.toSeq ==
+ Seq(
+ "id",
+ "seq",
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName
+ )
+ )
+ }
+
+ test("AutoCdcMergeFlow.schema appends the SCD2 framework columns even when a
selection " +
+ "excludes a user column") {
+ // The framework columns are appended after the user selection is applied
to the source
+ // schema, so no ExcludeColumns selection over the user data can strip
them: they always
+ // appear in the output.
+ val resolvedFlow = newAutoCdcMergeFlow(
+ sourceDf = threeColumnSourceDf(),
+ storedAsScdType = ScdType.Type2,
+ columnSelection = Some(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("name")))
+ )
+ )
+
+ // The excluded user column is gone, but every framework column is still
present.
+ assert(!resolvedFlow.schema.fieldNames.contains("name"))
+ Seq(
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName
+ ).foreach { frameworkCol =>
+ assert(
+ resolvedFlow.schema.fieldNames.contains(frameworkCol),
+ s"expected framework column $frameworkCol in
${resolvedFlow.schema.fieldNames.toSeq}"
+ )
+ }
+ }
+
+ test("AutoCdcMergeFlow.schema lets a selection exclude a source column that
collides with " +
+ "a non-prefixed framework column, re-adding it from the framework") {
+ // Two of the three SCD2 framework columns -- __START_AT and __END_AT --
do not carry the
+ // reserved AutoCDC prefix, so a source change feed may legitimately
contain columns with
+ // those names. The user excludes them via the column selection; they are
dropped from the
+ // user-data portion and the engine's own framework columns are appended
in their place, so
+ // each still appears exactly once in the output with the framework's type
(the sequencing
+ // type), not the source column's type.
+ //
+ // The third framework column, _cdc_metadata, DOES carry the reserved
prefix, so a source
+ // that contains it is rejected outright at flow construction -- it can
never reach column
+ // selection and so is deliberately out of scope here. That rejection is
covered separately by
+ // "AutoCdcMergeFlow rejects a source df column whose name equals the
reserved CDC metadata
+ // column".
+ val session = spark
+ import session.implicits._
+ // Source carries String-typed __START_AT / __END_AT columns alongside the
data columns.
+ val sourceDf = MemoryStream[(Int, String, Option[Long], String, String)]
+ .toDS()
+ .toDF(
+ "id",
+ "name",
+ "seq",
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName)
+
+ val resolvedFlow = newAutoCdcMergeFlow(
+ sourceDf = sourceDf,
+ storedAsScdType = ScdType.Type2,
+ columnSelection = Some(
+ ColumnSelection.ExcludeColumns(
+ Seq(
+ UnqualifiedColumnName(Scd2BatchProcessor.startAtColName),
+ UnqualifiedColumnName(Scd2BatchProcessor.endAtColName))
+ )
+ )
+ )
+
+ // Each framework column appears exactly once (the source copy was
excluded, the framework
+ // copy appended) ...
+ assert(
+ resolvedFlow.schema.fieldNames.count(_ ==
Scd2BatchProcessor.startAtColName) == 1)
+ assert(
+ resolvedFlow.schema.fieldNames.count(_ ==
Scd2BatchProcessor.endAtColName) == 1)
+ // ... and carries the framework (sequencing) type, not the source String
type.
+ assert(resolvedFlow.schema(Scd2BatchProcessor.startAtColName).dataType ==
LongType)
+ assert(resolvedFlow.schema(Scd2BatchProcessor.endAtColName).dataType ==
LongType)
+ // Full expected shape: the retained data columns followed by the
framework columns.
+ assert(
+ resolvedFlow.schema.fieldNames.toSeq ==
+ Seq(
+ "id",
+ "name",
+ "seq",
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName
+ )
+ )
+ }
+
+ test("AutoCdcMergeFlow.schema's SCD2 framework columns use the resolved
sequencing type") {
+ // A non-Long sequencing expression must flow through to __START_AT /
__END_AT and the
+ // metadata struct's record-start-at field.
+ val session = spark
+ import session.implicits._
+ val sourceDf = MemoryStream[(Int, String, Int)].toDS().toDF("id", "name",
"seq")
+
+ val resolvedFlow = newAutoCdcMergeFlow(sourceDf, storedAsScdType =
ScdType.Type2)
+
+ assert(resolvedFlow.schema(Scd2BatchProcessor.startAtColName).dataType ==
IntegerType)
+ assert(resolvedFlow.schema(Scd2BatchProcessor.endAtColName).dataType ==
IntegerType)
+ assert(
+ resolvedFlow.schema(AutoCdcReservedNames.cdcMetadataColName).dataType ==
+ Scd2BatchProcessor.cdcMetadataColSchema(IntegerType)
+ )
+ }
+
+ test("AutoCdcMergeFlow.schema supports a multi-column (struct) sequencing
expression") {
+ // A multi-column sequence is expressed as a struct over the ordering
columns; its resolved
+ // type is the corresponding StructType, which must flow into __START_AT /
__END_AT and the
+ // metadata struct's record-start-at field unchanged, including each
field's own nullability.
+ val session = spark
+ import session.implicits._
+ // seq1 is a non-null Int; seq2 is a nullable Long (Option[Long]), so the
struct carries
+ // mixed per-field nullability.
+ val sourceDf =
+ MemoryStream[(Int, String, Int, Option[Long])].toDS().toDF("id", "name",
"seq1", "seq2")
+
+ val resolvedFlow = newAutoCdcMergeFlow(
+ sourceDf = sourceDf,
+ sequencing = F.struct(F.col("seq1"), F.col("seq2")),
+ storedAsScdType = ScdType.Type2
+ )
+
+ // Top-level nullability and inner-field nullability are independent. The
__START_AT /
+ // __END_AT columns must themselves stay nullable -- an open/unset
interval bound is
+ // represented as NULL, which reconciliation relies on -- regardless of
the sequencing
+ // struct's field nullability. A struct column is free to be NULL at the
top level
+ // regardless of its fields' nullability.
+ val startAtField = resolvedFlow.schema(Scd2BatchProcessor.startAtColName)
+ val endAtField = resolvedFlow.schema(Scd2BatchProcessor.endAtColName)
+ assert(startAtField.nullable, "__START_AT must be nullable")
+ assert(endAtField.nullable, "__END_AT must be nullable")
+
+ // The struct data type flows through unchanged, preserving each field's
own nullability:
+ // seq1 stays non-null, seq2 stays nullable.
+ val expectedSeqType = new StructType()
+ .add("seq1", IntegerType, nullable = false)
+ .add("seq2", LongType, nullable = true)
+ assert(startAtField.dataType == expectedSeqType)
+ assert(endAtField.dataType == expectedSeqType)
+ assert(
+ resolvedFlow.schema(AutoCdcReservedNames.cdcMetadataColName).dataType ==
+ Scd2BatchProcessor.cdcMetadataColSchema(expectedSeqType)
)
}
@@ -349,6 +529,28 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
)
}
+ test("AutoCdcMergeFlow.load() schema matches AutoCdcMergeFlow.schema for
SCD2") {
+ // The empty-df load path must reproduce the SCD2 augmented schema
exactly, including the
+ // trailing framework columns.
+ val resolvedFlow = newAutoCdcMergeFlow(
+ sourceDf = threeColumnSourceDf(),
+ storedAsScdType = ScdType.Type2
+ )
+ val loadedDf = resolvedFlow.load(asStreaming = true)
+ assert(loadedDf.schema == resolvedFlow.schema)
+ assert(
+ loadedDf.schema.fieldNames.toSeq ==
+ Seq(
+ "id",
+ "name",
+ "seq",
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName
+ )
+ )
+ }
+
//
===========================================================================================
// AutoCdcMergeFlow reserved-prefix validation tests
//
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
