This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new af2b7fa3fc13 [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with
the dataflow graph
af2b7fa3fc13 is described below
commit af2b7fa3fc13923d237eb2b9d65d9cf1fe3e1df8
Author: Andreas Neumann <[email protected]>
AuthorDate: Tue Jul 14 14:29:11 2026 -0700
[SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph
### What changes were proposed in this pull request? Hook up the two AUTO
CDC SQL constructs introduced in SPARK-56249 to the Spark Declarative Pipelines
dataflow graph in `SqlGraphRegistrationContext`:
1. `CREATE STREAMING TABLE <name> FLOW AUTO CDC FROM <source> ...`, parsed
into `CreateStreamingTableAutoCdc`, now registers the streaming table and an
`AutoCdcFlow` that targets it.
2. `CREATE FLOW <name> AS AUTO CDC INTO <target> FROM <source> ...`, parsed
into a `CreateFlowCommand` wrapping an `AutoCdcIntoCommand`, now registers an
`AutoCdcFlow` from the named flow into the target dataset.
A shared `buildChangeArgs` helper converts the parse-time expressions and
unresolved attributes into the `ChangeArgs` consumed by `AutoCdcFlow` (keys,
sequencing, delete condition, include/exclude column selection). SQL AUTO CDC
only supports SCD Type 1, matching the Connect/proto path.
### Why are the changes needed?
Before this change the AUTO CDC SQL syntax parsed successfully but was
rejected during graph registration, so it could not be used to define pipeline
datasets or flows.
### Does this PR introduce _any_ user-facing change?
Yes. AUTO CDC SQL statements now register datasets and flows in a pipeline.
### How was this patch tested?
Added registration and end-to-end execution tests to `SqlPipelineSuite`
covering both syntaxes, the optional clauses (APPLY AS DELETE WHEN, COLUMNS,
COLUMNS * EXCEPT), and the multipart-flow-name error. Full `SqlPipelineSuite`
passes (50 tests).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 4.8
Closes #57176 from anew/hookup-sql-parser-with-dataflow-graph.
Authored-by: Andreas Neumann <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit c1fb0aa0bcd979e6b8f11b898c0882c0155e794a)
Signed-off-by: Szehon Ho <[email protected]>
---
.../spark/sql/catalyst/parser/AstBuilder.scala | 4 +-
...{AutoCdcIntoCommand.scala => AutoCdcInto.scala} | 29 ++-
.../sql/catalyst/plans/logical/v2Commands.scala | 8 +-
.../spark/sql/execution/SparkSqlParser.scala | 2 +-
.../spark/sql/execution/SparkStrategies.scala | 3 +-
.../spark/sql/execution/command/DDLSuite.scala | 48 ++++
.../execution/command/v2/AutoCdcParserSuite.scala | 24 +-
.../graph/SqlGraphRegistrationContext.scala | 272 ++++++++++++++++----
.../sql/pipelines/graph/SqlPipelineSuite.scala | 275 ++++++++++++++++++++-
9 files changed, 582 insertions(+), 83 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index 0b28afb281c8..13305ec1b8b7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -1369,11 +1369,11 @@ class AstBuilder extends DataTypeAstBuilder
withSchemaEvolution)
}
- protected def buildAutoCdcIntoCommand(ctx: AutoCdcCommandContext):
AutoCdcIntoCommand =
+ protected def buildAutoCdcInto(ctx: AutoCdcCommandContext): AutoCdcInto =
withOrigin(ctx) {
val target = UnresolvedIdentifier(visitMultipartIdentifier(ctx.target))
val params = parseAutoCdcParams(ctx.autoCdcParameters())
- AutoCdcIntoCommand(
+ AutoCdcInto(
targetTable = target,
source = params.source,
keys = params.keys,
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/AutoCdcIntoCommand.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/AutoCdcInto.scala
similarity index 66%
rename from
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/AutoCdcIntoCommand.scala
rename to
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/AutoCdcInto.scala
index eacdf51927a6..cfea98579c63 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/AutoCdcIntoCommand.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/AutoCdcInto.scala
@@ -18,19 +18,24 @@
package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
-import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.trees.BinaryLike
/**
- * Logical plan node for an AUTO CDC INTO command, used by Spark Declarative
Pipelines.
+ * Logical plan node for an AUTO CDC INTO operation, used by Spark Declarative
Pipelines.
*
* This represents a CDC (Change Data Capture) operation that applies an
ordered change event
* stream from [[source]] into [[targetTable]] using SCD Type 1 (upsert)
semantics.
*
- * This node serves as a parse-time placeholder for a pipeline CDC definition
and cannot be
- * executed directly. It will be interpreted by the pipeline submodule once
execution support
- * is added (SPARK-57402). The [[targetTable]] and [[source]] relations are
exposed as the node's
- * children (left and right respectively) so the analyzer resolves them
through the normal plan
- * resolution path.
+ * This node only ever appears as the flow operation of a
[[CreateFlowCommand]] (parsed from
+ * `CREATE FLOW ... AS AUTO CDC INTO ...`); there is no standalone `AUTO CDC
INTO` syntax. It is a
+ * plain [[LogicalPlan]] rather than a [[Command]] so that it is never eagerly
executed or planned
+ * on its own -- it is interpreted by the pipeline submodule during a pipeline
execution. Unlike a
+ * [[ParsedStatement]], it is not rewritten into another plan during analysis;
the pipeline
+ * submodule consumes it as-is, so it relies on the default resolution
semantics (resolved once its
+ * children and expressions are resolved) rather than being forced unresolved.
The [[targetTable]]
+ * and [[source]] relations are exposed as the node's children (left and right
respectively) so the
+ * analyzer resolves them through the normal plan resolution path.
*
* @param targetTable The target table to apply changes into, as an
`UnresolvedIdentifier`.
* Exposed as the node's left child.
@@ -49,7 +54,7 @@ import org.apache.spark.sql.catalyst.expressions.Expression
* except these). [[None]] when no COLUMNS clause was
specified. Mutually
* exclusive with [[includeColumns]].
*/
-case class AutoCdcIntoCommand(
+case class AutoCdcInto(
targetTable: LogicalPlan,
source: LogicalPlan,
keys: Seq[UnresolvedAttribute],
@@ -57,11 +62,15 @@ case class AutoCdcIntoCommand(
sequenceByExpr: Expression,
includeColumns: Option[Seq[UnresolvedAttribute]],
excludeColumns: Option[Seq[UnresolvedAttribute]]
-) extends BinaryCommand {
+) extends LogicalPlan with BinaryLike[LogicalPlan] {
override def left: LogicalPlan = targetTable
override def right: LogicalPlan = source
+ // This node is a parse-time placeholder that is never executed or projected
from directly; the
+ // pipeline submodule reads its fields instead. It therefore produces no
output columns.
+ override def output: Seq[Attribute] = Nil
+
override protected def withNewChildrenInternal(
- newLeft: LogicalPlan, newRight: LogicalPlan): AutoCdcIntoCommand =
+ newLeft: LogicalPlan, newRight: LogicalPlan): AutoCdcInto =
copy(targetTable = newLeft, source = newRight)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
index eed6fd7ca6ca..468b43d4915f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
@@ -803,7 +803,7 @@ case class CreateStreamingTableAsSelect(
/**
* Command parsed from `CREATE STREAMING TABLE ...` SQL syntax. This command
serves as a logical
- * representation of the matching SQL syntac and cannot be executed. It is
instead interpreted by
+ * representation of the matching SQL syntax and cannot be executed. It is
instead interpreted by
* the pipeline submodule during a pipeline execution.
*
* Differs from [[CreateStreamingTableAsSelect]] in that the AS [subquery]
clause is not provided
@@ -826,9 +826,9 @@ case class CreateStreamingTable(
/**
* Command parsed from `CREATE STREAMING TABLE <name> FLOW AUTO CDC ...` SQL
syntax.
- * This command serves as a parse-time placeholder for a pipeline CDC
definition and cannot be
- * executed directly. It will be interpreted by the pipeline submodule once
execution support
- * is added (SPARK-57402).
+ * This command serves as a parse-time placeholder for a pipeline CDC
definition and cannot
+ * be executed directly. It is instead interpreted by the pipeline submodule
during a pipeline
+ * execution.
*
* The target of the CDC operation is the streaming table itself (given by
[[name]]).
*
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 0f689c1705bf..39c6dda401d9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -1633,7 +1633,7 @@ class SparkSqlAstBuilder extends AstBuilder {
val flowHeaderCtx = ctx.createPipelineFlowHeader()
val ident = withIdentClause(flowHeaderCtx.flowName,
UnresolvedIdentifier(_))
val commentOpt = Option(flowHeaderCtx.commentSpec()).map(visitCommentSpec)
- val autoCdcInto = buildAutoCdcIntoCommand(ctx.autoCdcCommand())
+ val autoCdcInto = buildAutoCdcInto(ctx.autoCdcCommand())
CreateFlowCommand(
name = ident,
flowOperation = autoCdcInto,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index d89f7a919269..e8d29dfa7c5e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -1001,7 +1001,8 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
case cmv: CreateMaterializedViewAsSelect =>
throw
QueryCompilationErrors.unsupportedCreatePipelineDatasetQueryExecutionError(
pipelineDatasetType = "MATERIALIZED VIEW")
- case cst: CreateStreamingTableAsSelect =>
+ case _: CreateStreamingTableAsSelect | _: CreateStreamingTable |
+ _: CreateStreamingTableAutoCdc =>
throw
QueryCompilationErrors.unsupportedCreatePipelineDatasetQueryExecutionError(
pipelineDatasetType = "STREAMING TABLE")
case _ => Nil
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 92e9dc1a2e59..148f4de20a24 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -2541,6 +2541,36 @@ abstract class DDLSuite extends QueryTest with
DDLSuiteBase {
)
}
+ test("CREATE STREAMING TABLE without subquery cannot be directly executed") {
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ sql("CREATE STREAMING TABLE table1")
+ },
+ condition =
"UNSUPPORTED_FEATURE.CREATE_PIPELINE_DATASET_QUERY_EXECUTION",
+ sqlState = "0A000",
+ parameters = Map("pipelineDatasetType" -> "STREAMING TABLE")
+ )
+ }
+
+ test("CREATE STREAMING TABLE FLOW AUTO CDC cannot be directly executed") {
+ withTable("cdc_src") {
+ sql("CREATE TABLE cdc_src AS SELECT 1 AS id, 1 AS ts")
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ sql(
+ """CREATE STREAMING TABLE table1
+ |FLOW AUTO CDC
+ |FROM STREAM(cdc_src)
+ |KEYS (id)
+ |SEQUENCE BY ts""".stripMargin)
+ },
+ condition =
"UNSUPPORTED_FEATURE.CREATE_PIPELINE_DATASET_QUERY_EXECUTION",
+ sqlState = "0A000",
+ parameters = Map("pipelineDatasetType" -> "STREAMING TABLE")
+ )
+ }
+ }
+
test(s"CREATE FLOW statement cannot be directly executed") {
sql("CREATE TABLE table1 AS SELECT 1")
sql("CREATE TABLE table2 AS SELECT 2")
@@ -2553,6 +2583,24 @@ abstract class DDLSuite extends QueryTest with
DDLSuiteBase {
parameters = Map.empty
)
}
+
+ test("CREATE FLOW AS AUTO CDC cannot be directly executed") {
+ withTable("cdc_src") {
+ sql("CREATE TABLE cdc_src AS SELECT 1 AS id, 1 AS ts")
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ sql(
+ """CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM(cdc_src)
+ |KEYS (id)
+ |SEQUENCE BY ts""".stripMargin)
+ },
+ condition = "UNSUPPORTED_FEATURE.CREATE_FLOW_QUERY_EXECUTION",
+ sqlState = "0A000",
+ parameters = Map.empty
+ )
+ }
+ }
}
object FakeLocalFsFileSystem {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AutoCdcParserSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AutoCdcParserSuite.scala
index 975ea7fb1559..c9d28edaf6a6 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AutoCdcParserSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AutoCdcParserSuite.scala
@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.analysis.{
UnresolvedRelation}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.logical.{
- AutoCdcIntoCommand,
+ AutoCdcInto,
CreateFlowCommand,
CreateStreamingTableAutoCdc,
LogicalPlan,
@@ -71,7 +71,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
assert(cmd.name.asInstanceOf[UnresolvedIdentifier].nameParts ==
Seq("myflow"))
assert(cmd.comment.isEmpty)
- val cdc = cmd.flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc = cmd.flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.targetTable.asInstanceOf[UnresolvedIdentifier].nameParts ==
Seq("target"))
val source = streamSource(cdc.source)
assert(source.multipartIdentifier == Seq("source"))
@@ -89,7 +89,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|KEYS (id)
|SEQUENCE BY ts""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
val source = streamSource(cdc.source)
assert(source.multipartIdentifier == Seq("mycat", "myschema", "source"))
}
@@ -124,7 +124,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|KEYS (k)
|SEQUENCE BY ts""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.targetTable.asInstanceOf[UnresolvedIdentifier].nameParts ==
Seq("myschema", "mytable"))
}
@@ -136,7 +136,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|KEYS (k)
|SEQUENCE BY ts""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.targetTable.asInstanceOf[UnresolvedIdentifier].nameParts ==
Seq("mycat", "myschema", "mytable"))
}
@@ -149,7 +149,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|APPLY AS DELETE WHEN op = 'DELETE'
|SEQUENCE BY ts""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.deleteCondition.isDefined)
assert(cdc.deleteCondition.get.sql.contains("op"))
}
@@ -162,7 +162,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|SEQUENCE BY ts
|COLUMNS (id, name, value)""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.includeColumns.get.map(_.name) == Seq("id", "name", "value"))
assert(cdc.excludeColumns.isEmpty)
}
@@ -175,7 +175,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|SEQUENCE BY ts
|COLUMNS * EXCEPT (op, ts)""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.includeColumns.isEmpty)
assert(cdc.excludeColumns.get.map(_.name) == Seq("op", "ts"))
}
@@ -189,7 +189,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|SEQUENCE BY timestamp
|COLUMNS (key1, key2, key3, timestamp)""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.keys.map(_.name) == Seq("key1", "key2"))
assert(cdc.deleteCondition.isDefined)
assert(cdc.sequenceByExpr == UnresolvedAttribute("timestamp"))
@@ -523,7 +523,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|KEYS (id)
|SEQUENCE BY ts""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
val source = streamSource(cdc.source)
assert(source.multipartIdentifier == Seq("source"))
}
@@ -535,7 +535,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
|KEYS (id)
|SEQUENCE BY ts""".stripMargin)
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.source.isStreaming)
val alias = cdc.source.asInstanceOf[SubqueryAlias]
assert(alias.alias == "s")
@@ -550,7 +550,7 @@ class AutoCdcParserSuite extends CommandSuiteBase with
AnalysisTest {
// The subquery wraps the STREAM read in Project/Filter/SubqueryAlias
nodes; isStreaming
// propagates up through them, so the whole source is recognized as
streaming.
- val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcIntoCommand]
+ val cdc =
plan.asInstanceOf[CreateFlowCommand].flowOperation.asInstanceOf[AutoCdcInto]
assert(cdc.source.isStreaming)
assert(cdc.source.isInstanceOf[SubqueryAlias])
}
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
index 625844de74cb..36a1f17209a2 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
@@ -20,12 +20,16 @@ import scala.collection.mutable
import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, SparkSession}
-import org.apache.spark.sql.catalyst.QueryPlanningTracker
-import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
-import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand,
CreateMaterializedViewAsSelect, CreateStreamingTable,
CreateStreamingTableAsSelect, CreateView, InsertIntoStatement, LogicalPlan}
+import org.apache.spark.sql.Column
+import org.apache.spark.sql.catalyst.{QueryPlanningTracker, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute,
UnresolvedRelation}
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.plans.logical.{AutoCdcInto,
CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable,
CreateStreamingTableAsSelect, CreateStreamingTableAutoCdc, CreateView,
InsertIntoStatement, LogicalPlan}
import org.apache.spark.sql.catalyst.util.StringUtils
+import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.execution.command.{CreateViewCommand,
SetCatalogCommand, SetCommand, SetNamespaceCommand}
import org.apache.spark.sql.pipelines.Language
+import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ColumnSelection,
ScdType, UnqualifiedColumnName}
import org.apache.spark.sql.types.StructType
/**
@@ -162,6 +166,10 @@ class SqlGraphRegistrationContext(
case createStreamingTableCommand: CreateStreamingTable =>
// CREATE STREAMING TABLE [ streaming_table_name ] [ options ]
CreateStreamingTableHandler.handle(createStreamingTableCommand,
queryOrigin)
+ case createStreamingTableAutoCdcCommand: CreateStreamingTableAutoCdc =>
+ // CREATE STREAMING TABLE [ streaming_table_name ] [ options ]
+ // FLOW AUTO CDC FROM [ source ] KEYS ( ... ) SEQUENCE BY [ expr ]
...
+
CreateStreamingTableAutoCdcHandler.handle(createStreamingTableAutoCdcCommand,
queryOrigin)
case createFlowCommand: CreateFlowCommand =>
// CREATE FLOW [ flow_name ] AS INSERT INTO [ destination_name ] BY
NAME
CreateFlowHandler.handle(createFlowCommand, queryOrigin)
@@ -183,6 +191,9 @@ class SqlGraphRegistrationContext(
)
.identifier
+ val (partitionCols, clusterCols) =
+ PartitionHelper.splitPartitionAndClusterColumns(cst.partitioning,
queryOrigin)
+
// Register streaming table as a table.
graphRegistrationContext.registerTable(
Table(
@@ -190,8 +201,8 @@ class SqlGraphRegistrationContext(
comment = cst.tableSpec.comment,
specifiedSchema =
Option.when(cst.columns.nonEmpty)(StructType(cst.columns.map(_.toV1Column))),
- partitionCols =
Option(PartitionHelper.applyPartitioning(cst.partitioning, queryOrigin)),
- clusterCols = None,
+ partitionCols = Option(partitionCols),
+ clusterCols = Option.when(clusterCols.nonEmpty)(clusterCols),
properties = cst.tableSpec.properties,
origin = queryOrigin.copy(
objectName = Option(stIdentifier.unquotedString),
@@ -215,6 +226,9 @@ class SqlGraphRegistrationContext(
)
.identifier
+ val (partitionCols, clusterCols) =
+ PartitionHelper.splitPartitionAndClusterColumns(cst.partitioning,
queryOrigin)
+
// Register streaming table as a table.
graphRegistrationContext.registerTable(
Table(
@@ -222,8 +236,8 @@ class SqlGraphRegistrationContext(
comment = cst.tableSpec.comment,
specifiedSchema =
Option.when(cst.columns.nonEmpty)(StructType(cst.columns.map(_.toV1Column))),
- partitionCols =
Option(PartitionHelper.applyPartitioning(cst.partitioning, queryOrigin)),
- clusterCols = None,
+ partitionCols = Option(partitionCols),
+ clusterCols = Option.when(clusterCols.nonEmpty)(clusterCols),
properties = cst.tableSpec.properties,
origin = queryOrigin.copy(
objectName = Option(stIdentifier.unquotedString),
@@ -256,6 +270,117 @@ class SqlGraphRegistrationContext(
}
}
+ /**
+ * Converts the parse-time AUTO CDC parameters (catalyst expressions and
unresolved attributes)
+ * into the [[ChangeArgs]] consumed by an [[AutoCdcFlow]]. Shared by the two
SQL AUTO CDC entry
+ * points: `CREATE STREAMING TABLE ... FLOW AUTO CDC ...` and `CREATE FLOW
... AS AUTO CDC INTO`.
+ *
+ * SQL AUTO CDC syntax only supports SCD Type 1, so
[[ChangeArgs.storedAsScdType]] is always
+ * [[ScdType.Type1]]. [[includeColumns]] and [[excludeColumns]] are mutually
exclusive at the
+ * grammar level; the guard here is defensive.
+ */
+ private def buildChangeArgs(
+ keys: Seq[UnresolvedAttribute],
+ sequenceByExpr: Expression,
+ deleteCondition: Option[Expression],
+ includeColumns: Option[Seq[UnresolvedAttribute]],
+ excludeColumns: Option[Seq[UnresolvedAttribute]],
+ queryOrigin: QueryOrigin): ChangeArgs = {
+ val columnSelection: Option[ColumnSelection] = (includeColumns,
excludeColumns) match {
+ case (Some(_), Some(_)) =>
+ throw SqlGraphElementRegistrationException(
+ msg = "AUTO CDC cannot specify both COLUMNS and COLUMNS * EXCEPT.",
+ queryOrigin = queryOrigin
+ )
+ case (Some(included), None) =>
+
Option(ColumnSelection.IncludeColumns(included.map(toUnqualifiedColumnName)))
+ case (None, Some(excluded)) =>
+
Option(ColumnSelection.ExcludeColumns(excluded.map(toUnqualifiedColumnName)))
+ case (None, None) =>
+ None
+ }
+
+ ChangeArgs(
+ keys = keys.map(toUnqualifiedColumnName),
+ sequencing = Column(sequenceByExpr),
+ storedAsScdType = ScdType.Type1,
+ deleteCondition = deleteCondition.map(Column(_)),
+ columnSelection = columnSelection
+ )
+ }
+
+ private def toUnqualifiedColumnName(attr: UnresolvedAttribute):
UnqualifiedColumnName =
+ UnqualifiedColumnName(attr.nameParts)
+
+ private object CreateStreamingTableAutoCdcHandler {
+ def handle(cst: CreateStreamingTableAutoCdc, queryOrigin: QueryOrigin):
Unit = {
+ val stIdentifier = GraphIdentifierManager
+ .parseAndQualifyTableIdentifier(
+ rawTableIdentifier = IdentifierHelper.toTableIdentifier(cst.name),
+ currentCatalog = context.getCurrentCatalogOpt,
+ currentDatabase = context.getCurrentDatabaseOpt
+ )
+ .identifier
+
+ if (cst.columns.nonEmpty) {
+ throw SqlGraphElementRegistrationException(
+ msg = "Explicit column lists are not supported for AUTO CDC
streaming tables; " +
+ "omit the column list and let schema be inferred from the flow.",
+ queryOrigin = queryOrigin)
+ }
+
+ val (partitionCols, clusterCols) =
+ PartitionHelper.splitPartitionAndClusterColumns(cst.partitioning,
queryOrigin)
+
+ // Register the streaming table as a table. The streaming table is
itself the target of the
+ // CDC operation.
+ graphRegistrationContext.registerTable(
+ Table(
+ identifier = stIdentifier,
+ comment = cst.tableSpec.comment,
+ specifiedSchema = None,
+ partitionCols = Option(partitionCols),
+ clusterCols = Option.when(clusterCols.nonEmpty)(clusterCols),
+ properties = cst.tableSpec.properties,
+ origin = queryOrigin.copy(
+ objectName = Option(stIdentifier.unquotedString),
+ objectType = Option(QueryOriginType.Table.toString)
+ ),
+ format = cst.tableSpec.provider,
+ normalizedPath = None,
+ isStreamingTable = true
+ )
+ )
+
+ // Register the AutoCDC flow that backs this streaming table. Both the
flow and its
+ // destination are the streaming table itself.
+ graphRegistrationContext.registerFlow(
+ AutoCdcFlow(
+ identifier = stIdentifier,
+ destinationIdentifier = stIdentifier,
+ func = FlowAnalysis.createFlowFunctionFromLogicalPlan(cst.source),
+ sqlConf = context.getSqlConf,
+ queryContext = QueryContext(
+ currentCatalog = context.getCurrentCatalogOpt,
+ currentDatabase = context.getCurrentDatabaseOpt
+ ),
+ origin = queryOrigin.copy(
+ objectName = Option(stIdentifier.unquotedString),
+ objectType = Option(QueryOriginType.Flow.toString)
+ ),
+ changeArgs = buildChangeArgs(
+ keys = cst.keys,
+ sequenceByExpr = cst.sequenceByExpr,
+ deleteCondition = cst.deleteCondition,
+ includeColumns = cst.includeColumns,
+ excludeColumns = cst.excludeColumns,
+ queryOrigin = queryOrigin
+ )
+ )
+ )
+ }
+ }
+
private object CreateMaterializedViewAsSelectHandler {
def handle(cmv: CreateMaterializedViewAsSelect, queryOrigin: QueryOrigin):
Unit = {
val mvIdentifier = GraphIdentifierManager
@@ -266,6 +391,9 @@ class SqlGraphRegistrationContext(
)
.identifier
+ val (partitionCols, clusterCols) =
+ PartitionHelper.splitPartitionAndClusterColumns(cmv.partitioning,
queryOrigin)
+
// Register materialized view as a table.
graphRegistrationContext.registerTable(
Table(
@@ -273,8 +401,8 @@ class SqlGraphRegistrationContext(
comment = cmv.tableSpec.comment,
specifiedSchema =
Option.when(cmv.columns.nonEmpty)(StructType(cmv.columns.map(_.toV1Column))),
- partitionCols =
Option(PartitionHelper.applyPartitioning(cmv.partitioning, queryOrigin)),
- clusterCols = None,
+ partitionCols = Option(partitionCols),
+ clusterCols = Option.when(clusterCols.nonEmpty)(clusterCols),
properties = cmv.tableSpec.properties,
origin = queryOrigin.copy(
objectName = Option(mvIdentifier.unquotedString),
@@ -415,7 +543,7 @@ class SqlGraphRegistrationContext(
)
.identifier
- val (flowTargetDatasetIdentifier, flowQueryLogicalPlan) =
cf.flowOperation match {
+ cf.flowOperation match {
case i: InsertIntoStatement =>
validateInsertIntoFlow(i, queryOrigin)
val flowTargetDatasetName = i.table match {
@@ -427,22 +555,55 @@ class SqlGraphRegistrationContext(
queryOrigin = queryOrigin
)
}
- val qualifiedFlowTargetDatasetName = GraphIdentifierManager
- .parseAndQualifyTableIdentifier(
- rawTableIdentifier = flowTargetDatasetName,
- currentCatalog = context.getCurrentCatalogOpt,
- currentDatabase = context.getCurrentDatabaseOpt
+ graphRegistrationContext.registerFlow(
+ UntypedFlow(
+ identifier = flowIdentifier,
+ destinationIdentifier =
qualifyDestinationIdentifier(flowTargetDatasetName),
+ func = FlowAnalysis.createFlowFunctionFromLogicalPlan(i.query),
+ sqlConf = context.getSqlConf,
+ once = false,
+ queryContext = QueryContext(
+ currentCatalog = context.getCurrentCatalogOpt,
+ currentDatabase = context.getCurrentDatabaseOpt
+ ),
+ origin = queryOrigin
)
- .identifier
- (qualifiedFlowTargetDatasetName, i.query)
+ )
+ case a: AutoCdcInto =>
+ val flowTargetDatasetName =
IdentifierHelper.toTableIdentifier(a.targetTable)
+ graphRegistrationContext.registerFlow(
+ AutoCdcFlow(
+ identifier = flowIdentifier,
+ destinationIdentifier =
qualifyDestinationIdentifier(flowTargetDatasetName),
+ func = FlowAnalysis.createFlowFunctionFromLogicalPlan(a.source),
+ sqlConf = context.getSqlConf,
+ queryContext = QueryContext(
+ currentCatalog = context.getCurrentCatalogOpt,
+ currentDatabase = context.getCurrentDatabaseOpt
+ ),
+ origin = queryOrigin,
+ changeArgs = buildChangeArgs(
+ keys = a.keys,
+ sequenceByExpr = a.sequenceByExpr,
+ deleteCondition = a.deleteCondition,
+ includeColumns = a.includeColumns,
+ excludeColumns = a.excludeColumns,
+ queryOrigin = queryOrigin
+ )
+ )
+ )
case _ =>
throw SqlGraphElementRegistrationException(
- msg = "Unable flow type. Only INSERT INTO flows are supported.",
+ msg = "Unknown flow type. Only INSERT INTO and AUTO CDC INTO flows
are supported.",
queryOrigin = queryOrigin
)
}
+ }
- val qualifiedDestinationIdentifier = GraphIdentifierManager
+ /** Qualifies a raw flow target dataset identifier against the current
catalog/database. */
+ private def qualifyDestinationIdentifier(
+ flowTargetDatasetIdentifier: TableIdentifier): TableIdentifier =
+ GraphIdentifierManager
.parseAndQualifyFlowIdentifier(
rawFlowIdentifier = flowTargetDatasetIdentifier,
currentCatalog = context.getCurrentCatalogOpt,
@@ -450,22 +611,6 @@ class SqlGraphRegistrationContext(
)
.identifier
- graphRegistrationContext.registerFlow(
- UntypedFlow(
- identifier = flowIdentifier,
- destinationIdentifier = qualifiedDestinationIdentifier,
- func =
FlowAnalysis.createFlowFunctionFromLogicalPlan(flowQueryLogicalPlan),
- sqlConf = context.getSqlConf,
- once = false,
- queryContext = QueryContext(
- currentCatalog = context.getCurrentCatalogOpt,
- currentDatabase = context.getCurrentDatabaseOpt
- ),
- origin = queryOrigin
- )
- )
- }
-
private def validateInsertIntoFlow(
insertIntoStatement: InsertIntoStatement,
queryOrigin: QueryOrigin
@@ -557,25 +702,24 @@ class SqlGraphRegistrationContext(
}
object PartitionHelper {
- import org.apache.spark.sql.connector.expressions.{IdentityTransform,
Transform}
+ import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
IdentityTransform, Transform}
- def applyPartitioning(partitioning: Seq[Transform], queryOrigin:
QueryOrigin): Seq[String] = {
- partitioning.foreach {
- case _: IdentityTransform =>
- case other =>
- throw SqlGraphElementRegistrationException(
- msg = s"Invalid partitioning transform ($other)",
- queryOrigin = queryOrigin
- )
- }
- partitioning.collect {
+ /**
+ * Splits the parsed transforms of a `CREATE STREAMING TABLE`/`CREATE
MATERIALIZED VIEW` statement
+ * into partition columns (`PARTITIONED BY`) and cluster columns (`CLUSTER
BY`).
+ *
+ * The parser folds both clauses into a single `partitioning` sequence --
`PARTITIONED BY` as
+ * [[IdentityTransform]]s and `CLUSTER BY` as a single
[[ClusterByTransform]] -- and guarantees
+ * the two clauses are mutually exclusive. This mirrors how the
Connect/Python path populates
+ * `partitionCols` and `clusterCols` from its proto fields.
+ *
+ * @return a pair of (partition columns, cluster columns).
+ */
+ def splitPartitionAndClusterColumns(
+ partitioning: Seq[Transform],
+ queryOrigin: QueryOrigin): (Seq[String], Seq[String]) = {
+ val partitionCols = partitioning.collect {
case t: IdentityTransform =>
- if (t.references.length != 1) {
- throw SqlGraphElementRegistrationException(
- msg = "Only single column based partitioning is supported.",
- queryOrigin = queryOrigin
- )
- }
if (t.ref.fieldNames().length != 1) {
throw SqlGraphElementRegistrationException(
msg = "Multipart partition identifier not allowed.",
@@ -584,6 +728,30 @@ object PartitionHelper {
}
t.ref.fieldNames().head
}
+ val clusterCols = partitioning.collect {
+ case ClusterByTransform(columnNames) =>
+ columnNames.map { ref =>
+ if (ref.fieldNames().length != 1) {
+ throw SqlGraphElementRegistrationException(
+ msg = "Multipart cluster identifier not allowed.",
+ queryOrigin = queryOrigin
+ )
+ }
+ ref.fieldNames().head
+ }
+ }.flatten
+ // Reject any transform that is neither an identity partition nor a
clustering transform (e.g.
+ // a non-identity partition transform like `year(col)`).
+ partitioning.foreach {
+ case _: IdentityTransform =>
+ case ClusterByTransform(_) =>
+ case other =>
+ throw SqlGraphElementRegistrationException(
+ msg = s"Invalid partitioning transform ($other)",
+ queryOrigin = queryOrigin
+ )
+ }
+ (partitionCols, clusterCols)
}
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
index 90132c7ea7bb..3297709174c5 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
@@ -18,7 +18,9 @@ package org.apache.spark.sql.pipelines.graph
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.catalyst.parser.ParseException
-import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{Identifier,
SharedTablesInMemoryRowLevelOperationTableCatalog, TableCatalog}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType}
import org.apache.spark.sql.pipelines.utils.{PipelineTest,
TestGraphRegistrationContext}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{LongType, StructType}
@@ -267,6 +269,37 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
)
}
+ test("Cluster cols correctly registered for all pipeline dataset SQL forms")
{
+ val autoCdcSuffix =
+ s"""
+ |FLOW AUTO CDC
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id1)
+ |SEQUENCE BY id1""".stripMargin
+
+ // (statement, table name) for each of the four SQL forms that accept
CLUSTER BY.
+ val cases = Seq(
+ // CREATE STREAMING TABLE (no query)
+ ("CREATE STREAMING TABLE st_plain CLUSTER BY (id1, id2);", "st_plain"),
+ // CREATE STREAMING TABLE ... AS SELECT
+ ("CREATE STREAMING TABLE st_as CLUSTER BY (id1, id2) " +
+ s"AS SELECT * FROM STREAM $externalTable1Ident;", "st_as"),
+ // CREATE STREAMING TABLE ... FLOW AUTO CDC
+ (s"CREATE STREAMING TABLE st_cdc CLUSTER BY (id1, id2) $autoCdcSuffix;",
"st_cdc"),
+ // CREATE MATERIALIZED VIEW ... AS SELECT
+ ("CREATE MATERIALIZED VIEW mv CLUSTER BY (id1, id2) " +
+ "AS SELECT id AS id1, id AS id2 FROM range(1, 2);", "mv"))
+
+ cases.foreach { case (sqlText, tableName) =>
+ val graph = unresolvedDataflowGraphFromSql(sqlText = sqlText)
+ val table = graph.tables.find(_.identifier ==
fullyQualifiedIdentifier(tableName)).get
+ assert(table.clusterCols.contains(Seq("id1", "id2")),
+ s"Expected clusterCols Seq(id1, id2) for '$tableName', got
${table.clusterCols}")
+ assert(table.partitionCols.forall(_.isEmpty),
+ s"Expected no partitionCols for '$tableName', got
${table.partitionCols}")
+ }
+ }
+
test("MV/ST with partition columns works") {
withTable("mv", "st") {
val unresolvedDataflowGraph = unresolvedDataflowGraphFromSql(
@@ -1080,4 +1113,244 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
}
}
}
+
+ //
===========================================================================
+ // AUTO CDC syntax registration and execution.
+ //
+ // Two SQL forms register an [[AutoCdcFlow]] into the dataflow graph:
+ // 1. CREATE STREAMING TABLE <name> FLOW AUTO CDC FROM <source> KEYS (...)
SEQUENCE BY <expr>
+ // 2. CREATE FLOW <name> AS AUTO CDC INTO <target> FROM <source> KEYS
(...) SEQUENCE BY <expr>
+ // SQL AUTO CDC only supports SCD Type 1.
+ //
===========================================================================
+
+ /** Returns the single unresolved [[AutoCdcFlow]] registered for the given
flow identifier. */
+ private def autoCdcFlowFor(graph: DataflowGraph, name: String): AutoCdcFlow
= {
+ val ident = fullyQualifiedIdentifier(name)
+ graph.flows.collect {
+ case f: AutoCdcFlow if f.identifier == ident => f
+ }.headOption.getOrElse(
+ fail(s"No AutoCdcFlow registered for identifier ${ident.unquotedString}")
+ )
+ }
+
+ test("CREATE STREAMING TABLE FLOW AUTO CDC registers a streaming table and
an AutoCDC flow") {
+ val graph = unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE st
+ |FLOW AUTO CDC
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id)
+ |SEQUENCE BY id
+ |""".stripMargin
+ )
+
+ // The streaming table is registered as a table.
+ assert(graph.tables.exists(_.identifier == fullyQualifiedIdentifier("st")))
+
+ // The backing flow is an AutoCdcFlow whose flow and destination are the
streaming table.
+ val flow = autoCdcFlowFor(graph, "st")
+ assert(flow.destinationIdentifier == fullyQualifiedIdentifier("st"))
+ assert(flow.changeArgs.keys.map(_.name) == Seq("id"))
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ assert(flow.changeArgs.deleteCondition.isEmpty)
+ assert(flow.changeArgs.columnSelection.isEmpty)
+ }
+
+ test("CREATE FLOW AS AUTO CDC INTO registers an AutoCDC flow targeting a
streaming table") {
+ val graph = unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE target;
+ |CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id)
+ |SEQUENCE BY id
+ |""".stripMargin
+ )
+
+ assert(graph.tables.exists(_.identifier ==
fullyQualifiedIdentifier("target")))
+
+ // The flow identifier is the named flow, and its destination is the
target streaming table.
+ val flow = autoCdcFlowFor(graph, "f")
+ assert(flow.destinationIdentifier == fullyQualifiedIdentifier("target"))
+ assert(flow.changeArgs.keys.map(_.name) == Seq("id"))
+ assert(flow.changeArgs.storedAsScdType == ScdType.Type1)
+ }
+
+ test("AUTO CDC optional clauses map onto ChangeArgs") {
+ val graph = unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE target;
+ |CREATE FLOW f AS AUTO CDC INTO target
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id)
+ |APPLY AS DELETE WHEN id = 0
+ |SEQUENCE BY id
+ |COLUMNS * EXCEPT (id)
+ |""".stripMargin
+ )
+
+ val flow = autoCdcFlowFor(graph, "f")
+ assert(flow.changeArgs.deleteCondition.isDefined)
+ flow.changeArgs.columnSelection match {
+ case Some(ColumnSelection.ExcludeColumns(cols)) =>
assert(cols.map(_.name) == Seq("id"))
+ case other => fail(s"Expected ExcludeColumns(id), got $other")
+ }
+ }
+
+ test("AUTO CDC COLUMNS include list maps onto ChangeArgs") {
+ val graph = unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE st
+ |FLOW AUTO CDC
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id)
+ |SEQUENCE BY id
+ |COLUMNS (id)
+ |""".stripMargin
+ )
+
+ autoCdcFlowFor(graph, "st").changeArgs.columnSelection match {
+ case Some(ColumnSelection.IncludeColumns(cols)) =>
assert(cols.map(_.name) == Seq("id"))
+ case other => fail(s"Expected IncludeColumns(id), got $other")
+ }
+ }
+
+ test("Explicit column list is rejected for CREATE STREAMING TABLE FLOW AUTO
CDC") {
+ val ex = intercept[SqlGraphElementRegistrationException] {
+ unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE st (id INT, name STRING)
+ |FLOW AUTO CDC
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id)
+ |SEQUENCE BY id
+ |""".stripMargin
+ )
+ }
+ assert(ex.getMessage.contains("Explicit column lists are not supported for
AUTO CDC"))
+ }
+
+ test("Multipart AUTO CDC flow name is not supported") {
+ Seq("a.b", "a.b.c").foreach { flowIdentifier =>
+ val ex = intercept[AnalysisException] {
+ unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE target;
+ |CREATE FLOW $flowIdentifier AS AUTO CDC INTO target
+ |FROM STREAM $externalTable1Ident
+ |KEYS (id)
+ |SEQUENCE BY id
+ |""".stripMargin
+ )
+ }
+ checkError(
+ exception = ex,
+ condition = "MULTIPART_FLOW_NAME_NOT_SUPPORTED",
+ parameters = Map("flowName" -> flowIdentifier)
+ )
+ }
+ }
+
+ //
---------------------------------------------------------------------------
+ // End-to-end execution. AutoCDC applies its microbatch to the target via
the DataFrame
+ // MERGE API, which requires a row-level-operation-capable v2 catalog, so
these tests scaffold
+ // one inline (mirroring the AutoCDC E2E suites) rather than using the
default `spark_catalog`.
+ //
---------------------------------------------------------------------------
+
+ private val rowLevelCatalog: String = "cat"
+ private val rowLevelNamespace: String = "ns1"
+
+ private def withRowLevelAutoCdcCatalog(testBody: => Unit): Unit = {
+ spark.conf.set(
+ s"spark.sql.catalog.$rowLevelCatalog",
+ classOf[SharedTablesInMemoryRowLevelOperationTableCatalog].getName
+ )
+ // Surface flow failures on the first attempt instead of retrying.
+ spark.conf.set(SQLConf.PIPELINES_MAX_FLOW_RETRY_ATTEMPTS.key, "0")
+ spark.sql(s"CREATE NAMESPACE IF NOT EXISTS
$rowLevelCatalog.$rowLevelNamespace")
+ try {
+ testBody
+ } finally {
+ SharedTablesInMemoryRowLevelOperationTableCatalog.reset()
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$rowLevelCatalog")
+
spark.sessionState.conf.unsetConf(SQLConf.PIPELINES_MAX_FLOW_RETRY_ATTEMPTS.key)
+ }
+ }
+
+ /** Build a target row's `_cdc_metadata` struct value (deleteSequence,
upsertSequence). */
+ private def cdcMeta(deleteSeq: Option[Long], upsertSeq: Option[Long]): Row =
+ Row(deleteSeq.orNull, upsertSeq.orNull)
+
+ test("CREATE STREAMING TABLE FLOW AUTO CDC upserts rows into the target
end-to-end") {
+ withRowLevelAutoCdcCatalog {
+ // Source and target both live in the row-level catalog. Streaming over
a static table
+ // replays all rows in one microbatch, exercising the SCD1 upsert path.
Two versions of
+ // key 1 test latest-wins.
+ val source = s"$rowLevelCatalog.$rowLevelNamespace.cdc_source"
+ spark.sql(s"CREATE TABLE $source (id INT, name STRING, version BIGINT)")
+ spark.sql(
+ s"INSERT INTO $source VALUES (1, 'alice', 1), (1, 'alice2', 2), (2,
'bob', 1)")
+
+ val target = s"$rowLevelCatalog.$rowLevelNamespace.target"
+ val graph = unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE $target
+ |FLOW AUTO CDC
+ |FROM STREAM $source
+ |KEYS (id)
+ |SEQUENCE BY version
+ |""".stripMargin
+ )
+
+ startPipelineAndWaitForCompletion(graph)
+
+ // Key 1 converges to the highest-sequenced upsert (alice2 @ v2); key 2
lands as-is.
+ checkAnswer(
+ spark.table(target),
+ Seq(
+ Row(1, "alice2", 2L, cdcMeta(None, Some(2L))),
+ Row(2, "bob", 1L, cdcMeta(None, Some(1L)))
+ )
+ )
+ }
+ }
+
+ test("CREATE FLOW AS AUTO CDC INTO applies deletes and column exclusion
end-to-end") {
+ withRowLevelAutoCdcCatalog {
+ // Source carries a control column `op` that drives the delete condition
and is excluded
+ // from the target projection.
+ val source = s"$rowLevelCatalog.$rowLevelNamespace.cdc_source"
+ spark.sql(s"CREATE TABLE $source (id INT, name STRING, version BIGINT,
op STRING)")
+ spark.sql(
+ s"""INSERT INTO $source VALUES
+ | (1, 'alice', 1, 'UPSERT'),
+ | (2, 'bob', 1, 'UPSERT'),
+ | (2, 'bob', 2, 'DELETE')
+ |""".stripMargin)
+
+ val target = s"$rowLevelCatalog.$rowLevelNamespace.target"
+ val graph = unresolvedDataflowGraphFromSql(
+ sqlText = s"""
+ |CREATE STREAMING TABLE $target;
+ |CREATE FLOW f AS AUTO CDC INTO $target
+ |FROM STREAM $source
+ |KEYS (id)
+ |APPLY AS DELETE WHEN op = 'DELETE'
+ |SEQUENCE BY version
+ |COLUMNS * EXCEPT (op)
+ |""".stripMargin
+ )
+
+ startPipelineAndWaitForCompletion(graph)
+
+ // Key 2's delete @ v2 supersedes its upsert @ v1; only key 1 remains.
The `op` column is
+ // excluded from the target schema.
+ checkAnswer(
+ spark.table(target),
+ Seq(Row(1, "alice", 1L, cdcMeta(None, Some(1L))))
+ )
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
