[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

@@ -1396,15 +1396,16 @@ object Scd2BatchProcessor {
   private[pipelines] 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

@@ -253,6 +253,7 @@ class AutoCdcMergeFlow(
     val funcResult: FlowFunctionResult
 ) extends ResolvedFlow {
   requireReservedPrefixAbsentInSourceColumns()
+  requireReservedFrameworkColumnsAbsentInSourceColumns()
 
   def changeArgs: ChangeArgs = flow.changeArgs
 
@@ -397,6 +398,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

@@ -368,63 +368,42 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession {
     }
   }
 
-  test("AutoCdcMergeFlow.schema lets a selection exclude a source column that collides with " +
-    "a non-prefixed framework column, re-adding it from the framework") {
-    // Two of the three SCD2 framework columns -- __START_AT and __END_AT -- do not carry the
-    // reserved AutoCDC prefix, so a source change feed may legitimately contain columns with
-    // those names. The user excludes them via the column selection; they are dropped from the
-    // user-data portion and the engine's own framework columns are appended in their place, so
-    // each still appears exactly once in the output with the framework's type (the sequencing
-    // type), not the source column's type.
-    //
-    // The third framework column, _cdc_metadata, DOES carry the reserved prefix, so a source
-    // that contains it is rejected outright at flow construction -- it can never reach column
-    // selection and so is deliberately out of scope here. That rejection is covered separately by
-    // "AutoCdcMergeFlow rejects a source df column whose name equals the reserved CDC metadata
-    // column".
-    val session = spark
-    import session.implicits._
-    // Source carries String-typed __START_AT / __END_AT columns alongside the data columns.
-    val sourceDf = MemoryStream[(Int, String, Option[Long], String, String)]
-      .toDS()
-      .toDF(
-        "id",
-        "name",
-        "seq",
-        Scd2BatchProcessor.startAtColName,
-        Scd2BatchProcessor.endAtColName)
+  test("AutoCdcMergeFlow rejects a source column named after a non-prefixed framework column " +
+    "even when a selection would exclude it") {
+    // Behavior change introduced by this PR (SPARK-57251): __START_AT / __END_AT do not carry the
+    // reserved AutoCDC prefix, so before this change a source could legitimately contain columns
+    // with those names and exclude them via the column selection. The new
+    // requireReservedFrameworkColumnsAbsentInSourceColumns guard runs against the RAW source
+    // schema (before column selection is applied), so such a source is now rejected outright at
+    // flow construction -- an ExcludeColumns selection cannot rescue it. Allowing an explicit
+    // opt-out (validating post-selection instead) is tracked separately by SPARK-58325.
+    val sourceDf = sourceDfWithExtraColumns(
+      Scd2BatchProcessor.startAtColName -> StringType,
+      Scd2BatchProcessor.endAtColName -> StringType)
 
-    val resolvedFlow = newAutoCdcMergeFlow(
-      sourceDf = sourceDf,
-      storedAsScdType = ScdType.Type2,
-      columnSelection = Some(
-        ColumnSelection.ExcludeColumns(
-          Seq(
-            UnqualifiedColumnName(Scd2BatchProcessor.startAtColName),
-            UnqualifiedColumnName(Scd2BatchProcessor.endAtColName))
+    checkError(
+      exception = intercept[AnalysisException] {
+        newAutoCdcMergeFlow(
+          sourceDf = sourceDf,
+          storedAsScdType = ScdType.Type2,
+          columnSelection = Some(
+            ColumnSelection.ExcludeColumns(
+              Seq(
+                UnqualifiedColumnName(Scd2BatchProcessor.startAtColName),
+                UnqualifiedColumnName(Scd2BatchProcessor.endAtColName))
+            )
+          )
         )
-      )
-    )
-
-    // Each framework column appears exactly once (the source copy was excluded, the framework
-    // copy appended) ...
-    assert(
-      resolvedFlow.schema.fieldNames.count(_ == Scd2BatchProcessor.startAtColName) == 1)
-    assert(
-      resolvedFlow.schema.fieldNames.count(_ == Scd2BatchProcessor.endAtColName) == 1)
-    // ... and carries the framework (sequencing) type, not the source String type.
-    assert(resolvedFlow.schema(Scd2BatchProcessor.startAtColName).dataType == LongType)
-    assert(resolvedFlow.schema(Scd2BatchProcessor.endAtColName).dataType == LongType)
-    // Full expected shape: the retained data columns followed by the framework columns.
-    assert(
-      resolvedFlow.schema.fieldNames.toSeq ==
-      Seq(
-        "id",
-        "name",
-        "seq",
-        Scd2BatchProcessor.startAtColName,
-        Scd2BatchProcessor.endAtColName,
-        AutoCdcReservedNames.cdcMetadataColName
+      },
+      condition = "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT",
+      sqlState = "42710",
+      parameters = Map(
+        "caseSensitivity" -> CaseSensitivityLabels.CaseInsensitive,
+        "columnName" -> Scd2BatchProcessor.startAtColName,
+        "schemaName" -> "changeDataFeed",
+        "scdType" -> ScdType.Type2.label,
+        "reservedColumnNames" ->
+          Seq(Scd2BatchProcessor.endAtColName, Scd2BatchProcessor.startAtColName).mkString(", ")
       )
     )
   }
@@ -701,6 +680,122 @@ 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 does NOT fire and the flow constructs successfully,
+    // keeping the lowercase column as an ordinary user column in the schema.
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      val nonConflictingName = Scd2BatchProcessor.startAtColName.toLowerCase(Locale.ROOT)
+      val sourceDf = sourceDfWithExtraColumns(nonConflictingName -> StringType)
+
+      val resolvedFlow = newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2)
+      assert(resolvedFlow.schema.fieldNames.contains(nonConflictingName))
+    }
+  }
+
   // ===========================================================================================
   // AutoCdcMergeFlow keys-presence validation tests (requireKeysPresentInSelectedSchema)
   // ===========================================================================================