[SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end

common/utils/src/main/resources/error/error-conditions.json

@@ -343,12 +343,6 @@
     ],
     "sqlState" : "42710"
   },
-  "AUTOCDC_SCD2_NOT_SUPPORTED" : {
-    "message" : [
-      "AutoCDC flows do not currently support SCD Type 2 transformations."
-    ],
-    "sqlState" : "0A000"
-  },
   "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE" : {
     "message" : [
       "Cannot start AutoCDC flow: the target table <tableName> (format: <format>) does not support row-level operations. AutoCDC requires a target backed by a connector that supports MERGE."

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala

@@ -1335,7 +1335,7 @@ object Scd2BatchProcessor {
    * CDC metadata column field that represents the exact time (sequence) of the CDC event that
    * produced this row. Null only for synthetic decomposition tails.
    */
-  private[autocdc] val recordStartAtFieldName: String = "__RECORD_START_AT"
+  private[pipelines] val recordStartAtFieldName: String = "__RECORD_START_AT"
 
   /**
    * Aux-table only column that holds the microbatch id by which a row was logically

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala

@@ -28,7 +28,12 @@ import org.apache.spark.sql.{Dataset, Row}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.classic.ClassicConversions._
 import org.apache.spark.sql.classic.SparkSession
-import org.apache.spark.sql.pipelines.autocdc.{Scd1BatchProcessor, Scd1ForeachBatchHandler}
+import org.apache.spark.sql.pipelines.autocdc.{
+  Scd1BatchProcessor,
+  Scd1ForeachBatchHandler,
+  Scd2BatchProcessor,
+  Scd2ForeachBatchHandler
+}
 import org.apache.spark.sql.pipelines.graph.QueryOrigin.ExceptionHelpers
 import org.apache.spark.sql.pipelines.util.SparkSessionUtils
 import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
@@ -349,3 +354,48 @@ class Scd1MergeStreamingWrite(
       .start()
   }
 }
+
+/**
+ * A [[StreamingFlowExecution]] that applies a CDC event stream to a target [[Table]] via
+ * SCD Type 2 MERGE semantics.
+ */
+class Scd2MergeStreamingWrite(
+    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 {
+
+  override def getOrigin: QueryOrigin = flow.origin
+
+  override def startStream(): StreamingQuery = {
+    val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
+
+    // The auxiliary table is created and evolved during dataset materialization (see
+    // [[DatasetManager]]), so it already exists by the time this flow executes; resolve its
+    // identifier to hand to the foreachBatch handler.
+    val auxiliaryTableIdentifier = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+
+    val foreachBatchHandler = Scd2ForeachBatchHandler(
+      batchProcessor = Scd2BatchProcessor(
+        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()
+  }
+}

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala

@@ -17,7 +17,6 @@
 
 package org.apache.spark.sql.pipelines.graph
 
-import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.pipelines.autocdc.ScdType
 import org.apache.spark.sql.streaming.Trigger
 
@@ -96,10 +95,21 @@ class FlowPlanner(
               case _ => unsupportedDestinationType(acmf, output)
             }
           case ScdType.Type2 =>
-            throw new AnalysisException(
-              errorClass = "AUTOCDC_SCD2_NOT_SUPPORTED",
-              messageParameters = Map.empty
-            )
+            val flowMetadata = FlowSystemMetadata(updateContext, acmf, graph)
+            output match {
+              case o: Table =>
+                new Scd2MergeStreamingWrite(
+                  identifier = acmf.identifier,
+                  flow = acmf,
+                  graph = graph,
+                  updateContext = updateContext,
+                  checkpointPath = flowMetadata.latestCheckpointLocation,
+                  trigger = triggerFor(acmf),
+                  destination = o,
+                  sqlConf = acmf.sqlConf
+                )
+              case _ => unsupportedDestinationType(acmf, output)
+            }
         }
       case _ =>
         throw new UnsupportedOperationException(

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala

@@ -732,19 +732,6 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession {
     }
   }
 
-  test(
-    "the reserved framework-column check runs before the SCD2-not-supported gate"
-  ) {
-    // The reserved-name error is more actionable than AUTOCDC_SCD2_NOT_SUPPORTED, so it must win
-    // for an SCD2 flow that both is unsupported and carries a colliding source column. This also
-    // keeps the check meaningful today (before SCD2 is supported) and correct once it lands.
-    val sourceDf = sourceDfWithExtraColumns(Scd2BatchProcessor.startAtColName -> StringType)
-    val ex = intercept[AnalysisException] {
-      newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2)
-    }
-    assert(ex.getCondition == "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT")
-  }
-
   test(
     "an SCD1 flow with a source column matching an SCD2-only reserved name is allowed"
   ) {

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcGraphExecutionTestMixin.scala

@@ -29,6 +29,7 @@ import org.apache.spark.sql.pipelines.autocdc.{
   ChangeArgs,
   ColumnSelection,
   Scd1BatchProcessor,
+  Scd2BatchProcessor,
   ScdType,
   UnqualifiedColumnName
 }
@@ -140,19 +141,36 @@ trait AutoCdcGraphExecutionTestMixin extends BeforeAndAfterEach {
   }
 
   /**
-   * DDL fragment for the AutoCDC metadata column appended to every SCD1 target table. Use
+   * DDL fragment for the reserved CDC metadata column appended to every SCD1 target table. Use
    * inside a `CREATE TABLE` statement, for example:
-   *   `CREATE TABLE t (id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)`
+   *   `CREATE TABLE t (id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)`
    *
    * Assumes sequence type is BIGINT (Long).
    */
-  protected val cdcMetadataDdl: String = {
+  protected val scd1MetadataDdl: String = {
     val col = AutoCdcReservedNames.cdcMetadataColName
     val del = Scd1BatchProcessor.cdcDeleteSequenceFieldName
     val ups = Scd1BatchProcessor.cdcUpsertSequenceFieldName
     s"$col STRUCT<$del:BIGINT,$ups:BIGINT> NOT NULL"
   }
 
+  /**
+   * DDL fragment for the reserved framework columns appended to every SCD2 target table: the
+   * visible interval bounds `__START_AT` / `__END_AT` plus the CDC metadata column. Encapsulates
+   * the full SCD2 reserved-column set (the analog of [[scd1MetadataDdl]], which for SCD1 is just
+   * the metadata column). Use inside a `CREATE TABLE` statement, for example:
+   *   `CREATE TABLE t (id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)`
+   *
+   * Assumes sequence type is BIGINT (Long).
+   */
+  protected val scd2MetadataDdl: String = {
+    val col = AutoCdcReservedNames.cdcMetadataColName
+    val startAt = Scd2BatchProcessor.startAtColName
+    val endAt = Scd2BatchProcessor.endAtColName
+    val recordStartAt = Scd2BatchProcessor.recordStartAtFieldName
+    s"$startAt BIGINT, $endAt BIGINT, $col STRUCT<$recordStartAt:BIGINT> NOT NULL"
+  }
+
   /**
    * Insert a pre-existing row into a target table, populating the CDC metadata struct so the
    * row looks as if a previous AutoCDC run upserted it at sequencing version [[sequence]].

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala → sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala

@@ -22,12 +22,12 @@ import scala.util.Random
 
 import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
 import org.apache.spark.sql.functions
-import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, UnqualifiedColumnName}
-import org.apache.spark.sql.pipelines.graph.AutoCdcScd1OutOfOrderConvergenceSuite.SourceRow
+import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.graph.AutoCdcOutOfOrderConvergenceSuite.SourceRow
 import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
 import org.apache.spark.sql.test.SharedSparkSession
 
-object AutoCdcScd1OutOfOrderConvergenceSuite {
+object AutoCdcOutOfOrderConvergenceSuite {
   /**
    * A single CDC event in the source stream.
    *
@@ -49,11 +49,11 @@ object AutoCdcScd1OutOfOrderConvergenceSuite {
 }
 
 /**
- * Differential test for the SCD1 AutoCDC merge's order-invariance property: feeding the same
- * randomly-generated CDC event stream as a single sorted micro-batch and as several shuffled
- * micro-batches must converge to the same target table contents.
+ * Differential test for the AutoCDC merge's order-invariance property, for both SCD Type 1 and
+ * SCD Type 2: feeding the same randomly-generated CDC event stream as a single sorted micro-batch
+ * and as several shuffled micro-batches must converge to the same target table contents.
  */
-class AutoCdcScd1OutOfOrderConvergenceSuite
+class AutoCdcOutOfOrderConvergenceSuite
     extends ExecutionTest
     with SharedSparkSession
     with AutoCdcGraphExecutionTestMixin {
@@ -77,7 +77,7 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
   // by setting this property. Mirrors the convention used by `RandomDataGenerator` and other Spark
   // suites that expose tunables via `spark.sql.test.<feature>` system properties.
   private val seedSystemProperty: String =
-    "spark.sql.test.autocdc.scd1OutOfOrderConvergenceSeed"
+    "spark.sql.test.autocdc.outOfOrderConvergenceSeed"
 
   private def resolveTestSeed(): Long = {
     Option(System.getProperty(seedSystemProperty)).map(_.toLong).getOrElse(Random.nextLong())
@@ -125,10 +125,11 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
     events.sortBy(_.sequence).toSeq
   }
 
-  /** Build a pipeline context with a single SCD1 AutoCDC flow reading from `stream`. */
+  /** Build a pipeline context with a single AutoCDC flow of `scdType` reading from `stream`. */
   private def buildPipelineContext(
       targetTable: String,
-      stream: MemoryStream[SourceRow]): TestGraphRegistrationContext = {
+      stream: MemoryStream[SourceRow],
+      scdType: ScdType): TestGraphRegistrationContext = {
     new TestGraphRegistrationContext(spark) {
       registerTable(targetTable, catalog = Some(catalog), database = Some(namespace))
       registerFlow(autoCdcFlow(
@@ -140,20 +141,31 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
         deleteCondition = Some(functions.col(isDeleteColumn) === true),
         columnSelection = Some(ColumnSelection.ExcludeColumns(
           Seq(UnqualifiedColumnName(isDeleteColumn))
-        ))
+        )),
+        scdType = scdType
       ))
     }
   }
 
-  private def createTargetTable(targetTable: String): Unit = {
+  /**
+   * DDL fragment for the SCD-type-specific reserved columns a target table carries after the
+   * user-selected data columns: the CDC metadata column for SCD1, and the interval bounds plus
+   * metadata column for SCD2. The sequencing type is BIGINT here.
+   */
+  private def reservedColumnsDdl(scdType: ScdType): String = scdType match {
+    case ScdType.Type1 => scd1MetadataDdl
+    case ScdType.Type2 => scd2MetadataDdl
+  }
+
+  private def createTargetTable(targetTable: String, scdType: ScdType): Unit = {
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.$targetTable (" +
       s"`$keyColumn` INT NOT NULL, " +
       s"`$nameColumn` STRING, " +
       s"`$amountColumn` INT, " +
       s"`$activeColumn` BOOLEAN, " +
       s"`$sequenceColumn` BIGINT NOT NULL, " +
-      s"$cdcMetadataDdl)"
+      s"${reservedColumnsDdl(scdType)})"
     )
   }
 
@@ -164,7 +176,7 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
     )
   }
 
-  private def runConvergenceTest(seed: Long): Unit = {
+  private def runConvergenceTest(seed: Long, scdType: ScdType): Unit = {
     val session = spark
     import session.implicits._
 
@@ -173,22 +185,28 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
     val shuffledEventStream = rand.shuffle(sortedEventStream)
 
     withClue(
-      s"\nseed=$seed (rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
+      s"\nscdType=${scdType.label} seed=$seed " +
+      s"(rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
       s"events (${sortedEventStream.size} total, sorted by sequence):\n" +
       sortedEventStream.map(r => s"  $r").mkString("\n") + "\n"
     ) {
-      val inOrderTable = "inorder_target"
-      val outOfOrderTable = "outoforder_target"
-      createTargetTable(inOrderTable)
-      createTargetTable(outOfOrderTable)
+      // Table names are scd-type-suffixed purely for readability: the SCD1 and SCD2 tests run as
+      // separate test cases and the mixin's afterEach resets the catalog between them, so they
+      // could not collide even with identical names; the suffix just makes a failing run's tables
+      // self-identifying.
+      val suffix = scdType.label.toLowerCase(java.util.Locale.ROOT)
+      val inOrderTable = s"inorder_target_$suffix"
+      val outOfOrderTable = s"outoforder_target_$suffix"
+      createTargetTable(inOrderTable, scdType)
+      createTargetTable(outOfOrderTable, scdType)
 
       val inOrderStream = MemoryStream[SourceRow]
-      val inOrderCtx = buildPipelineContext(inOrderTable, inOrderStream)
+      val inOrderCtx = buildPipelineContext(inOrderTable, inOrderStream, scdType)
       inOrderStream.addData(sortedEventStream: _*)
       runPipeline(inOrderCtx)
 
       val outOfOrderStream = MemoryStream[SourceRow]
-      val outOfOrderCtx = buildPipelineContext(outOfOrderTable, outOfOrderStream)
+      val outOfOrderCtx = buildPipelineContext(outOfOrderTable, outOfOrderStream, scdType)
       val totalEvents = shuffledEventStream.size
       (0 until numOutOfOrderBatches).foreach { batchIndex =>
         val batchStart = batchIndex * totalEvents / numOutOfOrderBatches
@@ -197,11 +215,18 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
         runPipeline(outOfOrderCtx)
       }
 
+      // Only the user-visible target must converge. The auxiliary tables legitimately differ by
+      // arrival order (e.g. deletedByBatchId stamps and cross-batch GC depend on how events are
+      // batched), so they are not compared.
       assertTargetsConverge(inOrderTable, outOfOrderTable)
     }
   }
 
   test("SCD1 merge converges across micro-batch shuffling for randomly generated CDC events") {
-    runConvergenceTest(resolveTestSeed())
+    runConvergenceTest(resolveTestSeed(), ScdType.Type1)
+  }
+
+  test("SCD2 merge converges across micro-batch shuffling for randomly generated CDC events") {
+    runConvergenceTest(resolveTestSeed(), ScdType.Type2)
   }
 }