[SPARK-57251][SDP] Validate SCD2 reserved framework columns at AutoCDC flow construction

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

@@ -331,6 +331,12 @@
     ],
     "sqlState" : "22023"
   },
+  "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT" : {
+    "message" : [
+      "The column `<columnName>` in the <schemaName> schema collides with a reserved AutoCDC <scdType> column name (using <caseSensitivity> column name comparison). The following column names are reserved by AutoCDC and cannot appear in the source: <reservedColumnNames>. Rename or remove the column."
+    ],
+    "sqlState" : "42710"
+  },
   "AUTOCDC_RESERVED_COLUMN_NAME_PREFIX_CONFLICT" : {
     "message" : [
       "The column `<columnName>` in the <schemaName> schema collides with the reserved AutoCDC column name prefix `<reservedColumnNamePrefix>` (using <caseSensitivity> column name comparison). Rename or remove the column."

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

@@ -1085,15 +1085,16 @@ object Scd2BatchProcessor {
   private[autocdc] val endAtColName: String = "__END_AT"
 
   /**
-   * Column names reserved by AutoCDC that will be projected onto the microbatch and
-   * eventually persisted in the target table. If the user's source dataframe contains any of
-   * these columns, SCD2 reconciliation will fail.
+   * Column names reserved by AutoCDC that are projected onto the microbatch and persisted in the
+   * target table. A source dataframe must not contain any of them.
    *
-   * TODO(SPARK-57251): validate at [[AutoCdcMergeFlow]] construction time that the source
-   *   schema and column selection do not collide with these reserved names, so we fail fast
-   *   with a user-actionable error instead of silently overwriting them at preprocess time.
+   * [[startAtColName]] and [[endAtColName]] do NOT carry the reserved
+   * [[AutoCdcReservedNames.prefix]], so a source-column collision with them is not caught by the
+   * prefix-based guard; [[org.apache.spark.sql.pipelines.graph.AutoCdcMergeFlow]] validates the
+   * source schema against the non-prefixed names in this set at construction time, failing fast
+   * instead of silently overwriting them at preprocess time.
    */
-  private val reservedFrameworkColNames: Set[String] = Set(
+  private[pipelines] val reservedFrameworkColNames: Set[String] = Set(
     startAtColName,
     endAtColName,
     AutoCdcReservedNames.cdcMetadataColName

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

@@ -30,6 +30,7 @@ import org.apache.spark.sql.pipelines.autocdc.{
   ChangeArgs,
   ColumnSelection,
   Scd1BatchProcessor,
+  Scd2BatchProcessor,
   ScdType
 }
 import org.apache.spark.sql.types.{DataType, StructField, StructType}
@@ -252,6 +253,7 @@ class AutoCdcMergeFlow(
     val funcResult: FlowFunctionResult
 ) extends ResolvedFlow {
   requireReservedPrefixAbsentInSourceColumns()
+  requireReservedFrameworkColumnsAbsentInSourceColumns()
 
   def changeArgs: ChangeArgs = flow.changeArgs
 
@@ -375,6 +377,45 @@ class AutoCdcMergeFlow(
     }
   }
 
+  /**
+   * Reject a source column that collides with an SCD2 reserved framework column not covered by
+   * [[requireReservedPrefixAbsentInSourceColumns]]: the prefix guard only rejects
+   * [[AutoCdcReservedNames.prefix]] names, but SCD2 also persists the non-prefixed
+   * [[Scd2BatchProcessor.startAtColName]] and [[Scd2BatchProcessor.endAtColName]]. Runs before
+   * [[schema]] is forced so the collision fails fast rather than being silently overwritten
+   * during preprocessing. No-op for SCD1, which has no such columns.
+   */
+  private def requireReservedFrameworkColumnsAbsentInSourceColumns(): Unit = {
+    val resolver = spark.sessionState.conf.resolver
+    val reservedPrefix = AutoCdcReservedNames.prefix
+
+    // Only the non-prefixed reserved names need checking here; prefixed ones are already rejected
+    // by requireReservedPrefixAbsentInSourceColumns.
+    val reservedNames: Set[String] = changeArgs.storedAsScdType match {
+      case ScdType.Type2 =>
+        Scd2BatchProcessor.reservedFrameworkColNames.filterNot(_.startsWith(reservedPrefix))
+      case ScdType.Type1 =>
+        Set.empty
+    }
+
+    df.schema.fieldNames
+      .find(name => reservedNames.exists(resolver(_, name)))
+      .foreach { conflictingColumnName =>
+        throw new AnalysisException(
+          errorClass = "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT",
+          messageParameters = Map(
+            "caseSensitivity" -> CaseSensitivityLabels.of(
+              spark.sessionState.conf.caseSensitiveAnalysis
+            ),
+            "columnName" -> conflictingColumnName,
+            "schemaName" -> "changeDataFeed",
+            "scdType" -> changeArgs.storedAsScdType.label,
+            "reservedColumnNames" -> reservedNames.toSeq.sorted.mkString(", ")
+          )
+        )
+      }
+  }
+
   /**
    * Validate all keys specified in changeArgs are actually present in the user-selected schema.
    */

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

@@ -499,6 +499,125 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession {
     }
   }
 
+  // ===========================================================================================
+  // AutoCdcMergeFlow reserved framework-column (non-prefixed) validation tests
+  //
+  // SCD2 persists framework columns __START_AT / __END_AT that do NOT carry the reserved
+  // AutoCDC prefix, so they are not caught by the prefix guard above. These tests lock in that a
+  // source column colliding with such a name is rejected at construction for SCD2, is allowed
+  // for SCD1 (which reserves no non-prefixed names), and that the check respects case-sensitivity.
+  // ===========================================================================================
+
+  /** The SCD2 reserved framework column names that are not covered by the reserved prefix. */
+  private val nonPrefixedScd2ReservedNames: Seq[String] =
+    Scd2BatchProcessor.reservedFrameworkColNames
+      .filterNot(_.startsWith(AutoCdcReservedNames.prefix))
+      .toSeq
+      .sorted
+
+  test("non-prefixed reserved names exist and are covered by this suite") {
+    // Guards against a future refactor renaming/removing __START_AT / __END_AT without updating
+    // the flow-construction validation: if this set ever empties, the tests below silently
+    // stop exercising anything.
+    assert(
+      nonPrefixedScd2ReservedNames == Seq("__END_AT", "__START_AT"),
+      s"Unexpected non-prefixed SCD2 reserved names: $nonPrefixedScd2ReservedNames"
+    )
+  }
+
+  test(
+    "an SCD2 flow with a source column colliding with a reserved framework column is rejected " +
+    "at construction"
+  ) {
+    nonPrefixedScd2ReservedNames.foreach { reservedName =>
+      val sourceDf = sourceDfWithExtraColumns(reservedName -> StringType)
+
+      checkError(
+        exception = intercept[AnalysisException] {
+          newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2)
+        },
+        condition = "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT",
+        sqlState = "42710",
+        parameters = Map(
+          "caseSensitivity" -> CaseSensitivityLabels.CaseInsensitive,
+          "columnName" -> reservedName,
+          "schemaName" -> "changeDataFeed",
+          "scdType" -> ScdType.Type2.label,
+          "reservedColumnNames" -> nonPrefixedScd2ReservedNames.mkString(", ")
+        )
+      )
+    }
+  }
+
+  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"
+  ) {
+    // SCD1 targets carry no non-prefixed framework columns, so __START_AT / __END_AT are ordinary
+    // user columns there. Construction succeeds and the column survives into the flow schema.
+    nonPrefixedScd2ReservedNames.foreach { reservedName =>
+      val sourceDf = sourceDfWithExtraColumns(reservedName -> StringType)
+      val resolvedFlow = newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type1)
+      assert(resolvedFlow.schema.fieldNames.contains(reservedName))
+    }
+  }
+
+  test(
+    "an uppercase reserved framework-column name is rejected for SCD2 when caseSensitive=false"
+  ) {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+      val conflictingName = Scd2BatchProcessor.startAtColName.toLowerCase(Locale.ROOT)
+      val sourceDf = sourceDfWithExtraColumns(conflictingName -> StringType)
+
+      checkError(
+        exception = intercept[AnalysisException] {
+          newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2)
+        },
+        condition = "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT",
+        sqlState = "42710",
+        parameters = Map(
+          "caseSensitivity" -> CaseSensitivityLabels.CaseInsensitive,
+          "columnName" -> conflictingName,
+          "schemaName" -> "changeDataFeed",
+          "scdType" -> ScdType.Type2.label,
+          "reservedColumnNames" -> nonPrefixedScd2ReservedNames.mkString(", ")
+        )
+      )
+    }
+  }
+
+  test(
+    "a differently-cased reserved framework-column name does not trip the reserved check for " +
+    "SCD2 when caseSensitive=true"
+  ) {
+    // Under case-sensitive analysis, a lowercase variant is a distinct identifier and does not
+    // collide with the reserved (uppercase) framework name, consistent with the prefix guard.
+    // The reserved-name check therefore passes; construction then proceeds to force `schema`,
+    // which throws AUTOCDC_SCD2_NOT_SUPPORTED. Observing that error (rather than the reserved-name
+    // error) confirms the reserved check correctly did NOT fire on the differently-cased column.
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      val nonConflictingName = Scd2BatchProcessor.startAtColName.toLowerCase(Locale.ROOT)
+      val sourceDf = sourceDfWithExtraColumns(nonConflictingName -> StringType)
+
+      val ex = intercept[AnalysisException] {
+        newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2)
+      }
+      assert(ex.getCondition == "AUTOCDC_SCD2_NOT_SUPPORTED")
+    }
+  }
+
   // ===========================================================================================
   // AutoCdcMergeFlow keys-presence validation tests (requireKeysPresentInSelectedSchema)
   // ===========================================================================================