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: [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]
