AnishMahto commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3306960988
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ 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
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** 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
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ 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.
+ */
+ private 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 catalog = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ .asInstanceOf[TableCatalog]
+ 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 AutoCdcMergeWriteMixin {
+
+ override def getOrigin: QueryOrigin = flow.origin
+
+ override protected def changeArgs: ChangeArgs = flow.changeArgs
+
+ /**
+ * Resolved Spark [[DataType]] of the sequencing expression.
+ */
+ private def sequencingType: DataType =
+ flow.df.select(flow.changeArgs.sequencing).schema.head.dataType
Review Comment:
Yeah good call, I should just reuse.
--
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]
