Skip to content

[SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph #57176

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anew wants to merge 1 commit into
base: master
Choose a base branch
from
+384 −30
Open

[SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph #57176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
185 changes: 156 additions & 29 deletions ...nes/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,12 +20,16 @@ import scala.collection.mutable

import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.QueryPlanningTracker
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable, CreateStreamingTableAsSelect, CreateView, InsertIntoStatement, LogicalPlan}
import org.apache.spark.sql.Column
import org.apache.spark.sql.catalyst.{QueryPlanningTracker, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.logical.{AutoCdcIntoCommand, CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable, CreateStreamingTableAsSelect, CreateStreamingTableAutoCdc, CreateView, InsertIntoStatement, LogicalPlan}
import org.apache.spark.sql.catalyst.util.StringUtils
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.execution.command.{CreateViewCommand, SetCatalogCommand, SetCommand, SetNamespaceCommand}
import org.apache.spark.sql.pipelines.Language
import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ColumnSelection, ScdType, UnqualifiedColumnName}
import org.apache.spark.sql.types.StructType

/**
Expand Down Expand Up @@ -162,6 +166,10 @@ class SqlGraphRegistrationContext(
case createStreamingTableCommand: CreateStreamingTable =>
// CREATE STREAMING TABLE [ streaming_table_name ] [ options ]
CreateStreamingTableHandler.handle(createStreamingTableCommand, queryOrigin)
case createStreamingTableAutoCdcCommand: CreateStreamingTableAutoCdc =>

@szehon-ho szehon-ho Jul 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

// CREATE STREAMING TABLE [ streaming_table_name ] [ options ]
// FLOW AUTO CDC FROM [ source ] KEYS ( ... ) SEQUENCE BY [ expr ] ...
CreateStreamingTableAutoCdcHandler.handle(createStreamingTableAutoCdcCommand, queryOrigin)
case createFlowCommand: CreateFlowCommand =>
// CREATE FLOW [ flow_name ] AS INSERT INTO [ destination_name ] BY NAME
CreateFlowHandler.handle(createFlowCommand, queryOrigin)
Expand Down Expand Up @@ -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))),

@szehon-ho szehon-ho Jul 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

partitionCols = Option(PartitionHelper.applyPartitioning(cst.partitioning, queryOrigin)),
clusterCols = None,
properties = cst.tableSpec.properties,
origin = queryOrigin.copy(
objectName = Option(stIdentifier.unquotedString),
objectType = Option(QueryOriginType.Table.toString)
),
format = cst.tableSpec.provider,
normalizedPath = None,
isStreamingTable = true
)
)

// Register the AutoCDC flow that backs this streaming table. Both the flow and its
// destination are the streaming table itself.
graphRegistrationContext.registerFlow(
AutoCdcFlow(
identifier = stIdentifier,
destinationIdentifier = stIdentifier,
func = FlowAnalysis.createFlowFunctionFromLogicalPlan(cst.source),
sqlConf = context.getSqlConf,
queryContext = QueryContext(
currentCatalog = context.getCurrentCatalogOpt,
currentDatabase = context.getCurrentDatabaseOpt
),
origin = queryOrigin.copy(
objectName = Option(stIdentifier.unquotedString),
objectType = Option(QueryOriginType.Flow.toString)
),
changeArgs = buildChangeArgs(
keys = cst.keys,
sequenceByExpr = cst.sequenceByExpr,
deleteCondition = cst.deleteCondition,
includeColumns = cst.includeColumns,
excludeColumns = cst.excludeColumns,
queryOrigin = queryOrigin
)
)
)
}
}

private object CreateMaterializedViewAsSelectHandler {
def handle(cmv: CreateMaterializedViewAsSelect, queryOrigin: QueryOrigin): Unit = {
val mvIdentifier = GraphIdentifierManager
Expand Down Expand Up @@ -415,7 +525,7 @@ class SqlGraphRegistrationContext(
)
.identifier

val (flowTargetDatasetIdentifier, flowQueryLogicalPlan) = cf.flowOperation match {
cf.flowOperation match {
case i: InsertIntoStatement =>
validateInsertIntoFlow(i, queryOrigin)
val flowTargetDatasetName = i.table match {
Expand All @@ -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: AutoCdcIntoCommand =>
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.",
queryOrigin = queryOrigin
)
}
}

val qualifiedDestinationIdentifier = GraphIdentifierManager
/** Qualifies a raw flow target dataset identifier against the current catalog/database. */
private def qualifyDestinationIdentifier(
flowTargetDatasetIdentifier: TableIdentifier): TableIdentifier =
GraphIdentifierManager
.parseAndQualifyFlowIdentifier(
rawFlowIdentifier = flowTargetDatasetIdentifier,
currentCatalog = context.getCurrentCatalogOpt,
currentDatabase = context.getCurrentDatabaseOpt
)
.identifier

graphRegistrationContext.registerFlow(
UntypedFlow(
identifier = flowIdentifier,
destinationIdentifier = qualifiedDestinationIdentifier,
func = FlowAnalysis.createFlowFunctionFromLogicalPlan(flowQueryLogicalPlan),
sqlConf = context.getSqlConf,
once = false,
queryContext = QueryContext(
currentCatalog = context.getCurrentCatalogOpt,
currentDatabase = context.getCurrentDatabaseOpt
),
origin = queryOrigin
)
)
}

private def validateInsertIntoFlow(
insertIntoStatement: InsertIntoStatement,
queryOrigin: QueryOrigin
Expand Down
Loading