You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "AnishMahto (via GitHub)" <gi...@apache.org> on 2026/05/26 17:39:39 UTC
[PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
AnishMahto opened a new pull request, #56124:
URL: https://github.com/apache/spark/pull/56124
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
In the `PipelinesHandler`, register an `AutoCdcFlow` when a `DefineFlow` proto is received with `AUTO_CDC_FLOW_DETAILS`.
This is the final step in connecting a spark connect client to the spark connect server's SDP engine for AutoCDC flows. With these changes, a user should be able to run their SDP with AutoCDC flows using the pipelines CLI runner.
### Why are the changes needed?
Allows spark connect clients to actually register and execute AutoCDC flows within their pipelines.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Test graph registration and construction through Python client in `PythonPipelineSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored.
Generated-by: Claude-Opus-4.7-thinking-xhigh
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3308412372
##########
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:
I'm intentionally choosing not to do this. Of course we _could_ have a test for every possible branch, but tests come at a cost - our CI gets slower.
I think this one of those cases where its fair to trust the compiler here and not worth adding a test for; scala requires matches to be exhaustive , so by construction the last uncovered case (which is not doing anything interesting anyway, just throws an exception) will be hit appropriately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on PR #56124:
URL: https://github.com/apache/spark/pull/56124#issuecomment-4557083825
The `pyspark-pandas` CI failure _should_ be unrelated, these changes do not touch that module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307734768
##########
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:
Another great catch, I missed handling non attribute expressions, which should be rejected here (we strictly expect expressions that are expressing single part identifiers). Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3308396505
##########
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:
IMO these are not syntax errors but rather semantic errors; the user didn't supply a incorrect value type, but rather an invalid value that is still syntactically correct. Chose sqlState `22023` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3308398109
##########
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:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #56124:
URL: https://github.com/apache/spark/pull/56124#issuecomment-4557242267
Thanks, merging to master/4.x/4.2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3308416436
##########
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:
Went with option 2 and explicitly pass default catalog/database that doesn't match session defaults.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang closed pull request #56124: [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler`
URL: https://github.com/apache/spark/pull/56124
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307737719
##########
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:
No longer applicable after fix of source parsing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3309251324
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala:
##########
@@ -935,6 +942,364 @@ 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: 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 database") {
Review Comment:
After the rename in `ddf76ec98ad`, this test exercises only the *database* half of pipeline-default inheritance: `defaultCatalog = Some("spark_catalog")` is now equal to the session's default catalog (also `spark_catalog`), so the catalog component of the asserted identifier would resolve the same way even without the `defaultCatalog` parameter being plumbed through `buildAutoCdcFlow`. A regression that severed pipeline-catalog propagation would not be caught here.
Two options:
(a) acknowledge in the inline comment that only database inheritance is exercised (and rely on `multi-part \`source\` resolves...` below for catalog routing), or
(b) register a fresh catalog up front, e.g.
```scala
spark.conf.set("spark.sql.catalog.my_catalog", classOf[InMemoryCatalog].getName)
```
then revert `defaultCatalog` to `Some("my_catalog")` so the assertion meaningfully exercises catalog inheritance.
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,119 @@ 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.
+
+ if (!autoCdcDetails.hasSource) {
+ throw new AnalysisException("AUTOCDC_MISSING_SOURCE", Map.empty)
+ }
+ if (!autoCdcDetails.hasSequenceBy) {
+ throw new AnalysisException("AUTOCDC_MISSING_SEQUENCE_BY", Map.empty)
+ }
+
+ val sourcePlan: LogicalPlan = UnresolvedRelation(
+ multipartIdentifier = GraphIdentifierManager
+ .parseTableIdentifier(name = autoCdcDetails.getSource, spark = sessionHolder.session)
+ .nameParts,
+ isStreaming = true)
+
+ val toColumn: proto.Expression => Column = expr => Column(transformExpressionFunc(expr))
+
+ val asUnqualifiedColumnName: proto.Expression => UnqualifiedColumnName = expr =>
+ transformExpressionFunc(expr) match {
+ case a: UnresolvedAttribute => UnqualifiedColumnName(a.nameParts)
+ case other =>
+ throw new AnalysisException(
+ "AUTOCDC_NON_COLUMN_IDENTIFIER",
+ Map("expression" -> other.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) {
+ throw new AnalysisException("AUTOCDC_BOTH_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
+ }
+ }
+
+ // Get user specified SCD type, or default to SCD1 if unspecified.
+ 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")
+ }
+
+ val changeArgs = ChangeArgs(
+ keys = keys,
+ sequencing = toColumn(autoCdcDetails.getSequenceBy),
+ storedAsScdType = scdType,
+ deleteCondition =
+ Option.when(autoCdcDetails.hasApplyAsDeletes)(toColumn(autoCdcDetails.getApplyAsDeletes)),
+ columnSelection = columnSelection)
+
+ AutoCdcFlow(
+ identifier = flowIdentifier,
+ destinationIdentifier = destinationIdentifier,
+ func = FlowAnalysis.createFlowFunctionFromLogicalPlan(sourcePlan),
+ sqlConf = flow.getSqlConfMap.asScala.toMap,
+ queryContext = QueryContext(Option(defaultCatalog), Option(defaultDatabase)),
+ origin = QueryOrigin(
Review Comment:
The `QueryOrigin(...)` block here is identical to the one in the `RELATION_FLOW_DETAILS` arm at line 394 — same `filePath`, `line`, `objectType`, `objectName`, `language`. Now that two flow-construction paths in this file share the exact same origin shape, a small private helper would keep the third path (whenever one is added) from drifting:
```scala
private def flowOrigin(
flow: proto.PipelineCommand.DefineFlow,
flowIdentifier: TableIdentifier): QueryOrigin =
QueryOrigin(
filePath = Option.when(flow.getSourceCodeLocation.hasFileName)(
flow.getSourceCodeLocation.getFileName),
line = Option.when(flow.getSourceCodeLocation.hasLineNumber)(
flow.getSourceCodeLocation.getLineNumber),
objectType = Some(QueryOriginType.Flow.toString),
objectName = Option(flowIdentifier.unquotedString),
language = Some(Python()))
```
Non-blocking; feel free to fold in or defer.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307744746
##########
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:
Added:
1. `AutoCDC API: multi-part `source` resolves to the corresponding qualified dataset`
2. `AutoCDC API: specifying both column_list and except_column_list is rejected`
3. `AutoCDC API: registered flow survives graph resolution and validation end-to-end`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307403195
##########
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
Review Comment:
The "The Python API enforces the 'at most one' contract" claim isn't quite right: `python/pyspark/pipelines/api.py:create_auto_cdc_flow` documents the rule in its docstring but doesn't validate it — both `column_list` and `except_column_list` flow unchanged through `SparkConnectGraphElementRegistry.register_auto_cdc_flow` into the proto, and the downstream `ColumnSelection` model is one-or-the-other by construction so it never sees both. This server-side throw is therefore the only place in the full stack where the rule is actually enforced, not a defensive backstop.
Worth rewording — and, if you take @szehon-ho's adjacent suggestion to use `AnalysisException` with a structured error class, dropping the misleading framing here as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307732017
##########
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:
Good catch. Multipart `source` _should_ be supported, this was a miss on my part. Fixed the parsing so that it actually works for dotted identifiers, and added `AutoCDC API: multi-part `source` resolves to the corresponding qualified dataset` test to `PythonPipelineSuite`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3307747514
##########
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
Review Comment:
Dropped misleading comment and accepted the `AnalysisException` suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on PR #56124:
URL: https://github.com/apache/spark/pull/56124#issuecomment-4547037399
@szehon-ho @gengliangwang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57080][SDP] Register AutoCDC Flows from `PipelinesHandler` [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #56124:
URL: https://github.com/apache/spark/pull/56124#discussion_r3306757040
##########
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(
Review Comment:
nit (style): would say just remove the assignment key, value as there's nothing interesting
##########
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,
Review Comment:
nit (style): can we just directly import these to shorten it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org