[SPARK-58313][SDP] Validate SCD2 track-history columns at AutoCDC flow construction

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

@@ -21,6 +21,7 @@ import org.apache.spark.SparkException
 import org.apache.spark.sql.{functions => F}
 import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution, caseSensitiveResolution}
 import org.apache.spark.sql.catalyst.expressions.{CreateMap, If, Literal, RaiseError}
 import org.apache.spark.sql.catalyst.util.QuotingUtils
 import org.apache.spark.sql.classic.{DataFrame, ExpressionUtils}
@@ -922,28 +923,12 @@ case class Scd2BatchProcessor(
    * the eligible user-data columns (those not in [[ChangeArgs.keys]] or the framework
    * reserved set) filtered through [[ChangeArgs.trackHistorySelection]].
    */
-  private def computeTrackedHistoryColumns(df: DataFrame): Seq[String] = {
-    val conf = df.sparkSession.sessionState.conf
-    val resolver = conf.resolver
-
-    val keyColNames = changeArgs.keys.map(_.name)
-    val reservedColNames = Scd2BatchProcessor.reservedFrameworkColNames
-
-    val eligibleSchema = StructType(df.schema.fields.filterNot { field =>
-      reservedColNames.exists(resolver(_, field.name)) ||
-        keyColNames.exists(resolver(_, field.name))
-    })
-
-    ColumnSelection
-      .applyToSchema(
-        schemaName = "trackHistorySelection",
-        schema = eligibleSchema,
-        columnSelection = changeArgs.trackHistorySelection,
-        caseSensitive = conf.caseSensitiveAnalysis
-      )
-      .fieldNames
-      .toImmutableArraySeq
-  }
+  private def computeTrackedHistoryColumns(df: DataFrame): Seq[String] =
+    Scd2BatchProcessor.computeTrackedHistoryColumns(
+      schema = df.schema,
+      changeArgs = changeArgs,
+      caseSensitive = df.sparkSession.sessionState.conf.caseSensitiveAnalysis
+    )
 
   /**
    * Tag each post-reconciliation row with [[Scd2BatchProcessor.shouldRouteToAuxTableColName]]:
@@ -1410,6 +1395,43 @@ object Scd2BatchProcessor {
     AutoCdcReservedNames.cdcMetadataColName
   )
 
+  /**
+   * Resolve [[ChangeArgs.trackHistorySelection]] against `schema` and return the field names of
+   * the history-tracking columns: the eligible user-data columns (those that are neither
+   * [[ChangeArgs.keys]] nor framework reserved columns) filtered through the selection.
+   *
+   * This is the single source of truth for which columns define an SCD2 run. It is called both
+   * per-microbatch (against the reconciled dataframe's schema) and at
+   * [[org.apache.spark.sql.pipelines.graph.AutoCdcMergeFlow]] construction time (against the
+   * user-selected source schema), so an unresolvable or ineligible selection fails fast with a
+   * user-actionable [[org.apache.spark.sql.AnalysisException]] instead of surfacing mid-stream.
+   *
+   * Throws `AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA` if the selection references a column that is not
+   * an eligible history-tracking column in `schema` (i.e. absent, or a key/framework column).
+   */
+  private[pipelines] def computeTrackedHistoryColumns(
+      schema: StructType,
+      changeArgs: ChangeArgs,
+      caseSensitive: Boolean): Seq[String] = {
+    val resolver = if (caseSensitive) caseSensitiveResolution else caseInsensitiveResolution
+    val keyColNames = changeArgs.keys.map(_.name)
+
+    val eligibleSchema = StructType(schema.fields.filterNot { field =>
+      reservedFrameworkColNames.exists(resolver(_, field.name)) ||
+        keyColNames.exists(resolver(_, field.name))
+    })
+
+    ColumnSelection
+      .applyToSchema(
+        schemaName = "trackHistorySelection",
+        schema = eligibleSchema,
+        columnSelection = changeArgs.trackHistorySelection,
+        caseSensitive = caseSensitive
+      )
+      .fieldNames
+      .toImmutableArraySeq
+  }
+
   /**
    * Name of temporary column projected onto microbatch to compute the min sequencing value per
    * key within the microbatch.

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

@@ -267,6 +267,9 @@ class AutoCdcMergeFlow(
     // AutoCDC flows require all key columns to be present in the user-selected source schema,
     // so that they survive into the target table where SCD reconciliation needs them.
     requireKeysPresentInSelectedSchema(selectedSchema)
+    // SCD2 flows may specify history-tracking columns; validate they resolve to eligible columns
+    // of the selected schema at construction time, rather than failing mid-stream on first batch.
+    requireTrackHistoryColumnsResolvableInSelectedSchema(selectedSchema)
     selectedSchema
   }
 
@@ -417,4 +420,26 @@ class AutoCdcMergeFlow(
         )
       }
   }
+
+  /**
+   * Validate that this flow's [[ChangeArgs.trackHistorySelection]] (SCD2 `TRACK HISTORY ON ...`)
+   * resolves against the user-selected source schema at construction time. Without this, an
+   * unresolvable or ineligible (key/framework) tracking column would only surface when the first
+   * microbatch runs reconciliation, deep inside the SCD2 batch processor.
+   *
+   * Delegates to [[Scd2BatchProcessor.computeTrackedHistoryColumns]] -- the same resolution used at
+   * runtime -- so the two can never diverge; it throws `AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA` on an
+   * unresolvable selection. `trackHistorySelection` is `None` for SCD1 (enforced by [[ChangeArgs]])
+   * and for SCD2 flows that do not restrict tracking, in which case resolution is a no-op.
+   */
+  private def requireTrackHistoryColumnsResolvableInSelectedSchema(
+      selectedSchema: StructType): Unit = {
+    if (changeArgs.trackHistorySelection.isDefined) {
+      Scd2BatchProcessor.computeTrackedHistoryColumns(
+        schema = selectedSchema,
+        changeArgs = changeArgs,
+        caseSensitive = spark.sessionState.conf.caseSensitiveAnalysis
+      )
+    }
+  }
 }

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

@@ -165,13 +165,15 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession {
       keys: Seq[UnqualifiedColumnName] = Seq(UnqualifiedColumnName("id")),
       sequencing: Column = F.col("seq"),
       storedAsScdType: ScdType = ScdType.Type1,
-      columnSelection: Option[ColumnSelection] = None): AutoCdcMergeFlow = {
+      columnSelection: Option[ColumnSelection] = None,
+      trackHistorySelection: Option[ColumnSelection] = None): AutoCdcMergeFlow = {
     val flow = newAutoCdcFlow(
       changeArgs = ChangeArgs(
         keys = keys,
         sequencing = sequencing,
         storedAsScdType = storedAsScdType,
-        columnSelection = columnSelection
+        columnSelection = columnSelection,
+        trackHistorySelection = trackHistorySelection
       )
     )
     new AutoCdcMergeFlow(flow, successfulFuncResult(sourceDf))
@@ -767,4 +769,151 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession {
       )
     )
   }
+
+  // ===========================================================================================
+  // AutoCdcMergeFlow track-history validation tests
+  //
+  // SCD2 `TRACK HISTORY ON (...)` populates trackHistorySelection. These tests lock in that an
+  // unresolvable or ineligible (key / dropped-by-column-selection) tracking column is rejected at
+  // flow construction rather than deferring to the first microbatch's reconciliation, mirroring
+  // the keys-presence validator above. A resolvable selection passes the check and the flow
+  // constructs successfully.
+  // ===========================================================================================
+
+  test(
+    "an SCD2 flow tracking a non-existent column is rejected at construction"
+  ) {
+    // Eligible tracking columns from the 3-column source (id, name, seq), less the key `id`, are
+    // {name, seq}. `missing` is absent, so resolution against trackHistorySelection fails.
+    checkError(
+      exception = intercept[AnalysisException] {
+        newAutoCdcMergeFlow(
+          sourceDf = threeColumnSourceDf(),
+          storedAsScdType = ScdType.Type2,
+          trackHistorySelection = Some(
+            ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("missing")))
+          )
+        )
+      },
+      condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA",
+      parameters = Map(
+        "caseSensitivity" -> CaseSensitivityLabels.CaseInsensitive,
+        "schemaName" -> "trackHistorySelection",
+        "missingColumns" -> "missing",
+        "availableColumns" -> "name, seq"
+      )
+    )
+  }
+
+  test(
+    "an SCD2 flow tracking a key column is rejected at construction (ineligible)"
+  ) {
+    // A key is never an eligible history-tracking column, so it is absent from the eligible
+    // schema {name, seq} and resolution fails -- surfacing the misconfiguration eagerly.
+    checkError(
+      exception = intercept[AnalysisException] {
+        newAutoCdcMergeFlow(
+          sourceDf = threeColumnSourceDf(),
+          storedAsScdType = ScdType.Type2,
+          trackHistorySelection = Some(
+            ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("id")))
+          )
+        )
+      },
+      condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA",
+      parameters = Map(
+        "caseSensitivity" -> CaseSensitivityLabels.CaseInsensitive,
+        "schemaName" -> "trackHistorySelection",
+        "missingColumns" -> "id",
+        "availableColumns" -> "name, seq"
+      )
+    )
+  }
+
+  test(
+    "an SCD2 flow tracking a column dropped by columnSelection is rejected"
+  ) {
+    // `name` exists in the source but is excluded from the selected schema, so it is not an
+    // eligible tracking column. Eligible columns are then just {seq}.
+    checkError(
+      exception = intercept[AnalysisException] {
+        newAutoCdcMergeFlow(
+          sourceDf = threeColumnSourceDf(),
+          storedAsScdType = ScdType.Type2,
+          columnSelection = Some(
+            ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("name")))
+          ),
+          trackHistorySelection = Some(
+            ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))
+          )
+        )
+      },
+      condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA",
+      parameters = Map(
+        "caseSensitivity" -> CaseSensitivityLabels.CaseInsensitive,
+        "schemaName" -> "trackHistorySelection",
+        "missingColumns" -> "name",
+        "availableColumns" -> "seq"
+      )
+    )
+  }
+
+  test(
+    "an SCD2 flow with a resolvable track-history selection passes the check"
+  ) {
+    // `name` is an eligible tracking column, so the construction-time check passes and the flow
+    // constructs successfully, resolving its SCD2 schema (no AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA).
+    val resolvedFlow = newAutoCdcMergeFlow(
+      sourceDf = threeColumnSourceDf(),
+      storedAsScdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))
+      )
+    )
+    assert(resolvedFlow.schema.fieldNames.contains(AutoCdcReservedNames.cdcMetadataColName))
+  }
+
+  test(
+    "track-history validation respects case-insensitive analysis"
+  ) {
+    // With caseSensitive=false, `NAME` resolves to the eligible `name`, so the check passes and
+    // the flow constructs successfully.
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+      val resolvedFlow = newAutoCdcMergeFlow(
+        sourceDf = threeColumnSourceDf(),
+        storedAsScdType = ScdType.Type2,
+        trackHistorySelection = Some(
+          ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME")))
+        )
+      )
+      assert(resolvedFlow.schema.fieldNames.contains(AutoCdcReservedNames.cdcMetadataColName))
+    }
+  }
+
+  test(
+    "track-history validation respects case-sensitive analysis"
+  ) {
+    // With caseSensitive=true, `NAME` is a distinct identifier from the eligible `name` and does
+    // not resolve, so the construction-time check rejects it.
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      checkError(
+        exception = intercept[AnalysisException] {
+          newAutoCdcMergeFlow(
+            sourceDf = threeColumnSourceDf(),
+            storedAsScdType = ScdType.Type2,
+            trackHistorySelection = Some(
+              ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME")))
+            )
+          )
+        },
+        condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA",
+        parameters = Map(
+          "caseSensitivity" -> CaseSensitivityLabels.CaseSensitive,
+          "schemaName" -> "trackHistorySelection",
+          "missingColumns" -> "NAME",
+          "availableColumns" -> "name, seq"
+        )
+      )
+    }
+  }
 }