Re: [PR] [SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph [spark]

["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