szehon-ho commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307940566
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala:
##########
@@ -935,6 +942,398 @@ class PythonPipelineSuite
assert(ex.getMessage.contains("table_with_wrong_struct_schema"))
}
+ private def buildAutoCdcFlow(pipelineSource: String): AutoCdcFlow = {
+ val graph = buildGraph(pipelineSource)
+ graph.flows
+ .collectFirst { case f: AutoCdcFlow => f }
+ .getOrElse(fail(s"Expected an AutoCdcFlow in the graph, got:
${graph.flows}"))
+ }
+
+ test("AutoCDC API: minimal flow registers an AutoCdcFlow with default name
and SCD1 default") {
+ 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",
+ |)
+ |""".stripMargin)
+
+ assert(flow.identifier == graphIdentifier("target"))
+ assert(flow.destinationIdentifier == graphIdentifier("target"))
+ assert(flow.changeArgs.keys == Seq(UnqualifiedColumnName("value")))
+ assert(flow.changeArgs.sequencing.expr.sql == "timestamp")
+ assert(flow.changeArgs.deleteCondition.isEmpty)
+ assert(flow.changeArgs.columnSelection.isEmpty)
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ }
+
+ test("AutoCDC API: composite keys are forwarded to ChangeArgs in order") {
+ 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", "timestamp"],
+ | sequence_by = "timestamp",
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.keys ==
+ Seq(UnqualifiedColumnName("value"), UnqualifiedColumnName("timestamp")))
+ }
+
+ test("AutoCDC API: apply_as_deletes is forwarded as a delete condition
column") {
+ 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",
+ | apply_as_deletes = "value % 2 = 0",
+ |)
+ |""".stripMargin)
+
+ val deleteCondition = flow.changeArgs.deleteCondition.getOrElse(
+ fail("expected apply_as_deletes to populate deleteCondition"))
+ assert(deleteCondition.expr.sql.contains("value"))
+ assert(deleteCondition.expr.sql.contains("0"))
+ }
+
+ test("AutoCDC API: column_list is forwarded as IncludeColumns") {
+ 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",
+ | column_list = ["value", "timestamp"],
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.columnSelection.contains(
+ ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("value"),
UnqualifiedColumnName("timestamp")))))
+ }
+
+ test("AutoCDC API: except_column_list is forwarded as ExcludeColumns") {
+ 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",
+ | except_column_list = ["timestamp"],
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.columnSelection.contains(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("timestamp")))))
+ }
+
+ test("AutoCDC API: explicit `name` is honored as the flow identifier") {
+ 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",
+ | name = "my_flow",
+ |)
+ |""".stripMargin)
+
+ assert(flow.identifier == graphIdentifier("my_flow"))
+ assert(flow.destinationIdentifier == graphIdentifier("target"))
+ }
+
+ test("AutoCDC API: explicit stored_as_scd_type=1 is forwarded as
ScdType.Type1") {
Review Comment:
Friendly ping on my earlier comment — this test still largely duplicates
`"AutoCDC API: minimal flow registers an AutoCdcFlow with default name and SCD1
default"` above. Both end up exercising the same `SCD_TYPE_1 |
SCD_TYPE_UNSPECIFIED => ScdType.Type1` arm of:
```scala
val scdType: ScdType = autoCdcDetails.getStoredAsScdType match {
case proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_1 |
proto.PipelineCommand.DefineFlow.SCDType.SCD_TYPE_UNSPECIFIED =>
ScdType.Type1
case other =>
throw new UnsupportedOperationException(s"Unsupported AutoCDC SCD type:
$other")
}
```
The genuinely uncovered arm is `case other`. Suggest replacing this with a
proto-level negative test that constructs an `AutoCdcFlowDetails` with a
non-`SCD_TYPE_1` value (e.g. directly via the proto builder) and asserts
`buildAutoCdcFlow` throws `UnsupportedOperationException`. Then we'd have one
test per arm of the `match` instead of two tests for the same arm. Non-blocking.
Generated-by: Claude-Opus-4.7
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -388,13 +400,123 @@ private[connect] object PipelinesHandler extends Logging
{
objectName = Option(flowIdentifier.unquotedString),
language = Some(Python()))))
case proto.PipelineCommand.DefineFlow.DetailsCase.AUTO_CDC_FLOW_DETAILS
=>
- throw new UnsupportedOperationException("AutoCdcFlowDetails is not yet
implemented.")
+ graphElementRegistry.registerFlow(
+ buildAutoCdcFlow(
+ autoCdcDetails = flow.getAutoCdcFlowDetails,
+ flow = flow,
+ flowIdentifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ defaultCatalog = defaultCatalog,
+ defaultDatabase = defaultDatabase,
+ sessionHolder = sessionHolder,
+ transformExpressionFunc = transformExpressionFunc))
case other =>
throw new UnsupportedOperationException(s"Unsupported DefineFlow
details case: $other")
}
flowIdentifier
}
+ /**
+ * Build an [[AutoCdcFlow]] from the proto-supplied AutoCDC flow details.
+ *
+ * The flow's source expression is encoded by the Python client as a
streaming-table name; we
+ * model that on the server side as a streaming [[UnresolvedRelation]] so
that pipelines flow
+ * analysis (which already handles `STREAM(t)` references) can resolve it
against the rest of
+ * the dataflow graph.
+ */
+ private def buildAutoCdcFlow(
+ autoCdcDetails: AutoCdcFlowDetails,
+ flow: proto.PipelineCommand.DefineFlow,
+ flowIdentifier: TableIdentifier,
+ destinationIdentifier: TableIdentifier,
+ defaultCatalog: String,
+ defaultDatabase: String,
+ sessionHolder: SessionHolder,
+ transformExpressionFunc: proto.Expression => Expression): AutoCdcFlow = {
+ // TODO(SPARK-57092): apply_as_truncates is declared on AutoCdcFlowDetails
but is not yet
+ // honored by the engine; wire it through once SCD1 truncate support
lands.
+ // TODO(SPARK-57093): ignore_null_updates_column_list and
ignore_null_updates_except_column_list
+ // are declared on AutoCdcFlowDetails but are not yet honored by the
engine; wire them
+ // through once SCD1 ignore-null support lands.
+ val sourcePlan: LogicalPlan = UnresolvedRelation(
Review Comment:
Both `AutoCdcFlowDetails.source` and `AutoCdcFlowDetails.sequence_by` are
`optional` in the proto:
```proto
message AutoCdcFlowDetails {
optional string source = 1;
...
optional Expression sequence_by = 3;
...
}
```
The Python `create_auto_cdc_flow` API always populates both, but a
non-Python client (or a future direct proto producer) could send a `DefineFlow`
with either field absent. Today the failure modes are:
- missing `source` → `parseMultipartIdentifier("")` throws a low-level parse
error.
- missing `sequence_by` → `transformExpressionFunc` hits the `NOT_SET` arm
and throws a generic message.
Neither matches the structured errors we throw for other missing fields
elsewhere in `defineFlow`. Suggest two cheap guards at the top of
`buildAutoCdcFlow`:
```scala
if (!autoCdcDetails.hasSource) {
throw new AnalysisException("AUTOCDC_MISSING_SOURCE", Map.empty)
}
if (!autoCdcDetails.hasSequenceBy) {
throw new AnalysisException("AUTOCDC_MISSING_SEQUENCE_BY", Map.empty)
}
```
plus corresponding `error-conditions.json` entries. Unreachable from the
Python client today, but cheap insurance against non-Python clients and keeps
the AUTOCDC error surface uniformly structured — same spirit as
@gengliangwang's pushback on the "Python validates it" framing for the
column-list conflict. Non-blocking.
Generated-by: Claude-Opus-4.7
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -244,6 +250,12 @@
],
"sqlState" : "42000"
},
+ "AUTOCDC_NON_COLUMN_IDENTIFIER" : {
+ "message" : [
+ "Expected a column identifier; got the non-attribute expression
`<expression>`. AutoCDC keys, sequence_by, column_list, and except_column_list
must reference unqualified column names."
+ ],
+ "sqlState" : "42703"
Review Comment:
Discussion / nit: `42703` is "undefined column", but this error fires when
the user supplies a syntactically valid expression that is the *wrong shape*
(not an `UnresolvedAttribute`) — the column may or may not exist, we haven't
looked yet. A closer SQLSTATE is `42601` (syntax error) or `42000` (general
invalid SQL).
That said, the adjacent `AUTOCDC_MULTIPART_COLUMN_IDENTIFIER` also uses
`42703`, so this is consistent with existing AUTOCDC convention — happy to
defer if you'd rather not litigate the family-wide choice in this PR. If you
agree `42601` is a better fit, worth changing both while the AUTOCDC error
surface is still settling. Non-blocking either way.
Generated-by: Claude-Opus-4.7
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala:
##########
@@ -935,6 +942,398 @@ class PythonPipelineSuite
assert(ex.getMessage.contains("table_with_wrong_struct_schema"))
}
+ private def buildAutoCdcFlow(pipelineSource: String): AutoCdcFlow = {
+ val graph = buildGraph(pipelineSource)
+ graph.flows
+ .collectFirst { case f: AutoCdcFlow => f }
+ .getOrElse(fail(s"Expected an AutoCdcFlow in the graph, got:
${graph.flows}"))
+ }
+
+ test("AutoCDC API: minimal flow registers an AutoCdcFlow with default name
and SCD1 default") {
+ 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",
+ |)
+ |""".stripMargin)
+
+ assert(flow.identifier == graphIdentifier("target"))
+ assert(flow.destinationIdentifier == graphIdentifier("target"))
+ assert(flow.changeArgs.keys == Seq(UnqualifiedColumnName("value")))
+ assert(flow.changeArgs.sequencing.expr.sql == "timestamp")
+ assert(flow.changeArgs.deleteCondition.isEmpty)
+ assert(flow.changeArgs.columnSelection.isEmpty)
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ }
+
+ test("AutoCDC API: composite keys are forwarded to ChangeArgs in order") {
+ 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", "timestamp"],
+ | sequence_by = "timestamp",
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.keys ==
+ Seq(UnqualifiedColumnName("value"), UnqualifiedColumnName("timestamp")))
+ }
+
+ test("AutoCDC API: apply_as_deletes is forwarded as a delete condition
column") {
+ 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",
+ | apply_as_deletes = "value % 2 = 0",
+ |)
+ |""".stripMargin)
+
+ val deleteCondition = flow.changeArgs.deleteCondition.getOrElse(
+ fail("expected apply_as_deletes to populate deleteCondition"))
+ assert(deleteCondition.expr.sql.contains("value"))
+ assert(deleteCondition.expr.sql.contains("0"))
+ }
+
+ test("AutoCDC API: column_list is forwarded as IncludeColumns") {
+ 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",
+ | column_list = ["value", "timestamp"],
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.columnSelection.contains(
+ ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("value"),
UnqualifiedColumnName("timestamp")))))
+ }
+
+ test("AutoCDC API: except_column_list is forwarded as ExcludeColumns") {
+ 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",
+ | except_column_list = ["timestamp"],
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.columnSelection.contains(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("timestamp")))))
+ }
+
+ test("AutoCDC API: explicit `name` is honored as the flow identifier") {
+ 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",
+ | name = "my_flow",
+ |)
+ |""".stripMargin)
+
+ assert(flow.identifier == graphIdentifier("my_flow"))
+ assert(flow.destinationIdentifier == graphIdentifier("target"))
+ }
+
+ test("AutoCDC API: explicit stored_as_scd_type=1 is forwarded as
ScdType.Type1") {
+ 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 = 1,
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ }
+
+ test("AutoCDC API: multi-part `keys` column is rejected at flow
registration") {
+ val ex = intercept[RuntimeException] {
+ 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 = ["a.b"],
+ | sequence_by = "timestamp",
+ |)
+ |""".stripMargin)
+ }
+ assert(ex.getMessage.contains("AUTOCDC_MULTIPART_COLUMN_IDENTIFIER"))
+ }
+
+ test("AutoCDC API: multi-part `column_list` entry is rejected at flow
registration") {
+ val ex = intercept[RuntimeException] {
+ 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",
+ | column_list = ["nested.field"],
+ |)
+ |""".stripMargin)
+ }
+ assert(ex.getMessage.contains("AUTOCDC_MULTIPART_COLUMN_IDENTIFIER"))
+ }
+
+ test("AutoCDC API: Column-object form of keys/sequence_by/apply_as_deletes
is honored") {
+ val flow = buildAutoCdcFlow(
+ """
+ |from pyspark.sql.functions import col, expr
+ |
+ |@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 = [col("value")],
+ | sequence_by = col("timestamp"),
+ | apply_as_deletes = expr("value % 2 = 0"),
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.keys == Seq(UnqualifiedColumnName("value")))
+ assert(flow.changeArgs.sequencing.expr.sql == "timestamp")
+ val deleteCondition = flow.changeArgs.deleteCondition.getOrElse(
+ fail("expected apply_as_deletes to populate deleteCondition"))
+ assert(deleteCondition.expr.sql.contains("value"))
+ assert(deleteCondition.expr.sql.contains("0"))
+ }
+
+ test("AutoCDC API: graph resolves with the source streaming table as the
flow's input") {
+ val graph = 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",
+ |)
+ |""".stripMargin).resolve()
+
+ val resolvedFlow = graph.resolvedFlow(graphIdentifier("target"))
+ assert(resolvedFlow.inputs == Set(graphIdentifier("src")))
+ }
+
+ test("AutoCDC API: single-part `source` inherits the pipeline's default
catalog/database") {
Review Comment:
This test and `"AutoCDC API: graph resolves with the source streaming table
as the flow's input"` 20 lines above are effectively asserting the same thing.
`graphIdentifier(...)` already returns the fully-qualified
`TableIdentifier("...", Some("default"), Some("spark_catalog"))`, so the prior
test (which builds with `default_catalog=None, default_database=None`) is also
implicitly asserting that the resolved input is qualified to
`spark_catalog.default.src` — it just falls through to session defaults that
happen to match the pipeline-level defaults this test passes explicitly.
Two ways to differentiate:
1. **Drop one.** The end-to-end test below already covers
resolution-with-defaults.
2. **Make this one actually test pipeline-level overrides:** pass
`defaultCatalog = Some("my_catalog")`, `defaultDatabase = Some("my_db")`
(values that do *not* match the session defaults) and assert
`resolvedFlow.inputs == Set(TableIdentifier("src", Some("my_db"),
Some("my_catalog")))`. That would actually exercise the
`QueryContext(Option(defaultCatalog), Option(defaultDatabase))` path threaded
through `buildAutoCdcFlow`, which is otherwise untested and would have caught a
regression if pipeline-level defaults ever stopped being propagated.
Option 2 is the more useful follow-up. Non-blocking either way.
Generated-by: Claude-Opus-4.7
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
