szehon-ho commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3575486686
##########
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:
If the user writes `CREATE STREAMING TABLE target (id INT, name STRING) FLOW
AUTO CDC ...`, we store the column list as `specifiedSchema`. But
`AutoCdcMergeFlow` always appends `_cdc_metadata` to the flow output schema, so
`GraphValidations.validateUserSpecifiedSchemas` later rejects the pipeline with
`USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE` (specified = data cols
only, inferred = data cols + `_cdc_metadata`). The parser allows this form
(`AutoCdcParserSuite`), but the new E2E tests avoid it by omitting the column
list.
Would it be better to reject it here instead of letting registration succeed
and failing at graph validation? For example:
```scala
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 source.",
queryOrigin = queryOrigin)
}
```
That gives a clear, immediate error for the single-statement form. The
tradeoff is it doesn't cover the two-statement pattern:
```sql
CREATE STREAMING TABLE target (id INT, name STRING);
CREATE FLOW f AS AUTO CDC INTO target ...
```
where `CreateStreamingTableHandler` registers the schema and a later `CREATE
FLOW` attaches AUTO CDC — rejection would need to live in validation (or we'd
need to reject column lists on all standalone `CREATE STREAMING TABLE`
statements, which is too broad).
Longer term, treating `specifiedSchema` as data-columns-only for AUTO CDC
targets (and appending `_cdc_metadata` during validation/materialization) is
probably the right UX. But if we want a small fix for this PR, rejecting
non-empty `cst.columns` here seems reasonable for the combined syntax — WDYT?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala:
##########
@@ -427,45 +537,62 @@ 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
+ )
+ )
+ 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
+ )
)
- .identifier
- (qualifiedFlowTargetDatasetName, i.query)
+ )
case _ =>
throw SqlGraphElementRegistrationException(
- msg = "Unable flow type. Only INSERT INTO flows are supported.",
+ msg = "Unable flow type. Only INSERT INTO and AUTO CDC INTO flows
are supported.",
Review Comment:
Nit: `"Unable flow type..."` → `"Unknown flow type..."` (or similar).
##########
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))),
+ partitionCols =
Option(PartitionHelper.applyPartitioning(cst.partitioning, queryOrigin)),
Review Comment:
`AutoCdcParserSuite` has a test that `CLUSTER BY` is honored at parse time —
it lands in `cst.partitioning` as a `ClusterByTransform`. But this call routes
all of `partitioning` through `PartitionHelper`, which only accepts
`IdentityTransform` (`PARTITIONED BY`) and throws `Invalid partitioning
transform (ClusterByTransform(...))` for `CLUSTER BY`. So the parser test and
pipeline registration disagree.
Options:
1. Split `ClusterByTransform` out of `partitioning` into `clusterCols` here
(Connect already sets `clusterCols` from clustering columns).
2. Reject `CLUSTER BY` explicitly in this handler with a clear error, and
adjust/remove the parser test expectation if we don't support it yet.
Same pattern exists in the other SQL table handlers
(`CreateStreamingTableHandler`, etc.), but AUTO CDC is the one with a parser
test claiming `CLUSTER BY` works.
--
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]
