You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anew (via GitHub)" <gi...@apache.org> on 2026/07/09 23:38:29 UTC
[PR] [SPARK-57402][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
anew opened a new pull request, #57176:
URL: https://github.com/apache/spark/pull/57176
### 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
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3573316920
##########
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:
great catch!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3576728663
##########
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:
... and https://issues.apache.org/jira/browse/SPARK-58118 for the longer-term 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3574459751
##########
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:
I addressed this, and in the process discovered that the same issue existed for a blank CREATE STREAMING TABLE (without a flow), and also for the CREATE FLOW AS AUTO CDC. The latter actually resulted in a `CreateFlow` command with a child `AutoCdcIntoCommand` that was a command in itself. I realized that this should not be a command (because it can never occur by itself, only within a CREATE FLOW), so I changed this to be named just `AutoCdcInto` and not extending `Command` anymore. Also added tests for all these cases. I think this is cleaner now.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3576699716
##########
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:
I created [SPARK-58116](https://issues.apache.org/jira/browse/SPARK-58116) for the quick fix.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3573361712
##########
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:
This is a very interesting question. What if the given schema does not match the keys+columns? That should be rejected, I guess. A similar case is where the the streaming table is declared without a flow but with a schema. In that case we do not know(at parse time) whether the flow that is also declared on its own is an Auto CDC flow? It appears to me that need to allow specifying the schema, but interpret it as exclusive of the metadata column.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57176:
URL: https://github.com/apache/spark/pull/57176#discussion_r3576728663
##########
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:
... and [SPARK-58118](https://issues.apache.org/jira/browse/SPARK-58118) for the longer-term 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57176: [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph
URL: https://github.com/apache/spark/pull/57176
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org