anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3575691155
##########
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:
Thanks for catching this — this matches where I landed too, and I went and
investigated both the current behavior and the "user specifies
data-columns-only" approach. Two findings worth sharing:
1. The two-statement form doesn't fail at validation today — it silently
skips validation entirely. I expected this case (CREATE STREAMING TABLE target
(id INT, name STRING) + separate CREATE FLOW ... AS AUTO CDC INTO target) to at
least be caught by `validateUserSpecifiedSchemas`, but it isn't. That check
keys the table lookup on the flow's identifier:
```
flows.flatMap(f => table.get(f.identifier))
```
which only matches when the flow is an implicit/default flow (flow id ==
table id) — i.e. the combined form. For a named flow, table.get(f.identifier)
is None, so the declared schema is never validated (its sibling
`validateFlowStreamingness` correctly keys on the destination identifier). So
the mismatch isn't even caught at graph-validation time for the two-statement
pattern — it surfaces later as a mid-stream UNRESOLVED_COLUMN ... _cdc_metadata
at materialization. That looks like a pre-existing bug independent of this PR,
it pre-exists also for Python, and I think it deserves its own fix (keying on
destinationIdentifier) + regression test.
2. How to fix it? I propose to make a quick fix for both SQL and Python to
use the destinationIdentifier for lookup, in a separate PR. Then in a separate
Jira, we can discuss whether we want to allow an explicit schema that omits the
metadata column, in all cases. This would be beyond the scope of this PR
because it changes existing, released behavior in a non-regressing way, but a
change nonetheless.
What do you think, @szehon-ho ?
--
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]
