szehon-ho commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3306785016
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -388,13 +400,107 @@ 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.")
+ val autoCdcDetails = flow.getAutoCdcFlowDetails
+ graphElementRegistry.registerFlow(
+ buildAutoCdcFlow(
+ autoCdcDetails = autoCdcDetails,
+ flow = flow,
+ flowIdentifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ defaultCatalog = defaultCatalog,
+ defaultDatabase = defaultDatabase,
+ 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: proto.PipelineCommand.DefineFlow.AutoCdcFlowDetails,
+ flow: proto.PipelineCommand.DefineFlow,
+ flowIdentifier: TableIdentifier,
+ destinationIdentifier: TableIdentifier,
+ defaultCatalog: String,
+ defaultDatabase: String,
+ transformExpressionFunc: proto.Expression => Expression): AutoCdcFlow = {
+ val sourcePlan: LogicalPlan = UnresolvedRelation(
+ multipartIdentifier =
scala.collection.immutable.Seq(autoCdcDetails.getSource),
+ isStreaming = true
+ )
Review Comment:
This wraps the user-supplied `source` string as a single-segment
`multipartIdentifier`. For a multi-part input (e.g. `"cat.db.tbl"`) the
resulting `UnresolvedRelation` is `Seq("cat.db.tbl")`, which `FlowAnalysis`
then funnels through `IdentifierHelper.toQuotedString` -> `` `cat.db.tbl` ``
and re-parses as a single-part identifier with literal dots -- so the lookup
will never resolve. Nothing in the proto, the Python `create_auto_cdc_flow`
API, or this method documents that `source` must be single-part, so this is a
silent footgun.
Suggest restricting to single-part and rejecting multi-part up front,
consistent with how `MULTIPART_FLOW_NAME_NOT_SUPPORTED` /
`MULTIPART_SINK_NAME_NOT_SUPPORTED` / `AUTOCDC_MULTIPART_COLUMN_IDENTIFIER` are
handled elsewhere:
```scala
val sourceIdent = GraphIdentifierManager.parseTableIdentifier(
name = autoCdcDetails.getSource,
spark = sessionHolder.session)
if (!IdentifierHelper.isSinglePartIdentifier(sourceIdent)) {
throw new AnalysisException(
"MULTIPART_AUTOCDC_SOURCE_NOT_SUPPORTED",
Map("source" -> autoCdcDetails.getSource))
}
val sourcePlan: LogicalPlan = UnresolvedRelation(
multipartIdentifier = sourceIdent.nameParts,
isStreaming = true)
```
Please also:
- Add the new error condition to
`common/utils/src/main/resources/error/error-conditions.json` (mirror
`MULTIPART_FLOW_NAME_NOT_SUPPORTED`).
- Update the proto comment for `AutoCdcFlowDetails.source` to: *"The name of
the CDC source to stream from. Must be a single-part name."*
- Update the Python `create_auto_cdc_flow` docstring (`:param source:`) to
state the same.
- Validate single-part in `create_auto_cdc_flow` so the failure is raised
client-side, mirroring the existing column-identifier check.
- Add a negative test in `PythonPipelineSuite` (and ideally
`test_graph_element_registry.py`) for a multipart source.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -388,13 +400,107 @@ 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.")
+ val autoCdcDetails = flow.getAutoCdcFlowDetails
+ graphElementRegistry.registerFlow(
+ buildAutoCdcFlow(
+ autoCdcDetails = autoCdcDetails,
+ flow = flow,
+ flowIdentifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ defaultCatalog = defaultCatalog,
+ defaultDatabase = defaultDatabase,
+ 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: proto.PipelineCommand.DefineFlow.AutoCdcFlowDetails,
+ flow: proto.PipelineCommand.DefineFlow,
+ flowIdentifier: TableIdentifier,
+ destinationIdentifier: TableIdentifier,
+ defaultCatalog: String,
+ defaultDatabase: String,
+ transformExpressionFunc: proto.Expression => Expression): AutoCdcFlow = {
+ val sourcePlan: LogicalPlan = UnresolvedRelation(
+ multipartIdentifier =
scala.collection.immutable.Seq(autoCdcDetails.getSource),
+ isStreaming = true
+ )
+
+ val toColumn: proto.Expression => Column = expr =>
Column(transformExpressionFunc(expr))
+
+ val asUnqualifiedColumnName: proto.Expression => UnqualifiedColumnName =
+ expr => UnqualifiedColumnName(transformExpressionFunc(expr).sql)
Review Comment:
Going `Expression -> .sql -> string` works for the common cases (`"col"`,
`col("col")`), but a non-attribute expression (`expr("a + 1")`, function call,
literal) silently produces a column name of `"(a + 1)"`, which the downstream
multipart check (`AUTOCDC_MULTIPART_COLUMN_IDENTIFIER`) won't catch and which
will fail later with a less actionable error.
Consider validating that the transformed expression is an
`UnresolvedAttribute` / `Attribute` and extracting its name parts directly,
e.g.:
```scala
val asUnqualifiedColumnName: proto.Expression => UnqualifiedColumnName =
expr =>
transformExpressionFunc(expr) match {
case a: UnresolvedAttribute if a.nameParts.size == 1 =>
UnqualifiedColumnName(a.nameParts.head)
case a: UnresolvedAttribute =>
throw new AnalysisException(
"AUTOCDC_MULTIPART_COLUMN_IDENTIFIER",
Map("columnName" -> a.name, "nameParts" -> a.nameParts.mkString(",
")))
case other =>
throw new AnalysisException(
"AUTOCDC_NON_COLUMN_IDENTIFIER",
Map("expression" -> other.sql))
}
```
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala:
##########
@@ -935,6 +937,286 @@ 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(
+ throw new AssertionError(
+ 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(
+ throw new AssertionError("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: omitted stored_as_scd_type defaults to 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",
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ }
Review Comment:
This test largely duplicates `"AutoCDC API: minimal flow registers an
AutoCdcFlow with default name and SCD1 default"` above. Suggest removing it, or
differentiating by asserting on the explicit `SCD_TYPE_UNSPECIFIED` proto value
(e.g. via a direct proto-level test), so the two cover different surfaces.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -388,13 +400,107 @@ 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.")
+ val autoCdcDetails = flow.getAutoCdcFlowDetails
+ graphElementRegistry.registerFlow(
+ buildAutoCdcFlow(
+ autoCdcDetails = autoCdcDetails,
+ flow = flow,
+ flowIdentifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ defaultCatalog = defaultCatalog,
+ defaultDatabase = defaultDatabase,
+ 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: proto.PipelineCommand.DefineFlow.AutoCdcFlowDetails,
+ flow: proto.PipelineCommand.DefineFlow,
+ flowIdentifier: TableIdentifier,
+ destinationIdentifier: TableIdentifier,
+ defaultCatalog: String,
+ defaultDatabase: String,
+ transformExpressionFunc: proto.Expression => Expression): AutoCdcFlow = {
+ val sourcePlan: LogicalPlan = UnresolvedRelation(
+ multipartIdentifier =
scala.collection.immutable.Seq(autoCdcDetails.getSource),
+ isStreaming = true
+ )
+
+ val toColumn: proto.Expression => Column = expr =>
Column(transformExpressionFunc(expr))
+
+ val asUnqualifiedColumnName: proto.Expression => UnqualifiedColumnName =
+ expr => UnqualifiedColumnName(transformExpressionFunc(expr).sql)
+
+ val keys =
autoCdcDetails.getKeysList.asScala.toSeq.map(asUnqualifiedColumnName)
+
+ val columnSelection: Option[ColumnSelection] = {
+ val included = autoCdcDetails.getColumnListList.asScala.toSeq
+ val excluded = autoCdcDetails.getExceptColumnListList.asScala.toSeq
+ if (included.nonEmpty && excluded.nonEmpty) {
+ // The Python API enforces the "at most one" contract; we don't expect
both to be set
+ // here, but reject defensively to surface client/server mismatches
loudly.
+ throw new IllegalArgumentException(
+ "AutoCDC flow specifies both column_list and except_column_list; at
most one " +
+ "may be provided.")
Review Comment:
Other validation in `defineFlow` (e.g. `MULTIPART_FLOW_NAME_NOT_SUPPORTED`)
throws `AnalysisException` with a structured error class. Spark Connect clients
translate those into actionable Python errors. Suggest:
```scala
throw new AnalysisException(
"AUTOCDC_BOTH_COLUMN_LIST_AND_EXCEPT_COLUMN_LIST",
Map.empty)
```
and add the corresponding entry to `error-conditions.json`. Python already
rejects this case, so this is purely a defense for non-Python clients -- agree
it should stay, just with consistent error semantics.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -388,13 +400,107 @@ 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.")
+ val autoCdcDetails = flow.getAutoCdcFlowDetails
+ graphElementRegistry.registerFlow(
+ buildAutoCdcFlow(
+ autoCdcDetails = autoCdcDetails,
+ flow = flow,
+ flowIdentifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ defaultCatalog = defaultCatalog,
+ defaultDatabase = defaultDatabase,
+ 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: proto.PipelineCommand.DefineFlow.AutoCdcFlowDetails,
+ flow: proto.PipelineCommand.DefineFlow,
+ flowIdentifier: TableIdentifier,
+ destinationIdentifier: TableIdentifier,
+ defaultCatalog: String,
+ defaultDatabase: String,
+ transformExpressionFunc: proto.Expression => Expression): AutoCdcFlow = {
Review Comment:
The `AutoCdcFlowDetails` proto declares `apply_as_truncates` (line 174 of
`pipelines.proto`), `ignore_null_updates_column_list` (line 186), and
`ignore_null_updates_except_column_list` (line 189) -- none of which
`buildAutoCdcFlow` reads. The Python client doesn't send them today, but a
future client (or someone constructing the proto directly) could, and they'd be
silently dropped.
Suggest adding an explicit TODO near the top of `buildAutoCdcFlow` so the
omission is documented and discoverable:
```scala
// TODO(SPARK-XXXXX): apply_as_truncates, ignore_null_updates_column_list,
and
// ignore_null_updates_except_column_list are declared in the
AutoCdcFlowDetails proto
// but are not yet honored by the engine. Wire them through once supported.
```
File a follow-up JIRA so the TODO has a concrete tracking ticket.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala:
##########
@@ -388,13 +400,107 @@ 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.")
+ val autoCdcDetails = flow.getAutoCdcFlowDetails
+ graphElementRegistry.registerFlow(
+ buildAutoCdcFlow(
+ autoCdcDetails = autoCdcDetails,
+ flow = flow,
+ flowIdentifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ defaultCatalog = defaultCatalog,
+ defaultDatabase = defaultDatabase,
+ 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: proto.PipelineCommand.DefineFlow.AutoCdcFlowDetails,
+ flow: proto.PipelineCommand.DefineFlow,
+ flowIdentifier: TableIdentifier,
+ destinationIdentifier: TableIdentifier,
+ defaultCatalog: String,
+ defaultDatabase: String,
+ transformExpressionFunc: proto.Expression => Expression): AutoCdcFlow = {
+ val sourcePlan: LogicalPlan = UnresolvedRelation(
+ multipartIdentifier =
scala.collection.immutable.Seq(autoCdcDetails.getSource),
Review Comment:
Nit: the file imports `scala.collection.Seq` at the top; this single
fully-qualified `scala.collection.immutable.Seq(...)` stands out. Just use
`Seq(...)` for consistency.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala:
##########
@@ -935,6 +937,286 @@ 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(
+ throw new AssertionError(
+ 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(
+ throw new AssertionError("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: omitted stored_as_scd_type defaults to 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",
+ |)
+ |""".stripMargin)
+
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ }
+
+ 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(
+ throw new AssertionError("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")))
+ }
Review Comment:
Coverage gaps worth filling:
1. Multipart `source` is rejected (after the restriction in comment #1).
2. Server-side rejection when both `column_list` and `except_column_list`
are populated. This currently requires bypassing Python validation; a small
helper that builds an `AutoCdcFlowDetails` proto directly would do it. Without
this, the new defensive throw in `buildAutoCdcFlow` is unreachable in tests.
3. A single `buildGraph(...).resolve()` (or dry-run) test that exercises a
registered AutoCDC flow end-to-end through Connect, beyond just graph-input
resolution. This validates the PR description's claim about the pipelines CLI
runner.
--
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]
