[SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs by anew · Pull Request #57412 · apache/spark

anew · 2026-07-22T00:25:55Z

What changes were proposed in this pull request?

Expose Auto CDC SCD Type 2 through the Python and Spark Connect APIs. create_auto_cdc_flow now accepts stored_as_scd_type=2 (or "2") and the SCD2-only track_history_column_list / track_history_except_column_list parameters, which flow through to ChangeArgs(storedAsScdType = Type2, trackHistorySelection = ...) on the server.

Specifically:

Proto: add SCD_TYPE_2 to the SCDType enum and track_history_column_list (field 11) / track_history_except_column_list (field 12) to AutoCdcFlowDetails. Field numbers match the Databricks runtime extension proto for wire compatibility. Regenerate the Python stubs.
Python API (api.py, flow.py): widen stored_as_scd_type, add and normalize the track-history parameters, and reject track-history columns unless SCD2 is selected.
Serialization (spark_connect_graph_element_registry.py): map the SCD type and serialize the track-history columns.
Server (PipelinesHandler.scala): map SCD_TYPE_2 to ScdType.Type2, build trackHistorySelection, and reject specifying both track-history lists.
Add error condition AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST.

The engine's SCD2 execution path is landing in a parallel workstream and still rejects SCD2 during graph analysis (AUTOCDC_SCD2_NOT_SUPPORTED). This change wires the API to the ChangeArgs contract those changes plug into; end-to-end SCD2 execution starts working once the engine drops those throws, with no further API change required.

Why are the changes needed?

The Python and Spark Connect Auto CDC APIs previously only supported SCD Type 1. SCD Type 2 (history tracking) is a common CDC requirement and is already modeled in the engine's ChangeArgs/ScdType.

Does this PR introduce any user-facing change?

Yes. create_auto_cdc_flow accepts stored_as_scd_type=2 and the track_history_column_list / track_history_except_column_list parameters.

How was this patch tested?

New unit tests in test_auto_cdc_flow.py (SCD2 accept as int/str, track-history include/exclude, track-history-requires-SCD2 rejection) and PythonPipelineSuite.scala (SCD2 registration forwards to ChangeArgs, both-lists rejection). Ran the Python and Scala suites plus ruff/mypy.

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

Generated-by: Opus 4.8

… and Spark Connect APIs ### What changes were proposed in this pull request? Expose Auto CDC SCD Type 2 through the Python and Spark Connect APIs. `create_auto_cdc_flow` now accepts `stored_as_scd_type=2` (or `"2"`) and the SCD2-only `track_history_column_list` / `track_history_except_column_list` parameters, which flow through to `ChangeArgs(storedAsScdType = Type2, trackHistorySelection = ...)` on the server. Specifically: - Proto: add `SCD_TYPE_2` to the `SCDType` enum and `track_history_column_list` (field 11) / `track_history_except_column_list` (field 12) to `AutoCdcFlowDetails`. Field numbers match the Databricks runtime extension proto for wire compatibility. Regenerate the Python stubs. - Python API (`api.py`, `flow.py`): widen `stored_as_scd_type`, add and normalize the track-history parameters, and reject track-history columns unless SCD2 is selected. - Serialization (`spark_connect_graph_element_registry.py`): map the SCD type and serialize the track-history columns. - Server (`PipelinesHandler.scala`): map `SCD_TYPE_2` to `ScdType.Type2`, build `trackHistorySelection`, and reject specifying both track-history lists. - Add error condition `AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST`. The engine's SCD2 execution path is landing in a parallel workstream and still rejects SCD2 during graph analysis (`AUTOCDC_SCD2_NOT_SUPPORTED`). This change wires the API to the `ChangeArgs` contract those changes plug into; end-to-end SCD2 execution starts working once the engine drops those throws, with no further API change required. ### Why are the changes needed? The Python and Spark Connect Auto CDC APIs previously only supported SCD Type 1. SCD Type 2 (history tracking) is a common CDC requirement and is already modeled in the engine's `ChangeArgs`/`ScdType`. ### Does this PR introduce any user-facing change? Yes. `create_auto_cdc_flow` accepts `stored_as_scd_type=2` and the `track_history_column_list` / `track_history_except_column_list` parameters. ### How was this patch tested? New unit tests in `test_auto_cdc_flow.py` (SCD2 accept as int/str, track-history include/exclude, track-history-requires-SCD2 rejection) and `PythonPipelineSuite.scala` (SCD2 registration forwards to `ChangeArgs`, both-lists rejection). Ran the Python and Scala suites plus ruff/mypy. Co-authored-by: Isaac

…er-side Address review feedback: the handler already re-checked the both-lists case (AUTOCDC_BOTH_TRACK_HISTORY_...) but did not mirror the "track-history requires SCD2" check that the Python client performs. A raw (non-Python) Connect client sending SCD_TYPE_1 with a populated track-history list would reach ChangeArgs, whose validateTrackHistoryOnlyForScd2 guard raises SparkException.internalError (INTERNAL_ERROR) rather than a clean user-facing error. Mirror the check in PipelinesHandler.buildAutoCdcFlow with a new AUTOCDC_TRACK_HISTORY_REQUIRES_SCD2 AnalysisException, making the two track-history validations symmetric. Cover both the new requires-SCD2 guard and the existing both-lists guard with raw-proto tests in SparkDeclarativePipelinesServerSuite (the Python client rejects these client-side, so the raw stub is needed to exercise the server path). Co-authored-by: Isaac

szehon-ho · 2026-07-22T18:21:29Z

+    # a notion that only exists under SCD2. Reject them for any other SCD type. The engine
+    # enforces this too, but failing here gives a clearer, client-side error.
+    is_scd2 = stored_as_scd_type is not None and str(stored_as_scd_type) == "2"
+    if not is_scd2 and (


The empty-list case is treated as "provided". This gate uses is not None, so track_history_column_list=[] (or [] for the except variant) raises INVALID_MULTIPLE_ARGUMENT_CONDITIONS when the flow is not SCD2 -- even though an empty list serializes identically to an omitted one (to_plans([]) -> [] -> unset repeated field on the wire) and is thus a no-op. Consider gating on non-empty (e.g. truthiness of the lists) so an empty list is accepted/ignored consistently with how it is serialized.

Good point - addressed

szehon-ho · 2026-07-22T18:21:29Z

+            errorClass="INVALID_MULTIPLE_ARGUMENT_CONDITIONS",
+            messageParameters={
+                "arg_names": "track_history_column_list, track_history_except_column_list",
+                "condition": "specified unless stored_as_scd_type is 2 (or '2')",


Minor wording nit: the rendered message always names both parameters. With the template "[<arg_names>] cannot be <condition>.", a caller who passed only track_history_column_list still gets [track_history_column_list, track_history_except_column_list] cannot be specified unless stored_as_scd_type is 2 (or '2'). Accurate as a rule about the pair, but less precise than naming the argument actually supplied. Optional.

szehon-ho · 2026-07-22T18:21:29Z

    except_column_list = _normalize_optional_column_list(
        arg_name="except_column_list", column_list=except_column_list
    )
+    track_history_column_list = _normalize_optional_column_list(


Asymmetry with the docstring's "Only one ... can be specified" promise: the client validates the SCD2 requirement, but specifying both track_history_column_list and track_history_except_column_list is only rejected server-side (AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST in PipelinesHandler.scala). This mirrors the existing column_list/except_column_list handling (also server-only), so it is a consistent choice -- flagging only because the "both" case costs a server round-trip. Fine to leave as-is for consistency.

Went ahead and enforced this client-side for both pairs (b0f36da) rather than leaving it server-only — added a
_reject_include_and_except_together helper raising CANNOT_SET_TOGETHER, covering column_list/except_column_list and the track-history pair, so it now matches the docstrings' "only one" promise without a round-trip. Kept the server-side checks as defense for raw Connect clients and moved their coverage to raw-proto tests in SparkDeclarativePipelinesServerSuite.

…args Two review nits on the client-side "track history requires SCD2" check in create_auto_cdc_flow: - The gate used `is not None`, so an empty list (`track_history_column_list=[]`) raised INVALID_MULTIPLE_ARGUMENT_CONDITIONS even though an empty list serializes identically to an omitted one (an unset repeated field) and is a no-op. Gate on non-empty lists instead, so `[]` is accepted and ignored consistently with how it is serialized. - The error named both parameters regardless of which was supplied. Build the arg_names list from the arguments actually passed, so a caller who supplied only one sees only that one. Add a test that an empty track-history list is accepted without SCD2. Co-authored-by: Isaac

Reject specifying both an include list and its except counterpart client-side in create_auto_cdc_flow, for both the column_list/except_column_list pair and the track_history_column_list/track_history_except_column_list pair, raising CANNOT_SET_TOGETHER (a PySparkValueError). Previously only the server rejected this, costing a round-trip; the docstrings already promised "only one ... can be specified". Add a shared _reject_include_and_except_together helper that gates on non-empty lists, so an empty list (a no-op that serializes to an unset repeated field) is not treated as "specified". The server-side defenses (AUTOCDC_BOTH_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST and AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST) remain for raw Connect clients; move their coverage to raw-proto tests in SparkDeclarativePipelinesServerSuite and update the Python-driven tests in PythonPipelineSuite to assert the client-side CANNOT_SET_TOGETHER error. Co-authored-by: Isaac

The sql/connect modules enforce scalafmt in CI; reformat the AutoCDC raw-proto test helper and test to satisfy it. Formatting only, no behavior change. Co-authored-by: Isaac

anew · 2026-07-22T20:35:39Z

    except_column_list: Optional[Union[List[str], List[Column]]] = None,
-    stored_as_scd_type: Optional[Literal[1, "1"]] = None,
+    stored_as_scd_type: Optional[Literal[1, 2, "1", "2"]] = None,
+    track_history_column_list: Optional[Union[List[str], List[Column]]] = None,


The new track_history_* parameters are inserted before the
existing name parameter in a public function that accepts positional arguments. Existing calls that
pass name positionally now bind that string to track_history_column_list, then fail validation as
“expected list[str] or list[Column]”. To preserve compatibility, keep name in its old positional slot
and add the new parameters after it, ideally keyword-only:

name: Optional[str] = None, *, track_history_column_list: Optional[...] = None, track_history_except_column_list: Optional[...] = None,

[Flagged by Codex 5.5]

Restored name to its original positional slot and made both track_history_* params keyword-only (after *), so existing positional calls keep working and the new params can't be mis-bound positionally.

The track_history_column_list / track_history_except_column_list parameters were inserted before the existing `name` parameter, so an existing caller passing `name` positionally would bind that string to track_history_column_list and fail validation. Restore `name` to its original positional slot and add the new parameters after it as keyword-only (after `*`), preserving backward compatibility and preventing positional misuse. Co-authored-by: Isaac

… and Spark Connect APIs ### What changes were proposed in this pull request? Expose Auto CDC SCD Type 2 through the Python and Spark Connect APIs. `create_auto_cdc_flow` now accepts `stored_as_scd_type=2` (or `"2"`) and the SCD2-only `track_history_column_list` / `track_history_except_column_list` parameters, which flow through to `ChangeArgs(storedAsScdType = Type2, trackHistorySelection = ...)` on the server. Specifically: - Proto: add `SCD_TYPE_2` to the `SCDType` enum and `track_history_column_list` (field 11) / `track_history_except_column_list` (field 12) to `AutoCdcFlowDetails`. Field numbers match the Databricks runtime extension proto for wire compatibility. Regenerate the Python stubs. - Python API (`api.py`, `flow.py`): widen `stored_as_scd_type`, add and normalize the track-history parameters, and reject track-history columns unless SCD2 is selected. - Serialization (`spark_connect_graph_element_registry.py`): map the SCD type and serialize the track-history columns. - Server (`PipelinesHandler.scala`): map `SCD_TYPE_2` to `ScdType.Type2`, build `trackHistorySelection`, and reject specifying both track-history lists. - Add error condition `AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST`. The engine's SCD2 execution path is landing in a parallel workstream and still rejects SCD2 during graph analysis (`AUTOCDC_SCD2_NOT_SUPPORTED`). This change wires the API to the `ChangeArgs` contract those changes plug into; end-to-end SCD2 execution starts working once the engine drops those throws, with no further API change required. ### Why are the changes needed? The Python and Spark Connect Auto CDC APIs previously only supported SCD Type 1. SCD Type 2 (history tracking) is a common CDC requirement and is already modeled in the engine's `ChangeArgs`/`ScdType`. ### Does this PR introduce any user-facing change? Yes. `create_auto_cdc_flow` accepts `stored_as_scd_type=2` and the `track_history_column_list` / `track_history_except_column_list` parameters. ### How was this patch tested? New unit tests in `test_auto_cdc_flow.py` (SCD2 accept as int/str, track-history include/exclude, track-history-requires-SCD2 rejection) and `PythonPipelineSuite.scala` (SCD2 registration forwards to `ChangeArgs`, both-lists rejection). Ran the Python and Scala suites plus ruff/mypy. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Opus 4.8 Closes #57412 from anew/spark-58247-autocdc-scd2-python-api. Authored-by: Andreas Neumann <anew@apache.org> Signed-off-by: Szehon Ho <szehon.apache@gmail.com> (cherry picked from commit 3d59361) Signed-off-by: Szehon Ho <szehon.apache@gmail.com>

szehon-ho · 2026-07-23T00:26:23Z

Merge Summary:

merged into master 3d59361
merged into branch-4.x 5cdb1a6

Posted by merge_spark_pr.py

anew changed the title ~~[SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs~~ [WIP][SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs Jul 22, 2026

anew force-pushed the spark-58247-autocdc-scd2-python-api branch from 3d8641a to d1253ee Compare July 22, 2026 03:58

uros-b reviewed Jul 22, 2026

View reviewed changes

Comment thread sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala

szehon-ho reviewed Jul 22, 2026

View reviewed changes

anew changed the title ~~[WIP][SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs~~ [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs Jul 22, 2026

anew added 3 commits July 22, 2026 19:28

apply scalafmt to SparkDeclarativePipelinesServerSuite

479ca82

The sql/connect modules enforce scalafmt in CI; reformat the AutoCDC raw-proto test helper and test to satisfy it. Formatting only, no behavior change. Co-authored-by: Isaac

anew commented Jul 22, 2026

View reviewed changes

szehon-ho approved these changes Jul 22, 2026

View reviewed changes

szehon-ho closed this in 3d59361 Jul 23, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs#57412

[SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs#57412
anew wants to merge 6 commits into
apache:masterfrom
anew:spark-58247-autocdc-scd2-python-api

anew commented Jul 22, 2026

Uh oh!

Uh oh!

szehon-ho Jul 22, 2026

Uh oh!

anew Jul 22, 2026

Uh oh!

szehon-ho Jul 22, 2026

Uh oh!

anew Jul 22, 2026

Uh oh!

szehon-ho Jul 22, 2026

Uh oh!

anew Jul 22, 2026

Uh oh!

anew Jul 22, 2026

Uh oh!

anew Jul 22, 2026

Uh oh!

anew Jul 22, 2026

Uh oh!

szehon-ho commented Jul 23, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

anew commented Jul 22, 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!

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!

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!

szehon-ho commented Jul 23, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants