szehon-ho commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3307967712
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +313,185 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Base trait for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteBase {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ *
+ * Note that this is `CREATE TABLE IF NOT EXISTS`: when the aux table
already exists, its
+ * schema is left untouched and `auxiliaryTableSchema` is ignored. For SCD1,
they keys must be
+ * invariant across executions and the CDC metadata will always be present,
so this is correct.
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
+ val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+ // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
+ // target's format is unspecified (None), omit the USING clause and fall
back to the
+ // session's default source provider.
+ val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
+ spark.sql(
+ s"""CREATE TABLE IF NOT EXISTS
+ |${auxIdent.quotedString}
+ |(${auxiliaryTableSchema.toDDL}) $usingClause""".stripMargin
+ )
+ auxIdent
+ }
+
+ /**
+ * Validate that the target table's underlying connector implements
+ * [[SupportsRowLevelOperations]], which is the V2 connector contract for
MERGE/UPDATE/DELETE
+ * with rewrite - all operations that the AutoCDC transformation executes.
+ */
+ protected def requireDestinationSupportsRowLevelOps(): Unit = {
+ val (catalog, v2Identifier) = resolveTableCatalog(spark,
destination.identifier)
+ val destinationTable = catalog.loadTable(v2Identifier)
+
+ if (!destinationTable.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> destination.identifier.quotedString,
+ "format" -> destination.format.getOrElse("<session default>")
+ )
+ )
+ }
+ }
+
+ private def resolveTableCatalog(
+ spark: SparkSession,
+ ident: TableIdentifier): (TableCatalog, Identifier) = {
+ val catalogManager = spark.sessionState.catalogManager
+ val catalogPlugin = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ val catalog = catalogPlugin match {
+ case t: TableCatalog => t
+ case _ => throw
QueryCompilationErrors.missingCatalogTablesAbilityError(catalogPlugin)
+ }
+ val namespace = ident.database.getOrElse(
+ throw SparkException.internalError(
+ s"Cannot resolve table identifier ${ident.quotedString}: namespace is
unspecified."
+ )
+ )
+ (catalog, Identifier.of(Array(namespace), ident.table))
+ }
+}
+
+/**
+ * A [[StreamingFlowExecution]] that applies a CDC event stream to a target
[[Table]] via
+ * SCD Type 1 MERGE semantics.
+ */
+class Scd1MergeStreamingWrite(
+ val identifier: TableIdentifier,
+ val flow: AutoCdcMergeFlow,
+ val graph: DataflowGraph,
+ val updateContext: PipelineUpdateContext,
+ val checkpointPath: String,
+ val trigger: Trigger,
+ val destination: Table,
+ val sqlConf: Map[String, String]
+) extends StreamingFlowExecution with AutoCdcMergeWriteBase {
+
+ override def getOrigin: QueryOrigin = flow.origin
+
+ override protected def changeArgs: ChangeArgs = flow.changeArgs
+
+ override def startStream(): StreamingQuery = {
+ requireDestinationSupportsRowLevelOps()
+
+ val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
+
+ // The auxiliary table is created here (at flow execution) rather than
during flow resolution
+ // or dataset materialization for two reasons:
+ // 1. It is an internal state store: we deliberately keep it out of the
graph registration
+ // context's table set so that it is invisible to other flows and the
[[DatasetManager]]
+ // will never materialize it.
+ // 2. Its format must match the target table's, which only exists after
the target is
+ // materialized. Flow resolution must also stay side-effect free
(e.g. for dry runs).
+ val auxiliaryTableIdentifier = createAuxiliaryTableIfNotExists(spark =
updateContext.spark)
+
+ val foreachBatchHandler = Scd1ForeachBatchHandler(
+ batchProcessor = Scd1BatchProcessor(
+ changeArgs = flow.changeArgs,
+ resolvedSequencingType = flow.sequencingType
+ ),
+ auxiliaryTableIdentifier = auxiliaryTableIdentifier,
+ targetTableIdentifier = destination.identifier
+ )
+
+ sourceChangeDataFeed.writeStream
+ .queryName(displayName)
+ .option("checkpointLocation", checkpointPath)
+ .trigger(trigger)
+ .foreachBatch((batch: Dataset[Row], batchId: Long) => {
+ foreachBatchHandler.execute(batch, batchId)
+ })
+ .start()
+ }
+
+ override protected lazy val auxiliaryTableSchema: StructType =
+ // SCD1's auxiliary table is just keys + the CDC metadata struct; no user
data columns. Keys
+ // come first, in `changeArgs.keys` declaration order, to anchor the
per-key sequence
+ // watermark used to gate out-of-order events.
+ StructType(autoCdcKeyFields :+ cdcMetadataField)
+
+ /**
+ * AutoCDC key columns resolved out of the flow's augmented schema, in
+ * `changeArgs.keys` declaration order. Keys are guaranteed to be present in
the schema
+ * because [[AutoCdcMergeFlow.schema]] validates that.
+ */
+ private lazy val autoCdcKeyFields: Seq[StructField] = {
Review Comment:
ok
--
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]
