This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5cdb1a6e9088 [SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the
Auto CDC Python and Spark Connect APIs
5cdb1a6e9088 is described below
commit 5cdb1a6e9088e42b7c19aa6372484b341f16a1d2
Author: Andreas Neumann <[email protected]>
AuthorDate: Wed Jul 22 17:26:05 2026 -0700
[SPARK-58247][CONNECT][SDP] Support SCD Type 2 in the Auto CDC Python 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 <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 3d5936176cb23416e62d8dc15d406ac1b33c7064)
Signed-off-by: Szehon Ho <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 12 ++
python/pyspark/pipelines/api.py | 92 +++++++++++-
python/pyspark/pipelines/flow.py | 12 +-
.../spark_connect_graph_element_registry.py | 8 +-
.../pyspark/pipelines/tests/test_auto_cdc_flow.py | 156 ++++++++++++++++++++-
python/pyspark/sql/connect/proto/pipelines_pb2.py | 76 +++++-----
python/pyspark/sql/connect/proto/pipelines_pb2.pyi | 34 +++++
.../main/protobuf/spark/connect/pipelines.proto | 9 ++
.../sql/connect/pipelines/PipelinesHandler.scala | 32 ++++-
.../connect/pipelines/PythonPipelineSuite.scala | 110 ++++++++++++++-
.../SparkDeclarativePipelinesServerSuite.scala | 96 +++++++++++++
11 files changed, 581 insertions(+), 56 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 854923c02eb9..0ec3f4d76ed7 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -221,6 +221,12 @@
],
"sqlState" : "42613"
},
+ "AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST" : {
+ "message" : [
+ "AutoCDC flow specifies both `track_history_column_list` and
`track_history_except_column_list`; at most one may be provided."
+ ],
+ "sqlState" : "42613"
+ },
"AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA" : {
"message" : [
"Using <caseSensitivity> column name comparison, the following columns
are not present in the <schemaName> schema: <missingColumns>. Available
columns: <availableColumns>."
@@ -343,6 +349,12 @@
],
"sqlState" : "0A000"
},
+ "AUTOCDC_TRACK_HISTORY_REQUIRES_SCD2" : {
+ "message" : [
+ "AutoCDC flow specifies `track_history_column_list` or
`track_history_except_column_list` but is not stored as SCD Type 2;
history-tracking columns are only supported for SCD Type 2 target tables."
+ ],
+ "sqlState" : "42613"
+ },
"AVRO_CANNOT_WRITE_NULL_FIELD" : {
"message" : [
"Cannot write null value for field <name> defined as non-null Avro data
type <dataType>.",
diff --git a/python/pyspark/pipelines/api.py b/python/pyspark/pipelines/api.py
index 5cbc003708f2..19bd47e4af76 100644
--- a/python/pyspark/pipelines/api.py
+++ b/python/pyspark/pipelines/api.py
@@ -16,7 +16,7 @@
#
from typing import Callable, Dict, List, Literal, Optional, Union, overload
-from pyspark.errors import PySparkTypeError
+from pyspark.errors import PySparkTypeError, PySparkValueError
from pyspark.pipelines.graph_element_registry import
get_active_graph_element_registry
from pyspark.pipelines.type_error_utils import
validate_optional_list_of_str_arg
from pyspark.pipelines.flow import AutoCdcFlow, Flow, QueryFunction
@@ -536,8 +536,11 @@ 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,
name: Optional[str] = None,
+ *,
+ track_history_column_list: Optional[Union[List[str], List[Column]]] = None,
+ track_history_except_column_list: Optional[Union[List[str], List[Column]]]
= None,
) -> None:
"""
Create an Auto CDC flow into the target table from the Change Data Capture
(CDC) source.
@@ -581,8 +584,20 @@ def create_auto_cdc_flow(
PySpark Columns. Only one of column_list and except_column_list can be
specified. When \
this is specified, all columns in the `DataFrame` of the target table
except those in \
this list will be in the output table.
- :param stored_as_scd_type: The SCD type for the target table. Only 1 (or
"1") is supported. \
- When not specified, the server default applies.
+ :param stored_as_scd_type: The SCD type for the target table. 1 (or "1")
and 2 (or "2") are \
+ supported. When not specified, the server default applies.
+ :param track_history_column_list: SCD2-only. Columns whose value change
opens a new history \
+ record; two consecutive upsert events for the same key are coalesced
into the same \
+ history record when they agree on every tracked column. When not
specified, every \
+ eligible selected user column is tracked. This should be a list of
column identifiers \
+ without qualifiers, expressed as either Python strings or PySpark
Columns. Only one of \
+ track_history_column_list and track_history_except_column_list can be
specified, and \
+ both require stored_as_scd_type to be 2.
+ :param track_history_except_column_list: SCD2-only. Columns excluded from
history tracking. \
+ This should be a list of column identifiers without qualifiers,
expressed as either \
+ Python strings or PySpark Columns. Only one of
track_history_column_list and \
+ track_history_except_column_list can be specified, and both require
stored_as_scd_type \
+ to be 2.
:param name: The name of the flow for this create_auto_cdc_flow command.
When unspecified, \
this will build a "default flow" with name equal to the target name.
"""
@@ -627,6 +642,27 @@ 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(
+ arg_name="track_history_column_list",
column_list=track_history_column_list
+ )
+ track_history_except_column_list = _normalize_optional_column_list(
+ arg_name="track_history_except_column_list",
+ column_list=track_history_except_column_list,
+ )
+
+ # An include/except pair is mutually exclusive. The server enforces this
too, but failing
+ # here avoids a round-trip. An empty list serializes identically to an
omitted one (an unset
+ # repeated field on the wire), so it is a no-op; gate on non-empty lists
rather than
+ # `is not None`.
+ _reject_include_and_except_together(
+ "column_list", column_list, "except_column_list", except_column_list
+ )
+ _reject_include_and_except_together(
+ "track_history_column_list",
+ track_history_column_list,
+ "track_history_except_column_list",
+ track_history_except_column_list,
+ )
if isinstance(sequence_by, str):
sequence_by = _connect_expr(sequence_by)
@@ -652,16 +688,40 @@ 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.
+ #
+ # An empty list serializes identically to an omitted one (an unset
repeated field on the
+ # wire) and is therefore a no-op, so gate on non-empty lists rather than
`is not None`.
+ is_scd2 = stored_as_scd_type is not None and str(stored_as_scd_type) == "2"
+ supplied_track_history_args = [
+ arg_name
+ for arg_name, cols in (
+ ("track_history_column_list", track_history_column_list),
+ ("track_history_except_column_list",
track_history_except_column_list),
+ )
+ if cols
+ ]
+ if not is_scd2 and supplied_track_history_args:
+ raise PySparkValueError(
+ errorClass="INVALID_MULTIPLE_ARGUMENT_CONDITIONS",
+ messageParameters={
+ "arg_names": ", ".join(supplied_track_history_args),
+ "condition": "specified unless stored_as_scd_type is 2 (or
'2')",
+ },
+ )
+
source_code_location = get_caller_source_code_location(stacklevel=1)
flow = AutoCdcFlow(
@@ -674,12 +734,32 @@ def create_auto_cdc_flow(
column_list=column_list,
except_column_list=except_column_list,
stored_as_scd_type=stored_as_scd_type,
+ track_history_column_list=track_history_column_list,
+ track_history_except_column_list=track_history_except_column_list,
source_code_location=source_code_location,
)
get_active_graph_element_registry().register_auto_cdc_flow(flow)
+def _reject_include_and_except_together(
+ include_arg_name: str,
+ include_columns: Optional[List[Column]],
+ except_arg_name: str,
+ except_columns: Optional[List[Column]],
+) -> None:
+ """Reject specifying a non-empty include list and a non-empty except list
together.
+
+ An empty list is a no-op (it serializes to an unset repeated field,
indistinguishable from an
+ omitted argument), so only non-empty lists count as "specified".
+ """
+ if include_columns and except_columns:
+ raise PySparkValueError(
+ errorClass="CANNOT_SET_TOGETHER",
+ messageParameters={"arg_list": f"{include_arg_name} and
{except_arg_name}"},
+ )
+
+
def _normalize_optional_column_list(
arg_name: str,
column_list: Optional[Union[List[str], List[Column]]],
diff --git a/python/pyspark/pipelines/flow.py b/python/pyspark/pipelines/flow.py
index b1922454a551..8a0f44e2abe9 100644
--- a/python/pyspark/pipelines/flow.py
+++ b/python/pyspark/pipelines/flow.py
@@ -59,8 +59,12 @@ class AutoCdcFlow:
:param apply_as_deletes: Optional delete condition for the merge operation.
:param column_list: Optional columns to include in the output table.
:param except_column_list: Optional columns to exclude from the output
table.
- :param stored_as_scd_type: Optional SCD type for the target table. Only 1
(or "1") is \
- supported.
+ :param stored_as_scd_type: Optional SCD type for the target table. 1 (or
"1") and 2 (or "2") \
+ are supported.
+ :param track_history_column_list: Optional SCD2-only columns whose value
change opens a new \
+ history record.
+ :param track_history_except_column_list: Optional SCD2-only columns
excluded from history \
+ tracking.
:param source_code_location: The location of the source code that created
this flow.
"""
@@ -72,5 +76,7 @@ class AutoCdcFlow:
apply_as_deletes: Optional[Column]
column_list: Optional[List[Column]]
except_column_list: Optional[List[Column]]
- stored_as_scd_type: Optional[Literal[1, "1"]]
+ stored_as_scd_type: Optional[Literal[1, 2, "1", "2"]]
+ track_history_column_list: Optional[List[Column]]
+ track_history_except_column_list: Optional[List[Column]]
source_code_location: SourceCodeLocation
diff --git a/python/pyspark/pipelines/spark_connect_graph_element_registry.py
b/python/pyspark/pipelines/spark_connect_graph_element_registry.py
index 2eef264990a3..a46eaca5f4cb 100644
--- a/python/pyspark/pipelines/spark_connect_graph_element_registry.py
+++ b/python/pyspark/pipelines/spark_connect_graph_element_registry.py
@@ -148,9 +148,15 @@ class
SparkConnectGraphElementRegistry(GraphElementRegistry):
sequence_by=to_plan(flow.sequence_by),
column_list=to_plans(flow.column_list),
except_column_list=to_plans(flow.except_column_list),
+ track_history_column_list=to_plans(flow.track_history_column_list),
+
track_history_except_column_list=to_plans(flow.track_history_except_column_list),
)
if flow.stored_as_scd_type is not None:
- auto_cdc_details.stored_as_scd_type =
pb2.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_1
+ scd_type_by_value = {
+ "1": pb2.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_1,
+ "2": pb2.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_2,
+ }
+ auto_cdc_details.stored_as_scd_type =
scd_type_by_value[str(flow.stored_as_scd_type)]
if flow.apply_as_deletes is not None:
auto_cdc_details.apply_as_deletes.CopyFrom(to_plan(flow.apply_as_deletes))
diff --git a/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
b/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
index c52e8a09e8a6..c142e1b1562c 100644
--- a/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
+++ b/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
@@ -18,7 +18,7 @@
import unittest
from typing import cast
-from pyspark.errors import PySparkRuntimeError, PySparkTypeError
+from pyspark.errors import PySparkRuntimeError, PySparkTypeError,
PySparkValueError
from pyspark.sql import Column
from pyspark.testing.connectutils import (
should_test_connect,
@@ -117,6 +117,73 @@ class AutoCdcFlowConstructionTest(unittest.TestCase):
flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
self.assertEqual(flow.stored_as_scd_type, "1")
+ def test_create_auto_cdc_flow_stored_as_scd_type_2(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ stored_as_scd_type=2,
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertEqual(flow.stored_as_scd_type, 2)
+
+ def test_create_auto_cdc_flow_stored_as_scd_type_2_string(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ stored_as_scd_type="2",
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertEqual(flow.stored_as_scd_type, "2")
+
+ def test_create_auto_cdc_flow_with_track_history_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ stored_as_scd_type=2,
+ track_history_column_list=["val"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ assert flow.track_history_column_list is not None
+ self.assertEqual(len(flow.track_history_column_list), 1)
+ self.assertIsInstance(flow.track_history_column_list[0], Column)
+ self.assertIsNone(flow.track_history_except_column_list)
+
+ def test_create_auto_cdc_flow_with_track_history_except_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ stored_as_scd_type=2,
+ track_history_except_column_list=["op", "seq"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ assert flow.track_history_except_column_list is not None
+ self.assertEqual(len(flow.track_history_except_column_list), 2)
+ self.assertIsNone(flow.track_history_column_list)
+
def test_create_auto_cdc_flow_invalid_scd_type(self):
registry = LocalGraphElementRegistry()
with graph_element_registration_context(registry):
@@ -127,10 +194,44 @@ class AutoCdcFlowConstructionTest(unittest.TestCase):
source="s",
keys=[col("k")],
sequence_by=expr("seq"),
- stored_as_scd_type=2, # type: ignore[arg-type]
+ stored_as_scd_type=3, # type: ignore[arg-type]
)
self.assertEqual(ctx.exception.getCondition(), "NOT_EXPECTED_TYPE")
+ def test_create_auto_cdc_flow_track_history_requires_scd2(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ stored_as_scd_type=1,
+ track_history_column_list=["val"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"INVALID_MULTIPLE_ARGUMENT_CONDITIONS")
+
+ def
test_create_auto_cdc_flow_empty_track_history_list_is_accepted_without_scd2(self):
+ # An empty list serializes identically to an omitted one (an unset
repeated field), so it
+ # is a no-op and must not trip the "track history requires SCD2" check.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ track_history_column_list=[],
+ track_history_except_column_list=[],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertEqual(flow.track_history_column_list, [])
+ self.assertEqual(flow.track_history_except_column_list, [])
+
def test_create_auto_cdc_flow_with_except_column_list(self):
registry = LocalGraphElementRegistry()
with graph_element_registration_context(registry):
@@ -150,6 +251,57 @@ class AutoCdcFlowConstructionTest(unittest.TestCase):
for c in flow.except_column_list:
self.assertIsInstance(c, Column)
+ def test_create_auto_cdc_flow_rejects_both_column_list_and_except(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ column_list=["id"],
+ except_column_list=["op"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_both_track_history_column_list_and_except(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ stored_as_scd_type=2,
+ track_history_column_list=["val"],
+ track_history_except_column_list=["op"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_empty_column_lists_not_treated_as_both_set(self):
+ # Empty lists are no-ops (they serialize to unset repeated fields), so
an empty include
+ # list alongside a non-empty except list must not trip the
mutual-exclusivity check.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ column_list=[],
+ except_column_list=["op"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertEqual(flow.column_list, [])
+ assert flow.except_column_list is not None
+ self.assertEqual(len(flow.except_column_list), 1)
+
def test_create_auto_cdc_flow_rejects_non_str_target(self):
registry = LocalGraphElementRegistry()
with graph_element_registration_context(registry):
diff --git a/python/pyspark/sql/connect/proto/pipelines_pb2.py
b/python/pyspark/sql/connect/proto/pipelines_pb2.py
index ec8142c1d2aa..c32f27f26cdc 100644
--- a/python/pyspark/sql/connect/proto/pipelines_pb2.py
+++ b/python/pyspark/sql/connect/proto/pipelines_pb2.py
@@ -44,7 +44,7 @@ from pyspark.sql.connect.proto import types_pb2 as
spark_dot_connect_dot_types__
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
-
b'\n\x1dspark/connect/pipelines.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1aspark/connect/common.proto\x1a\x1fspark/connect/expressions.proto\x1a\x1dspark/connect/relations.proto\x1a\x19spark/connect/types.proto"\x83/\n\x0fPipelineCommand\x12h\n\x15\x63reate_dataflow_graph\x18\x01
\x01(\x0b\x32\x32.spark.connect.PipelineCommand.CreateDataflowGraphH\x00R\x13\x63reateDataflowGraph\x12R\n\rdefine_output\x18\x02
\x01(\x0b\x32+.s [...]
+
b'\n\x1dspark/connect/pipelines.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1aspark/connect/common.proto\x1a\x1fspark/connect/expressions.proto\x1a\x1dspark/connect/relations.proto\x1a\x19spark/connect/types.proto"\xcc\x30\n\x0fPipelineCommand\x12h\n\x15\x63reate_dataflow_graph\x18\x01
\x01(\x0b\x32\x32.spark.connect.PipelineCommand.CreateDataflowGraphH\x00R\x13\x63reateDataflowGraph\x12R\n\rdefine_output\x18\x02
\x01(\x0b\x32 [...]
)
_globals = globals()
@@ -87,10 +87,10 @@ if not _descriptor._USE_C_DESCRIPTORS:
_globals["_PIPELINEANALYSISCONTEXT"].fields_by_name[
"flow_name"
]._serialized_options = b"\030\001"
- _globals["_OUTPUTTYPE"]._serialized_start = 7966
- _globals["_OUTPUTTYPE"]._serialized_end = 8071
+ _globals["_OUTPUTTYPE"]._serialized_start = 8167
+ _globals["_OUTPUTTYPE"]._serialized_end = 8272
_globals["_PIPELINECOMMAND"]._serialized_start = 228
- _globals["_PIPELINECOMMAND"]._serialized_end = 6247
+ _globals["_PIPELINECOMMAND"]._serialized_end = 6448
_globals["_PIPELINECOMMAND_CREATEDATAFLOWGRAPH"]._serialized_start = 1265
_globals["_PIPELINECOMMAND_CREATEDATAFLOWGRAPH"]._serialized_end = 1573
_globals["_PIPELINECOMMAND_CREATEDATAFLOWGRAPH_SQLCONFENTRY"]._serialized_start
= 1474
@@ -112,43 +112,43 @@ if not _descriptor._USE_C_DESCRIPTORS:
_globals["_PIPELINECOMMAND_DEFINEOUTPUT_SINKDETAILS_OPTIONSENTRY"]._serialized_start
= 2795
_globals["_PIPELINECOMMAND_DEFINEOUTPUT_SINKDETAILS_OPTIONSENTRY"]._serialized_end
= 2853
_globals["_PIPELINECOMMAND_DEFINEFLOW"]._serialized_start = 2969
- _globals["_PIPELINECOMMAND_DEFINEFLOW"]._serialized_end = 4855
+ _globals["_PIPELINECOMMAND_DEFINEFLOW"]._serialized_end = 5056
_globals["_PIPELINECOMMAND_DEFINEFLOW_SQLCONFENTRY"]._serialized_start =
1474
_globals["_PIPELINECOMMAND_DEFINEFLOW_SQLCONFENTRY"]._serialized_end = 1532
_globals["_PIPELINECOMMAND_DEFINEFLOW_WRITERELATIONFLOWDETAILS"]._serialized_start
= 3703
_globals["_PIPELINECOMMAND_DEFINEFLOW_WRITERELATIONFLOWDETAILS"]._serialized_end
= 3800
_globals["_PIPELINECOMMAND_DEFINEFLOW_AUTOCDCFLOWDETAILS"]._serialized_start =
3803
- _globals["_PIPELINECOMMAND_DEFINEFLOW_AUTOCDCFLOWDETAILS"]._serialized_end
= 4623
- _globals["_PIPELINECOMMAND_DEFINEFLOW_RESPONSE"]._serialized_start = 4625
- _globals["_PIPELINECOMMAND_DEFINEFLOW_RESPONSE"]._serialized_end = 4683
- _globals["_PIPELINECOMMAND_DEFINEFLOW_SCDTYPE"]._serialized_start = 4685
- _globals["_PIPELINECOMMAND_DEFINEFLOW_SCDTYPE"]._serialized_end = 4736
- _globals["_PIPELINECOMMAND_EXECUTEOUTPUTFLOWS"]._serialized_start = 4858
- _globals["_PIPELINECOMMAND_EXECUTEOUTPUTFLOWS"]._serialized_end = 5214
- _globals["_PIPELINECOMMAND_STARTRUN"]._serialized_start = 5217
- _globals["_PIPELINECOMMAND_STARTRUN"]._serialized_end = 5539
- _globals["_PIPELINECOMMAND_DEFINESQLGRAPHELEMENTS"]._serialized_start =
5542
- _globals["_PIPELINECOMMAND_DEFINESQLGRAPHELEMENTS"]._serialized_end = 5741
-
_globals["_PIPELINECOMMAND_GETQUERYFUNCTIONEXECUTIONSIGNALSTREAM"]._serialized_start
= 5744
-
_globals["_PIPELINECOMMAND_GETQUERYFUNCTIONEXECUTIONSIGNALSTREAM"]._serialized_end
= 5902
-
_globals["_PIPELINECOMMAND_DEFINEFLOWQUERYFUNCTIONRESULT"]._serialized_start =
5905
- _globals["_PIPELINECOMMAND_DEFINEFLOWQUERYFUNCTIONRESULT"]._serialized_end
= 6231
- _globals["_PIPELINECOMMANDRESULT"]._serialized_start = 6250
- _globals["_PIPELINECOMMANDRESULT"]._serialized_end = 7002
-
_globals["_PIPELINECOMMANDRESULT_CREATEDATAFLOWGRAPHRESULT"]._serialized_start
= 6619
-
_globals["_PIPELINECOMMANDRESULT_CREATEDATAFLOWGRAPHRESULT"]._serialized_end =
6717
- _globals["_PIPELINECOMMANDRESULT_DEFINEOUTPUTRESULT"]._serialized_start =
6720
- _globals["_PIPELINECOMMANDRESULT_DEFINEOUTPUTRESULT"]._serialized_end =
6853
- _globals["_PIPELINECOMMANDRESULT_DEFINEFLOWRESULT"]._serialized_start =
6856
- _globals["_PIPELINECOMMANDRESULT_DEFINEFLOWRESULT"]._serialized_end = 6987
- _globals["_PIPELINEEVENTRESULT"]._serialized_start = 7004
- _globals["_PIPELINEEVENTRESULT"]._serialized_end = 7077
- _globals["_PIPELINEEVENT"]._serialized_start = 7079
- _globals["_PIPELINEEVENT"]._serialized_end = 7195
- _globals["_SOURCECODELOCATION"]._serialized_start = 7198
- _globals["_SOURCECODELOCATION"]._serialized_end = 7439
- _globals["_PIPELINEQUERYFUNCTIONEXECUTIONSIGNAL"]._serialized_start = 7442
- _globals["_PIPELINEQUERYFUNCTIONEXECUTIONSIGNAL"]._serialized_end = 7593
- _globals["_PIPELINEANALYSISCONTEXT"]._serialized_start = 7596
- _globals["_PIPELINEANALYSISCONTEXT"]._serialized_end = 7964
+ _globals["_PIPELINECOMMAND_DEFINEFLOW_AUTOCDCFLOWDETAILS"]._serialized_end
= 4808
+ _globals["_PIPELINECOMMAND_DEFINEFLOW_RESPONSE"]._serialized_start = 4810
+ _globals["_PIPELINECOMMAND_DEFINEFLOW_RESPONSE"]._serialized_end = 4868
+ _globals["_PIPELINECOMMAND_DEFINEFLOW_SCDTYPE"]._serialized_start = 4870
+ _globals["_PIPELINECOMMAND_DEFINEFLOW_SCDTYPE"]._serialized_end = 4937
+ _globals["_PIPELINECOMMAND_EXECUTEOUTPUTFLOWS"]._serialized_start = 5059
+ _globals["_PIPELINECOMMAND_EXECUTEOUTPUTFLOWS"]._serialized_end = 5415
+ _globals["_PIPELINECOMMAND_STARTRUN"]._serialized_start = 5418
+ _globals["_PIPELINECOMMAND_STARTRUN"]._serialized_end = 5740
+ _globals["_PIPELINECOMMAND_DEFINESQLGRAPHELEMENTS"]._serialized_start =
5743
+ _globals["_PIPELINECOMMAND_DEFINESQLGRAPHELEMENTS"]._serialized_end = 5942
+
_globals["_PIPELINECOMMAND_GETQUERYFUNCTIONEXECUTIONSIGNALSTREAM"]._serialized_start
= 5945
+
_globals["_PIPELINECOMMAND_GETQUERYFUNCTIONEXECUTIONSIGNALSTREAM"]._serialized_end
= 6103
+
_globals["_PIPELINECOMMAND_DEFINEFLOWQUERYFUNCTIONRESULT"]._serialized_start =
6106
+ _globals["_PIPELINECOMMAND_DEFINEFLOWQUERYFUNCTIONRESULT"]._serialized_end
= 6432
+ _globals["_PIPELINECOMMANDRESULT"]._serialized_start = 6451
+ _globals["_PIPELINECOMMANDRESULT"]._serialized_end = 7203
+
_globals["_PIPELINECOMMANDRESULT_CREATEDATAFLOWGRAPHRESULT"]._serialized_start
= 6820
+
_globals["_PIPELINECOMMANDRESULT_CREATEDATAFLOWGRAPHRESULT"]._serialized_end =
6918
+ _globals["_PIPELINECOMMANDRESULT_DEFINEOUTPUTRESULT"]._serialized_start =
6921
+ _globals["_PIPELINECOMMANDRESULT_DEFINEOUTPUTRESULT"]._serialized_end =
7054
+ _globals["_PIPELINECOMMANDRESULT_DEFINEFLOWRESULT"]._serialized_start =
7057
+ _globals["_PIPELINECOMMANDRESULT_DEFINEFLOWRESULT"]._serialized_end = 7188
+ _globals["_PIPELINEEVENTRESULT"]._serialized_start = 7205
+ _globals["_PIPELINEEVENTRESULT"]._serialized_end = 7278
+ _globals["_PIPELINEEVENT"]._serialized_start = 7280
+ _globals["_PIPELINEEVENT"]._serialized_end = 7396
+ _globals["_SOURCECODELOCATION"]._serialized_start = 7399
+ _globals["_SOURCECODELOCATION"]._serialized_end = 7640
+ _globals["_PIPELINEQUERYFUNCTIONEXECUTIONSIGNAL"]._serialized_start = 7643
+ _globals["_PIPELINEQUERYFUNCTIONEXECUTIONSIGNAL"]._serialized_end = 7794
+ _globals["_PIPELINEANALYSISCONTEXT"]._serialized_start = 7797
+ _globals["_PIPELINEANALYSISCONTEXT"]._serialized_end = 8165
# @@protoc_insertion_point(module_scope)
diff --git a/python/pyspark/sql/connect/proto/pipelines_pb2.pyi
b/python/pyspark/sql/connect/proto/pipelines_pb2.pyi
index f327a7e92ac8..70e04ad85042 100644
--- a/python/pyspark/sql/connect/proto/pipelines_pb2.pyi
+++ b/python/pyspark/sql/connect/proto/pipelines_pb2.pyi
@@ -519,12 +519,14 @@ class PipelineCommand(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
SCD_TYPE_UNSPECIFIED:
PipelineCommand.DefineFlow._SCDType.ValueType # 0
SCD_TYPE_1: PipelineCommand.DefineFlow._SCDType.ValueType # 1
+ SCD_TYPE_2: PipelineCommand.DefineFlow._SCDType.ValueType # 2
class SCDType(_SCDType, metaclass=_SCDTypeEnumTypeWrapper):
"""SCD Type for Auto CDC target tables."""
SCD_TYPE_UNSPECIFIED: PipelineCommand.DefineFlow.SCDType.ValueType # 0
SCD_TYPE_1: PipelineCommand.DefineFlow.SCDType.ValueType # 1
+ SCD_TYPE_2: PipelineCommand.DefineFlow.SCDType.ValueType # 2
class SqlConfEntry(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.Descriptor
@@ -588,6 +590,8 @@ class PipelineCommand(google.protobuf.message.Message):
COLUMN_LIST_FIELD_NUMBER: builtins.int
EXCEPT_COLUMN_LIST_FIELD_NUMBER: builtins.int
STORED_AS_SCD_TYPE_FIELD_NUMBER: builtins.int
+ TRACK_HISTORY_COLUMN_LIST_FIELD_NUMBER: builtins.int
+ TRACK_HISTORY_EXCEPT_COLUMN_LIST_FIELD_NUMBER: builtins.int
IGNORE_NULL_UPDATES_COLUMN_LIST_FIELD_NUMBER: builtins.int
IGNORE_NULL_UPDATES_EXCEPT_COLUMN_LIST_FIELD_NUMBER: builtins.int
source: builtins.str
@@ -625,6 +629,24 @@ class PipelineCommand(google.protobuf.message.Message):
stored_as_scd_type:
global___PipelineCommand.DefineFlow.SCDType.ValueType
"""SCD Type for target table."""
@property
+ def track_history_column_list(
+ self,
+ ) ->
google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
+ pyspark.sql.connect.proto.expressions_pb2.Expression
+ ]:
+ """SCD2 only. Columns whose value change opens a new history
record. When empty, every
+ eligible selected user column is tracked.
+ """
+ @property
+ def track_history_except_column_list(
+ self,
+ ) ->
google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
+ pyspark.sql.connect.proto.expressions_pb2.Expression
+ ]:
+ """SCD2 only. Columns excluded from history tracking. Mutually
exclusive with
+ track_history_column_list.
+ """
+ @property
def ignore_null_updates_column_list(
self,
) ->
google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
@@ -657,6 +679,14 @@ class PipelineCommand(google.protobuf.message.Message):
]
| None = ...,
stored_as_scd_type:
global___PipelineCommand.DefineFlow.SCDType.ValueType = ...,
+ track_history_column_list: collections.abc.Iterable[
+ pyspark.sql.connect.proto.expressions_pb2.Expression
+ ]
+ | None = ...,
+ track_history_except_column_list: collections.abc.Iterable[
+ pyspark.sql.connect.proto.expressions_pb2.Expression
+ ]
+ | None = ...,
ignore_null_updates_column_list: collections.abc.Iterable[
pyspark.sql.connect.proto.expressions_pb2.Expression
]
@@ -718,6 +748,10 @@ class PipelineCommand(google.protobuf.message.Message):
b"source",
"stored_as_scd_type",
b"stored_as_scd_type",
+ "track_history_column_list",
+ b"track_history_column_list",
+ "track_history_except_column_list",
+ b"track_history_except_column_list",
],
) -> None: ...
@typing.overload
diff --git a/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto
b/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto
index 7632ec95b9b5..7bfc5ca39826 100644
--- a/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto
+++ b/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto
@@ -182,6 +182,14 @@ message PipelineCommand {
// SCD Type for target table.
SCDType stored_as_scd_type = 10;
+ // SCD2 only. Columns whose value change opens a new history record.
When empty, every
+ // eligible selected user column is tracked.
+ repeated Expression track_history_column_list = 11;
+
+ // SCD2 only. Columns excluded from history tracking. Mutually exclusive
with
+ // track_history_column_list.
+ repeated Expression track_history_except_column_list = 12;
+
// Subset of columns to ignore null in updates.
repeated Expression ignore_null_updates_column_list = 14;
@@ -194,6 +202,7 @@ message PipelineCommand {
enum SCDType {
SCD_TYPE_UNSPECIFIED = 0;
SCD_TYPE_1 = 1;
+ SCD_TYPE_2 = 2;
}
// If true, define the flow as a one-time flow, such as for backfill.
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
index a7249ef70e49..ca41281a812b 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
@@ -478,17 +478,47 @@ 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] = {
+ val included = autoCdcDetails.getTrackHistoryColumnListList.asScala.toSeq
+ val excluded =
autoCdcDetails.getTrackHistoryExceptColumnListList.asScala.toSeq
+ if (included.nonEmpty && excluded.nonEmpty) {
+ throw new AnalysisException(
+ "AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST",
+ Map.empty)
+ } else if (included.nonEmpty) {
+
Some(ColumnSelection.IncludeColumns(included.map(asUnqualifiedColumnName)))
+ } else if (excluded.nonEmpty) {
+
Some(ColumnSelection.ExcludeColumns(excluded.map(asUnqualifiedColumnName)))
+ } else {
+ None
+ }
+ }
+
+ // History-tracking columns only make sense under SCD2. The Python client
rejects this
+ // client-side, but a raw Connect client can still send it, so mirror the
check here to
+ // surface a clean user-facing error rather than the internal error the
[[ChangeArgs]]
+ // guard would otherwise raise.
+ if (scdType != ScdType.Type2 && trackHistorySelection.isDefined) {
+ throw new AnalysisException("AUTOCDC_TRACK_HISTORY_REQUIRES_SCD2",
Map.empty)
+ }
+
val changeArgs = ChangeArgs(
keys = keys,
sequencing = toColumn(autoCdcDetails.getSequenceBy),
storedAsScdType = scdType,
deleteCondition =
Option.when(autoCdcDetails.hasApplyAsDeletes)(toColumn(autoCdcDetails.getApplyAsDeletes)),
- columnSelection = columnSelection)
+ columnSelection = columnSelection,
+ trackHistorySelection = trackHistorySelection)
AutoCdcFlow(
identifier = flowIdentifier,
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
index 8eaf008c5809..c21a4a9ac25b 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
@@ -1268,10 +1268,10 @@ class PythonPipelineSuite
}
test("AutoCDC API: specifying both column_list and except_column_list is
rejected") {
- // The Python create_auto_cdc_flow API does not currently enforce the "at
most one" contract
- // client-side, so the proto carries both lists to the server, where the
structured error is
- // raised. If/when a Python-side check is added, this test guards against
the server-side
- // defense being silently bypassed.
+ // The Python create_auto_cdc_flow API rejects the "at most one" violation
client-side with a
+ // CANNOT_SET_TOGETHER PySparkValueError, so the request never reaches the
server. The
+ // server-side defense (AUTOCDC_BOTH_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST)
is exercised directly
+ // against a raw proto in SparkDeclarativePipelinesServerSuite.
val ex = intercept[RuntimeException] {
buildGraph("""
|@dp.table
@@ -1290,7 +1290,7 @@ class PythonPipelineSuite
|)
|""".stripMargin)
}
-
assert(ex.getMessage.contains("AUTOCDC_BOTH_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST"))
+ assert(ex.getMessage.contains("CANNOT_SET_TOGETHER"))
}
test("AutoCDC API: registered flow survives graph resolution and validation
end-to-end") {
@@ -1323,6 +1323,106 @@ class PythonPipelineSuite
assert(mergeFlow.changeArgs.storedAsScdType == ScdType.Type1)
}
+ // SCD2 flows are only asserted at registration (via buildAutoCdcFlow)
rather than through
+ // resolve().validate(): the engine's SCD2 execution path is still landing
in a parallel
+ // workstream and rejects SCD2 during graph analysis. These tests pin the
API-to-ChangeArgs
+ // contract those changes plug into.
+ test("AutoCDC API: SCD2 with track_history_column_list forwards to
ChangeArgs") {
+ val flow = buildAutoCdcFlow("""
+ |@dp.table
+ |def src():
+ | return spark.readStream.format("rate").load()
+ |
+ |dp.create_streaming_table("target")
+ |
+ |dp.create_auto_cdc_flow(
+ | target = "target",
+ | source = "src",
+ | keys = ["value"],
+ | sequence_by = "timestamp",
+ | stored_as_scd_type = 2,
+ | track_history_column_list = ["value"],
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type2)
+ assert(
+ flow.changeArgs.trackHistorySelection.contains(
+ ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value")))))
+ }
+
+ test("AutoCDC API: SCD2 with track_history_except_column_list forwards to
ChangeArgs") {
+ val flow = buildAutoCdcFlow("""
+ |@dp.table
+ |def src():
+ | return spark.readStream.format("rate").load()
+ |
+ |dp.create_streaming_table("target")
+ |
+ |dp.create_auto_cdc_flow(
+ | target = "target",
+ | source = "src",
+ | keys = ["value"],
+ | sequence_by = "timestamp",
+ | stored_as_scd_type = 2,
+ | track_history_except_column_list = ["timestamp"],
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type2)
+ assert(
+ flow.changeArgs.trackHistorySelection.contains(
+
ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("timestamp")))))
+ }
+
+ test("AutoCDC API: SCD2 without track-history columns leaves selection
unset") {
+ val flow = buildAutoCdcFlow("""
+ |@dp.table
+ |def src():
+ | return spark.readStream.format("rate").load()
+ |
+ |dp.create_streaming_table("target")
+ |
+ |dp.create_auto_cdc_flow(
+ | target = "target",
+ | source = "src",
+ | keys = ["value"],
+ | sequence_by = "timestamp",
+ | stored_as_scd_type = 2,
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type2)
+ assert(flow.changeArgs.trackHistorySelection.isEmpty)
+ }
+
+ test("AutoCDC API: specifying both track_history column lists is rejected") {
+ // The Python create_auto_cdc_flow API rejects the "at most one" violation
client-side with a
+ // CANNOT_SET_TOGETHER PySparkValueError, so the request never reaches the
server. The
+ // server-side defense
(AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST) is
+ // exercised directly against a raw proto in
SparkDeclarativePipelinesServerSuite.
+ val ex = intercept[RuntimeException] {
+ buildGraph("""
+ |@dp.table
+ |def src():
+ | return spark.readStream.format("rate").load()
+ |
+ |dp.create_streaming_table("target")
+ |
+ |dp.create_auto_cdc_flow(
+ | target = "target",
+ | source = "src",
+ | keys = ["value"],
+ | sequence_by = "timestamp",
+ | stored_as_scd_type = 2,
+ | track_history_column_list = ["value"],
+ | track_history_except_column_list = ["timestamp"],
+ |)
+ |""".stripMargin)
+ }
+ assert(ex.getMessage.contains("CANNOT_SET_TOGETHER"))
+ }
+
/**
* Executes Python code in a separate process and returns the exit code.
*
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
index 3cb45fa6e172..7ff8504a3e6a 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
@@ -115,6 +115,102 @@ class SparkDeclarativePipelinesServerSuite
assert(ex.getMessage.contains("DEFINE_FLOW_ONCE_OPTION_NOT_SUPPORTED"))
}
+ // The Python client rejects these AutoCDC misconfigurations client-side, so
drive the raw
+ // Connect stub directly to exercise the server-side validation a non-Python
client would hit.
+ private def unresolvedColumn(name: String): Expression =
+ Expression
+ .newBuilder()
+ .setUnresolvedAttribute(
+
Expression.UnresolvedAttribute.newBuilder().setUnparsedIdentifier(name))
+ .build()
+
+ private def autoCdcDefineFlow(
+ graphId: String,
+ scdType: DefineFlow.SCDType,
+ columns: Seq[String] = Seq.empty,
+ exceptColumns: Seq[String] = Seq.empty,
+ trackHistoryColumns: Seq[String] = Seq.empty,
+ trackHistoryExceptColumns: Seq[String] = Seq.empty): DefineFlow = {
+ val autoCdcDetails = DefineFlow.AutoCdcFlowDetails
+ .newBuilder()
+ .setSource("src")
+ .addKeys(unresolvedColumn("id"))
+ .setSequenceBy(unresolvedColumn("ts"))
+ .setStoredAsScdType(scdType)
+ columns.foreach(c => autoCdcDetails.addColumnList(unresolvedColumn(c)))
+ exceptColumns.foreach(c =>
autoCdcDetails.addExceptColumnList(unresolvedColumn(c)))
+ trackHistoryColumns.foreach(c =>
+ autoCdcDetails.addTrackHistoryColumnList(unresolvedColumn(c)))
+ trackHistoryExceptColumns.foreach(c =>
+ autoCdcDetails.addTrackHistoryExceptColumnList(unresolvedColumn(c)))
+ DefineFlow
+ .newBuilder()
+ .setDataflowGraphId(graphId)
+ .setFlowName("target")
+ .setTargetDatasetName("target")
+ .setAutoCdcFlowDetails(autoCdcDetails)
+ .build()
+ }
+
+ test("AutoCDC: specifying both column_list and except_column_list is
rejected server-side") {
+ val ex = intercept[Exception] {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ sendPlan(
+ buildPlanFromPipelineCommand(
+ PipelineCommand
+ .newBuilder()
+ .setDefineFlow(
+ autoCdcDefineFlow(
+ graphId,
+ DefineFlow.SCDType.SCD_TYPE_1,
+ columns = Seq("id"),
+ exceptColumns = Seq("ts")))
+ .build()))
+ }
+ }
+
assert(ex.getMessage.contains("AUTOCDC_BOTH_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST"))
+ }
+
+ test("AutoCDC: track-history columns without SCD2 are rejected server-side")
{
+ val ex = intercept[Exception] {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ sendPlan(
+ buildPlanFromPipelineCommand(
+ PipelineCommand
+ .newBuilder()
+ .setDefineFlow(autoCdcDefineFlow(
+ graphId,
+ DefineFlow.SCDType.SCD_TYPE_1,
+ trackHistoryColumns = Seq("val")))
+ .build()))
+ }
+ }
+ assert(ex.getMessage.contains("AUTOCDC_TRACK_HISTORY_REQUIRES_SCD2"))
+ }
+
+ test("AutoCDC: specifying both track-history column lists is rejected
server-side") {
+ val ex = intercept[Exception] {
+ withRawBlockingStub { implicit stub =>
+ val graphId = createDataflowGraph
+ sendPlan(
+ buildPlanFromPipelineCommand(
+ PipelineCommand
+ .newBuilder()
+ .setDefineFlow(
+ autoCdcDefineFlow(
+ graphId,
+ DefineFlow.SCDType.SCD_TYPE_2,
+ trackHistoryColumns = Seq("val"),
+ trackHistoryExceptColumns = Seq("ts")))
+ .build()))
+ }
+ }
+ assert(
+
ex.getMessage.contains("AUTOCDC_BOTH_TRACK_HISTORY_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST"))
+ }
+
test(
"Cross dependency between SQL dataset and non-SQL dataset is valid and can
be registered") {
withRawBlockingStub { implicit stub =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
