You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anew (via GitHub)" <gi...@apache.org> on 2026/07/22 00:25:58 UTC
[PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
anew opened a new pull request, #57412:
URL: https://github.com/apache/spark/pull/57412
### 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
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3633648907
##########
python/pyspark/pipelines/api.py:
##########
@@ -536,7 +536,9 @@ def create_auto_cdc_flow(
apply_as_deletes: Optional[Union[str, Column]] = None,
column_list: Optional[Union[List[str], List[Column]]] = None,
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,
Review Comment:
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.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3633351299
##########
python/pyspark/pipelines/api.py:
##########
@@ -627,6 +641,13 @@ def create_auto_cdc_flow(
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(
Review Comment:
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.
##########
python/pyspark/pipelines/api.py:
##########
@@ -652,16 +673,31 @@ def create_auto_cdc_flow(
},
)
- if stored_as_scd_type is not None and str(stored_as_scd_type) != "1":
+ if stored_as_scd_type is not None and str(stored_as_scd_type) not in ("1", "2"):
raise PySparkTypeError(
errorClass="NOT_EXPECTED_TYPE",
messageParameters={
"arg_name": "stored_as_scd_type",
- "expected_type": "Literal[1, '1']",
+ "expected_type": "Literal[1, 2, '1', '2']",
"arg_type": type(stored_as_scd_type).__name__,
},
)
+ # Track-history columns describe how consecutive upserts are coalesced into history records,
+ # 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 (
+ track_history_column_list is not None or track_history_except_column_list is not None
+ ):
+ raise PySparkValueError(
+ 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')",
Review Comment:
Fixed
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3633620869
##########
python/pyspark/pipelines/api.py:
##########
@@ -536,7 +536,9 @@ def create_auto_cdc_flow(
apply_as_deletes: Optional[Union[str, Column]] = None,
column_list: Optional[Union[List[str], List[Column]]] = None,
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,
Review Comment:
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,
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3633644249
##########
python/pyspark/pipelines/api.py:
##########
@@ -536,7 +536,9 @@ def create_auto_cdc_flow(
apply_as_deletes: Optional[Union[str, Column]] = None,
column_list: Optional[Union[List[str], List[Column]]] = None,
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,
Review Comment:
[Flagged by Codex 5.5]
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3633353751
##########
python/pyspark/pipelines/api.py:
##########
@@ -652,16 +673,31 @@ def create_auto_cdc_flow(
},
)
- if stored_as_scd_type is not None and str(stored_as_scd_type) != "1":
+ if stored_as_scd_type is not None and str(stored_as_scd_type) not in ("1", "2"):
raise PySparkTypeError(
errorClass="NOT_EXPECTED_TYPE",
messageParameters={
"arg_name": "stored_as_scd_type",
- "expected_type": "Literal[1, '1']",
+ "expected_type": "Literal[1, 2, '1', '2']",
"arg_type": type(stored_as_scd_type).__name__,
},
)
+ # Track-history columns describe how consecutive upserts are coalesced into history records,
+ # 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 (
Review Comment:
Good point - addressed
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3632279011
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -478,17 +478,39 @@ private[connect] object PipelinesHandler extends Logging {
case proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_1 |
proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_UNSPECIFIED =>
ScdType.Type1
+ case proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_2 =>
+ ScdType.Type2
case other =>
throw new UnsupportedOperationException(s"Unsupported AutoCDC SCD type: $other")
}
+ // SCD2-only history-tracking column selection. Mirrors the column_list/except_column_list
+ // handling above: at most one side may be non-empty, and an empty selection means "track
+ // every eligible column" (None).
+ val trackHistorySelection: Option[ColumnSelection] = {
Review Comment:
Good catch — you're right that the validation was asymmetric. I've mirrored the "track-history requires SCD2" check server-side in buildAutoCdcFlow (new commit), so a raw Connect client now gets a clean AUTOCDC_TRACK_HISTORY_REQUIRES_SCD2 AnalysisException instead of the INTERNAL_ERROR from the ChangeArgs guard. Added raw-proto tests in SparkDeclarativePipelinesServerSuite covering both this
guard and the both-lists case (the Python client rejects these client-side, so the raw stub is needed to reach the server path). The Python-side check stays as INVALID_MULTIPLE_ARGUMENT_CONDITIONS since it's a different exception hierarchy, with a parallel message.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #57412:
URL: https://github.com/apache/spark/pull/57412#issuecomment-5052997575
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/3d5936176cb23416e62d8dc15d406ac1b33c7064
- merged into branch-4.x https://github.com/apache/spark/commit/5cdb1a6e9088e42b7c19aa6372484b341f16a1d2
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3631587610
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -478,17 +478,39 @@ private[connect] object PipelinesHandler extends Logging {
case proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_1 |
proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_UNSPECIFIED =>
ScdType.Type1
+ case proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_2 =>
+ ScdType.Type2
case other =>
throw new UnsupportedOperationException(s"Unsupported AutoCDC SCD type: $other")
}
+ // SCD2-only history-tracking column selection. Mirrors the column_list/except_column_list
+ // handling above: at most one side may be non-empty, and an empty selection means "track
+ // every eligible column" (None).
+ val trackHistorySelection: Option[ColumnSelection] = {
Review Comment:
Note: Server-side validation is asymmetric: the handler re-checks the both-lists case (AUTOCDC_BOTH_TRACK_HISTORY_...) but does NOT mirror the "track-history requires SCD2" check. A raw/non-Python Connect client sending SCD_TYPE_1 (a currently-functional path) with a populated track-history list reaches ChangeArgs, whose validateTrackHistoryOnlyForScd2 guard throws SparkException.internalError (INTERNAL_ERROR) rather than a clean user-facing AnalysisException (verified against master: the guard's own scaladoc designates the API layer as the validation point). Python clients are shielded by the new api.py check, so this is a defense-in-depth/UX gap, not a functional bug; but since the sibling both-lists check IS mirrored here, mirroring the requires-SCD2 check too would be consistent and is worth deciding before de-WIP.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57412: [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs
URL: https://github.com/apache/spark/pull/57412
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python and Spark Connect APIs [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57412:
URL: https://github.com/apache/spark/pull/57412#discussion_r3632737306
##########
python/pyspark/pipelines/api.py:
##########
@@ -652,16 +673,31 @@ def create_auto_cdc_flow(
},
)
- if stored_as_scd_type is not None and str(stored_as_scd_type) != "1":
+ if stored_as_scd_type is not None and str(stored_as_scd_type) not in ("1", "2"):
raise PySparkTypeError(
errorClass="NOT_EXPECTED_TYPE",
messageParameters={
"arg_name": "stored_as_scd_type",
- "expected_type": "Literal[1, '1']",
+ "expected_type": "Literal[1, 2, '1', '2']",
"arg_type": type(stored_as_scd_type).__name__,
},
)
+ # Track-history columns describe how consecutive upserts are coalesced into history records,
+ # 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 (
+ track_history_column_list is not None or track_history_except_column_list is not None
+ ):
+ raise PySparkValueError(
+ 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')",
Review Comment:
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.
##########
python/pyspark/pipelines/api.py:
##########
@@ -652,16 +673,31 @@ def create_auto_cdc_flow(
},
)
- if stored_as_scd_type is not None and str(stored_as_scd_type) != "1":
+ if stored_as_scd_type is not None and str(stored_as_scd_type) not in ("1", "2"):
raise PySparkTypeError(
errorClass="NOT_EXPECTED_TYPE",
messageParameters={
"arg_name": "stored_as_scd_type",
- "expected_type": "Literal[1, '1']",
+ "expected_type": "Literal[1, 2, '1', '2']",
"arg_type": type(stored_as_scd_type).__name__,
},
)
+ # Track-history columns describe how consecutive upserts are coalesced into history records,
+ # 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 (
Review Comment:
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.
##########
python/pyspark/pipelines/api.py:
##########
@@ -627,6 +641,13 @@ def create_auto_cdc_flow(
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(
Review Comment:
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.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org