diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 213fdf6efcbc1..bff7ea9bc264a 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/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."
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
index e41f223322bcb..125f6c895f411 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
+++ b/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
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
index d83526d14ab89..c36d824bdf4a6 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
+++ b/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()
+  }
+}
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala
index 8251780524a2d..9455e0a724bfc 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala
+++ b/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(
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
index d31b9383354cb..a3d26e6f47e55 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
+++ b/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"
   ) {
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcGraphExecutionTestMixin.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcGraphExecutionTestMixin.scala
index 9793605e884d9..305b03e0bdcfb 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcGraphExecutionTestMixin.scala
+++ b/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]].
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
similarity index 74%
rename from sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala
rename to sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
index fb3a179f25c71..f164b2e70f8ac 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala
+++ b/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,12 +141,23 @@ 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, " +
@@ -153,7 +165,7 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
       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)
   }
 }
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
index 10b8d0dfc5ae5..422b95b3c2678 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
@@ -45,7 +45,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Single MemoryStream reused across both pipeline runs so the streaming checkpoint can
@@ -87,7 +87,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Single MemoryStream reused across both runs so the streaming checkpoint can resume.
@@ -128,7 +128,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
     // leading column.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(name STRING, id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(name STRING, id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(String, Int, Long)]
@@ -163,7 +163,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
       s"(value STRING, id INT NOT NULL, region STRING NOT NULL, " +
-      s"version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(String, Int, String, Long)]
@@ -188,7 +188,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, Long)]
@@ -213,7 +213,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Single MemoryStream reused across both runs so the streaming checkpoint can resume.
@@ -267,7 +267,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
     // would have to be escaped by doubling, but none of these names contain one.
     val targetTableDdl = keyNames
       .map(name => s"`$name` STRING NOT NULL")
-      .mkString(", ") + s", version BIGINT NOT NULL, $cdcMetadataDdl"
+      .mkString(", ") + s", version BIGINT NOT NULL, $scd1MetadataDdl"
     spark.sql(s"CREATE TABLE $catalog.$namespace.target ($targetTableDdl)")
 
     // The AutoCDC API runs every key through `UnqualifiedColumnName.apply`, which calls
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
index 94ba7e20aed1f..549c79116d361 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1FullRefreshSuite.scala
@@ -43,7 +43,7 @@ class AutoCdcScd1FullRefreshSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Run #1: populate target + auxiliary table.
@@ -101,7 +101,7 @@ class AutoCdcScd1FullRefreshSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Run #1: delete at seq=10 sets a high watermark in the auxiliary table.
@@ -162,11 +162,11 @@ class AutoCdcScd1FullRefreshSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.t_a " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.t_b " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // streamA is replaced across runs because t_a is full-refreshed in run #2 (its streaming
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
index 1e97ab2c2cd20..1d27b7d69dab2 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
@@ -48,7 +48,7 @@ class AutoCdcScd1KeyDriftSuite
     // declaration differs between the two pipelines.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, region STRING NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, region STRING NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1 declares one key (`id`). Aux table is created with schema (id, _cdc_metadata).
@@ -83,7 +83,7 @@ class AutoCdcScd1KeyDriftSuite
     "KEY_SCHEMA_DRIFT") {
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(region STRING NOT NULL, id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(region STRING NOT NULL, id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1 declares two keys [region, id]. Without strict-equality, the dropped `region`
@@ -119,7 +119,7 @@ class AutoCdcScd1KeyDriftSuite
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
       s"(id INT NOT NULL, region STRING NOT NULL, country STRING NOT NULL, " +
-      s"version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1 declares [id, region].
@@ -156,10 +156,10 @@ class AutoCdcScd1KeyDriftSuite
     "triggers KEY_SCHEMA_DRIFT") {
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
-      s"""CREATE TABLE ${auxTableNameFor("target")} (id BIGINT NOT NULL, $cdcMetadataDdl) """ +
+      s"""CREATE TABLE ${auxTableNameFor("target")} (id BIGINT NOT NULL, $scd1MetadataDdl) """ +
       s"""TBLPROPERTIES ('${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '["id"]')"""
     )
 
@@ -184,7 +184,7 @@ class AutoCdcScd1KeyDriftSuite
   test("a composite key reorder ([a,b] -> [b,a]) does NOT trigger drift validation") {
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(a INT NOT NULL, b STRING NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(a INT NOT NULL, b STRING NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1 declares keys [a, b] (in that order). Drift validation is order-independent:
@@ -209,7 +209,7 @@ class AutoCdcScd1KeyDriftSuite
     // Target's `id` is nullable so the second pipeline's nullable-`id` source is accepted.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1: source carries `id INT NOT NULL` (Scala primitive `Int`), no metadata.
@@ -236,7 +236,7 @@ class AutoCdcScd1KeyDriftSuite
     // adding or removing backticks around the same logical column must NOT be detected as drift.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Long)]
@@ -256,7 +256,7 @@ class AutoCdcScd1KeyDriftSuite
     // pipeline #2's expected keys are matched against the recorded set).
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Long)]
@@ -277,7 +277,7 @@ class AutoCdcScd1KeyDriftSuite
     // `id` are distinct identifiers under that resolver, drift validation must fail.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Long)]
@@ -322,7 +322,7 @@ class AutoCdcScd1KeyDriftSuite
     // validation make the call.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Long)]
@@ -341,10 +341,10 @@ class AutoCdcScd1KeyDriftSuite
     // surface a structured AUTOCDC_INVALID_STATE error rather than silently mis-validating keys.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
-      s"CREATE TABLE ${auxTableNameFor("target")} (id INT NOT NULL, $cdcMetadataDdl)"
+      s"CREATE TABLE ${auxTableNameFor("target")} (id INT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, Long)]
@@ -372,10 +372,10 @@ class AutoCdcScd1KeyDriftSuite
     val malformedKeysArray = "not-a-json-array"
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
-      s"CREATE TABLE ${auxTableNameFor("target")} (id INT NOT NULL, $cdcMetadataDdl) " +
+      s"CREATE TABLE ${auxTableNameFor("target")} (id INT NOT NULL, $scd1MetadataDdl) " +
       s"TBLPROPERTIES ('${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '$malformedKeysArray')"
     )
 
@@ -406,10 +406,10 @@ class AutoCdcScd1KeyDriftSuite
     // validator cannot run without resolving every recorded key first.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
-      s"""CREATE TABLE ${auxTableNameFor("target")} (id INT NOT NULL, $cdcMetadataDdl) """ +
+      s"""CREATE TABLE ${auxTableNameFor("target")} (id INT NOT NULL, $scd1MetadataDdl) """ +
       s"""TBLPROPERTIES ('${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '["region"]')"""
     )
 
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
index f86c592e3cf71..7191f2e60cd08 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
@@ -46,11 +46,11 @@ class AutoCdcScd1MultiPipelineSuite
     // Two distinct target tables created up-front.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.t_a " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.t_b " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1 only knows about `t_a`. Its auxiliary table
@@ -98,7 +98,7 @@ class AutoCdcScd1MultiPipelineSuite
     // Pipeline #1 writes into target `src` via AutoCDC.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.src " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     val stream = MemoryStream[(Int, String, Long)]
     stream.addData((1, "alice", 1L), (2, "bob", 1L))
@@ -139,7 +139,7 @@ class AutoCdcScd1MultiPipelineSuite
     // but share the target table and its auxiliary table.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.shared_target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1: inserts rows with id=1 and id=2 at version=1.
@@ -197,7 +197,7 @@ class AutoCdcScd1MultiPipelineSuite
     // to schema-merge into the target.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.shared_target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1: source DF schema is (id, name, version); inserts id=1 and id=2.
@@ -267,7 +267,7 @@ class AutoCdcScd1MultiPipelineSuite
     // be schema-compatible with the first; only the AutoCDC `keys` differ between flows.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.shared_target " +
-      s"(id INT NOT NULL, name STRING NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Pipeline #1: AutoCDC flow keyed on `id`. Materializes the auxiliary table with schema
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
index fb3094aa2eb45..635e3d93de56d 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
@@ -55,7 +55,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // a NULL email; run #2 emits an upsert with a non-NULL email.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, String, Option[String], Long)]
@@ -94,7 +94,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // is strictly wider. Users must full-refresh the target to change column types.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, age INT, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, age INT, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Int, Long)]
@@ -138,7 +138,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // even when the new type is strictly narrower.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, payload BIGINT, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, payload BIGINT, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Long, Long)]
@@ -180,7 +180,7 @@ class AutoCdcScd1SchemaEvolutionSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Single MemoryStream of (id, name, email, version) shared across runs so the streaming
@@ -228,7 +228,7 @@ class AutoCdcScd1SchemaEvolutionSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Shared (id, name, version) stream so the streaming checkpoint resumes cleanly across
@@ -284,7 +284,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // [[ColumnSelection]] knob rather than the source DF's own schema.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, String, String, Long)]
@@ -330,7 +330,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // schema level (SDP's `SchemaMergingUtils.mergeSchemas` is a union, never a subtraction).
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, String, String, Long)]
@@ -378,7 +378,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // tightening [[ChangeArgs.columnSelection]].
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     // Same `MemoryStream[(Int, String, Option[String], Long)]` shape across runs; runs
@@ -430,7 +430,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
       s"(key INT NOT NULL, version BIGINT NOT NULL, " +
-      s"value STRUCT<a:INT,b:STRUCT<c:INT,d:INT>>, $cdcMetadataDdl)"
+      s"value STRUCT<a:INT,b:STRUCT<c:INT,d:INT>>, $scd1MetadataDdl)"
     )
 
     // Stream is (key, version, a, b_c, b_d). Each run reshapes into different `value`
@@ -485,7 +485,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
       s"(key INT NOT NULL, version BIGINT NOT NULL, " +
-      s"vals ARRAY<STRUCT<a:INT,b:STRUCT<c:INT>>>, $cdcMetadataDdl)"
+      s"vals ARRAY<STRUCT<a:INT,b:STRUCT<c:INT>>>, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, Long, Int, Int, Int)]
@@ -543,7 +543,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
       s"(key INT NOT NULL, version BIGINT NOT NULL, " +
-      s"vals ARRAY<STRUCT<a:INT,b:STRUCT<c:INT,d:INT>>>, $cdcMetadataDdl)"
+      s"vals ARRAY<STRUCT<a:INT,b:STRUCT<c:INT,d:INT>>>, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, Long, Int, Int, Int)]
@@ -600,7 +600,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
       spark.sql(
         s"CREATE TABLE $catalog.$namespace.target " +
-        s"(key INT NOT NULL, version BIGINT NOT NULL, value STRING, $cdcMetadataDdl)"
+        s"(key INT NOT NULL, version BIGINT NOT NULL, value STRING, $scd1MetadataDdl)"
       )
 
       val stream = MemoryStream[(Int, Long, String)]
@@ -648,7 +648,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // resolve `extra` to NULL.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, extra INT, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, extra INT, $scd1MetadataDdl)"
     )
     insertPreloadedRow(
       s"$catalog.$namespace.target",
@@ -684,7 +684,7 @@ class AutoCdcScd1SchemaEvolutionSuite
     // the target to change a column's type.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(key INT NOT NULL, version BIGINT NOT NULL, value TIMESTAMP, $cdcMetadataDdl)"
+      s"(key INT NOT NULL, version BIGINT NOT NULL, value TIMESTAMP, $scd1MetadataDdl)"
     )
 
     val stream1 = MemoryStream[(Int, Long, Timestamp)]
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
index f06b8c4615339..9ea7a41d0372e 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala
@@ -46,7 +46,7 @@ class AutoCdcScd1SinglePipelineSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, String, Long)]
@@ -80,7 +80,7 @@ class AutoCdcScd1SinglePipelineSuite
     // column, drives the deleteCondition, and is excluded from the target projection.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, String, Long, Boolean)]
@@ -122,11 +122,11 @@ class AutoCdcScd1SinglePipelineSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.t_a " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.t_b " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val streamA = MemoryStream[(Int, Long)]
@@ -177,7 +177,7 @@ class AutoCdcScd1SinglePipelineSuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$database.target_no_merge " +
-      s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
 
     val stream = MemoryStream[(Int, Long)]
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
index a5f3a13a012a6..0f777708398b6 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1TargetTableDurabilitySuite.scala
@@ -42,7 +42,7 @@ class AutoCdcScd1TargetTableDurabilitySuite
 
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     insertPreloadedRow(s"$catalog.$namespace.target", "1, 'alice', 5", 5L)
     insertPreloadedRow(s"$catalog.$namespace.target", "2, 'bob', 5", 5L)
@@ -81,7 +81,7 @@ class AutoCdcScd1TargetTableDurabilitySuite
     // Target was populated by some external process; this is the first AutoCDC run.
     spark.sql(
       s"CREATE TABLE $catalog.$namespace.target " +
-      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $cdcMetadataDdl)"
+      s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd1MetadataDdl)"
     )
     insertPreloadedRow(s"$catalog.$namespace.target", "1, 'alice', 1", 1L)
 
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala
new file mode 100644
index 0000000000000..e413399c4033e
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+  AutoCdcReservedNames,
+  ColumnSelection,
+  Scd2BatchProcessor,
+  ScdType,
+  UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end smoke tests for AutoCDC SCD Type 2 flows running within a single pipeline: one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes an SCD2 AutoCDC flow through the
+ * [[Scd2MergeStreamingWrite]] streaming write, and both the target table and the auxiliary
+ * table contents are asserted at the end.
+ *
+ * This exercises the full wiring landed for SCD2: the flow planner routing an SCD2
+ * [[AutoCdcMergeFlow]] to [[Scd2MergeStreamingWrite]], the auxiliary-table materialization, and
+ * the [[org.apache.spark.sql.pipelines.autocdc.Scd2ForeachBatchHandler]] reconciliation.
+ */
+class AutoCdcScd2SinglePipelineSuite
+    extends ExecutionTest
+    with SharedSparkSession
+    with AutoCdcGraphExecutionTestMixin {
+
+  /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+  private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+  /**
+   * DDL for an SCD2 target table with user columns `(id, name, version)` plus the framework
+   * columns `__START_AT` / `__END_AT` (sequencing type BIGINT) and the SCD2 `_cdc_metadata`
+   * struct. `version` is the sequencing column and, unless excluded via a column selection, is
+   * retained as an ordinary user column in the target.
+   */
+  private def createScd2Target(table: String): Unit = {
+    spark.sql(
+      s"CREATE TABLE $table (" +
+      s"id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+    )
+  }
+
+  test("SCD2: an upsert lands an open current record in an empty target table") {
+    val session = spark
+    import session.implicits._
+    createScd2Target(s"$catalog.$namespace.target")
+
+    val stream = MemoryStream[(Int, String, Long)]
+    stream.addData((1, "alice", 10L))
+
+    val ctx = new TestGraphRegistrationContext(spark) {
+      registerTable("target", catalog = Some(catalog), database = Some(namespace))
+      registerFlow(autoCdcFlow(
+        name = "auto_cdc_flow",
+        target = "target",
+        query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+
+    runPipeline(ctx)
+
+    // A single event opens a current record: START_AT = the event sequence, END_AT = null.
+    checkAnswer(
+      spark.table(s"$catalog.$namespace.target"),
+      Seq(Row(1, "alice", 10L, 10L, null, scd2Meta(10L)))
+    )
+  }
+
+  test("SCD2: an update to a key closes the prior record and opens a new one") {
+    val session = spark
+    import session.implicits._
+    createScd2Target(s"$catalog.$namespace.target")
+
+    val stream = MemoryStream[(Int, String, Long)]
+    stream.addData((1, "alice", 10L), (1, "alicia", 20L))
+
+    val ctx = new TestGraphRegistrationContext(spark) {
+      registerTable("target", catalog = Some(catalog), database = Some(namespace))
+      registerFlow(autoCdcFlow(
+        name = "auto_cdc_flow",
+        target = "target",
+        query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+
+    runPipeline(ctx)
+
+    // The first value is closed at the second event's sequence; the second value is open.
+    checkAnswer(
+      spark.table(s"$catalog.$namespace.target"),
+      Seq(
+        Row(1, "alice", 10L, 10L, 20L, scd2Meta(10L)),
+        Row(1, "alicia", 20L, 20L, null, scd2Meta(20L))
+      )
+    )
+  }
+
+  test("SCD2: a delete closes the current record with no open record remaining") {
+    val session = spark
+    import session.implicits._
+    // Target omits `is_delete`: the source carries it as a control column driving the delete
+    // condition, and it is excluded from the target projection.
+    createScd2Target(s"$catalog.$namespace.target")
+
+    val stream = MemoryStream[(Int, String, Long, Boolean)]
+    stream.addData((1, "alice", 10L, false), (1, null, 20L, true))
+
+    val ctx = new TestGraphRegistrationContext(spark) {
+      registerTable("target", catalog = Some(catalog), database = Some(namespace))
+      registerFlow(autoCdcFlow(
+        name = "auto_cdc_flow",
+        target = "target",
+        query = dfFlowFunc(stream.toDF().toDF("id", "name", "version", "is_delete")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        columnSelection = Some(
+          ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("is_delete")))
+        ),
+        deleteCondition = Some(functions.col("is_delete")),
+        scdType = ScdType.Type2
+      ))
+    }
+
+    runPipeline(ctx)
+
+    // The delete closes the open record at the delete's sequence; nothing remains open.
+    checkAnswer(
+      spark.table(s"$catalog.$namespace.target"),
+      Seq(Row(1, "alice", 10L, 10L, 20L, scd2Meta(10L)))
+    )
+  }
+
+  test("SCD2: the auxiliary table is materialized for the target") {
+    val session = spark
+    import session.implicits._
+    createScd2Target(s"$catalog.$namespace.target")
+
+    val stream = MemoryStream[(Int, String, Long)]
+    stream.addData((1, "alice", 10L))
+
+    val ctx = new TestGraphRegistrationContext(spark) {
+      registerTable("target", catalog = Some(catalog), database = Some(namespace))
+      registerFlow(autoCdcFlow(
+        name = "auto_cdc_flow",
+        target = "target",
+        query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+
+    runPipeline(ctx)
+
+    // The SCD2 auxiliary table exists and carries the aux-only deleted-by-batch-id marker column
+    // in addition to the full target row schema (user columns + the framework columns). Assert the
+    // exact field list, via the reserved-name constants, so a rename of any framework column (in
+    // particular the non-prefixed __START_AT / __END_AT, which a prefix check would not catch) is
+    // caught here rather than silently passing a substring match.
+    val auxColumns = spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq
+    assert(
+      auxColumns == Seq(
+        "id",
+        "name",
+        "version",
+        Scd2BatchProcessor.startAtColName,
+        Scd2BatchProcessor.endAtColName,
+        AutoCdcReservedNames.cdcMetadataColName,
+        Scd2BatchProcessor.deletedByBatchIdColName
+      )
+    )
+  }
+}