anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3576291574
##########
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 =
Review Comment:
In the existing Python API, there is no combined command - you have to
create the streaming table and the flow separately. But if you create the table
with a schema, then the flow schema must match it exactly. So it will fail if
you don't specify the metadata column. The same will happen in SQL if you
create the streaming table and the flow separately.
I am not opposed to restricting the combined syntax as you suggest, but it
does not solve the problem for separate declarations.
Let me follow your suggestion for now and disallow a schema in this case. I
will follow up with a separate Jira for the data-columns-only approach.
--
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]
