szehon-ho commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3562810910
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala:
##########
@@ -162,6 +166,10 @@ class SqlGraphRegistrationContext(
case createStreamingTableCommand: CreateStreamingTable =>
// CREATE STREAMING TABLE [ streaming_table_name ] [ options ]
CreateStreamingTableHandler.handle(createStreamingTableCommand,
queryOrigin)
+ case createStreamingTableAutoCdcCommand: CreateStreamingTableAutoCdc =>
Review Comment:
This registers `CreateStreamingTableAutoCdc` in the pipeline graph path, but
the normal Spark execution guard in `SparkStrategies.Pipelines` only rejects
`CreateFlowCommand` and `CreateStreamingTableAsSelect` today. If someone runs
this statement through regular `spark.sql` outside pipeline registration, it
may miss the curated unsupported `STREAMING TABLE` error. Can we add a matching
`CreateStreamingTableAutoCdc` case there?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala:
##########
@@ -256,6 +264,108 @@ 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
+
+ // 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 =
+
Option.when(cst.columns.nonEmpty)(StructType(cst.columns.map(_.toV1Column))),
Review Comment:
If users write `CREATE STREAMING TABLE target (id INT, name STRING) FLOW
AUTO CDC ...`, this stores the listed columns as the table’s full specified
schema. But Auto CDC later appends `_cdc_metadata` to the flow schema, so
validation can reject an otherwise natural data-column-only declaration. Could
we either reject column lists for this SQL form, or interpret them as the data
schema and add the metadata field internally before validation?
--
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]
