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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
