[SPARK-58319][SDP] Derive SCD2 AutoCDC target table schema by anew · Pull Request #57496 · apache/spark

anew · 2026-07-24T06:03:08Z

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 threwAUTOCDC_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). 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

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 and __END_AT (typed as the sequencing type, nullable), then the operational _cdc_metadata struct (non-null). To build this from the graph package, widen the visibility of Scd2BatchProcessor.startAtColName / endAtColName (private[autocdc] -> private[pipelines]) and constructCdcMetadataCol (private -> private[pipelines]). SCD2 AutoCDC flows are now constructible (their schema resolves) but still cannot run: the auxiliary-table derivation (AutoCdcAuxiliaryTable) and the flow planner (FlowPlanner) still reject SCD2. Those gates, and end-to-end wiring, are removed in follow-up subtasks under SPARK-56249. Adds tests in AutoCdcFlowSuite and replaces the obsolete "rejects SCD2 at construction" test. Co-authored-by: Isaac

anew · 2026-07-24T14:52:03Z

+      sourceDf = threeColumnSourceDf(),
+      storedAsScdType = ScdType.Type2,
+      columnSelection = Some(
+        ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("name")))


this test is meaningless as it stands, because the column selection does not attempt to exclude metadata columns. Can we:

change this test to only select "id" and "seq" via include columns

add a test that excludes the metadata columns via exclude columns and asserts that they still appear in the output schema

anew · 2026-07-24T14:52:44Z

+
+  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.


can you also add a test that uses a multi-column sequence?

good point - done

…ction tests Address review feedback on AutoCdcFlowSuite: - Change the column-selection schema test to use an IncludeColumns selection of the data columns (id, seq), so it meaningfully exercises narrowing. - Add a test asserting that an ExcludeColumns selection over a user column never strips the appended SCD2 framework columns (they always appear in the output). - Add a test that a multi-column (struct) sequencing expression flows through to __START_AT / __END_AT and the metadata struct's record-start-at field. Co-authored-by: Isaac

anew · 2026-07-24T15:28:17Z


+  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


What if a source table actually contains some framework columns, then the user must exclude them from the column selection. That should be possible. And we should test that they are dropped from the source but added to the target schema nonetheless.

added a test

… collide with framework columns A source change feed may legitimately contain columns named __START_AT / __END_AT, since those framework columns do not carry the reserved AutoCDC prefix. Add a test that such colliding source columns can be excluded via the column selection: they are dropped from the user-data portion and the engine's framework columns are appended in their place, so each appears exactly once in the output schema with the framework (sequencing) type rather than the source column's type. Also drop the now-inaccurate parenthetical in the adjacent exclude-user-column test that claimed framework column names cannot be named in a selection. Co-authored-by: Opus 4.8

… collide with framework columns A source change feed may legitimately contain columns named __START_AT / __END_AT, since those two framework columns do not carry the reserved AutoCDC prefix. Add a test that such colliding source columns can be excluded via the column selection: they are dropped from the user-data portion and the engine's framework columns are appended in their place, so each appears exactly once in the output schema with the framework (sequencing) type rather than the source column's type. The third framework column, _cdc_metadata, carries the reserved prefix and so a source containing it is rejected at flow construction (covered by an existing test); the new test's comment documents why it is deliberately out of scope here. Also drop the now-inaccurate parenthetical in the adjacent exclude-user-column test that claimed framework column names cannot be named in a selection. Co-authored-by: Opus 4.8

jose-torres

Small but important if accurate comment on nullability.

jose-torres · 2026-07-24T16:38:20Z

+      storedAsScdType = ScdType.Type2
+    )
+
+    // The struct's fields are non-nullable because they wrap concrete non-null source columns.


Isn't this wrong, though? NULL always has to be valid for the algorithm to work. Or is the top level of a struct with non-nullable fields still allowed to be null?

The nullable = false for the sub-fields here derives from the nullability of these fields in the memory stream. The top-level struct can still be nullable, and it has to be for the algorithm to work.

…p-level nullable The multi-column (struct) sequencing test asserted only the struct data type, so it was silent on the top-level nullability of __START_AT / __END_AT -- the property that actually matters, since an open/unset interval bound is represented as NULL. Add explicit assertions that both columns are top-level nullable, and clarify in the comment that top-level nullability is independent of the struct's inner-field nullability (a struct column may be NULL regardless of its fields being non-nullable). No production change: the schema already marks these columns nullable; this only makes the invariant explicit in the test. Co-authored-by: Opus 4.8

jose-torres

Will merge when CI (presumably) passes

… sequencing test Make seq2 a nullable Long (Option[Long]) in the multi-column (struct) sequencing test so the struct carries mixed per-field nullability, and assert that each field's own nullability flows through to __START_AT / __END_AT unchanged (seq1 non-null, seq2 nullable). The top-level framework columns remain nullable regardless, which the test continues to assert. Co-authored-by: Opus 4.8

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 <andreas.neumann@databricks.com> Signed-off-by: Jose Torres <jtorres@apache.org> (cherry picked from commit 938d5cd) Signed-off-by: Jose Torres <jtorres@apache.org>

jose-torres · 2026-07-24T20:26:14Z

merged to master and 4.x

anew commented Jul 24, 2026

View reviewed changes

anew added 2 commits July 24, 2026 15:31

jose-torres reviewed Jul 24, 2026

View reviewed changes

jose-torres approved these changes Jul 24, 2026

View reviewed changes

jose-torres closed this in 938d5cd Jul 24, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-58319][SDP] Derive SCD2 AutoCDC target table schema#57496

[SPARK-58319][SDP] Derive SCD2 AutoCDC target table schema#57496
anew wants to merge 6 commits into
apache:masterfrom
anew:spark-58319-derive-scd2-target-schema

anew commented Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

jose-torres left a comment

Uh oh!

jose-torres Jul 24, 2026

Uh oh!

anew Jul 24, 2026

Uh oh!

jose-torres left a comment

Uh oh!

jose-torres commented Jul 24, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

anew commented Jul 24, 2026

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jose-torres left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jose-torres left a comment

Choose a reason for hiding this comment

Uh oh!

jose-torres commented Jul 24, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants