From 048801f0cdd880dc5509e7336f1e26cf25e28dac Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Fri, 24 Jul 2026 04:56:16 +0000 Subject: [PATCH 1/2] [SPARK-58313][SQL] Validate SCD2 track-history columns at AutoCDC flow construction An SCD2 AutoCDC flow's history-tracking columns (`TRACK HISTORY ON ...`, i.e. ChangeArgs.trackHistorySelection) were only validated when the first microbatch ran reconciliation. An unresolvable or ineligible tracking column (one that is absent, a key, a framework column, or dropped by the column selection) therefore surfaced mid-stream, deep inside Scd2BatchProcessor, rather than eagerly at flow construction. Validate the selection at AutoCdcMergeFlow construction time, mirroring the existing key-presence check. The eligibility + resolution logic is extracted into a schema-based Scd2BatchProcessor.computeTrackedHistoryColumns helper that both the runtime path and the new construction-time validator call, so the two can never diverge. An unresolvable selection fails with the existing AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA error; the check is a no-op when trackHistorySelection is None (all SCD1 flows and unrestricted SCD2 flows). Adds tests in AutoCdcFlowSuite. Co-authored-by: Isaac --- .../autocdc/Scd2BatchProcessor.scala | 67 +++++--- .../spark/sql/pipelines/graph/Flow.scala | 26 +++ .../pipelines/autocdc/AutoCdcFlowSuite.scala | 158 +++++++++++++++++- 3 files changed, 227 insertions(+), 24 deletions(-) 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 6e5af84216b58..6891843fd867f 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 @@ -20,6 +20,7 @@ package org.apache.spark.sql.pipelines.autocdc import org.apache.spark.SparkException import org.apache.spark.sql.{functions => F} import org.apache.spark.sql.Column +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} @@ -920,28 +921,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 + ) } /** @@ -1099,6 +1084,44 @@ 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 + * (SPARK-58313). + * + * 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. diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala index dd4d1556afbf8..388a3c3e53bda 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala +++ b/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} @@ -266,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 } @@ -395,4 +399,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 (SPARK-58313). + * + * 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 + ) + } + } } 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 32374f8ecb048..24608ae725ab6 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 @@ -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)) @@ -565,4 +567,156 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { ) ) } + + // =========================================================================================== + // AutoCdcMergeFlow track-history validation tests (SPARK-58313) + // + // 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; construction then + // proceeds to force `schema`, which currently throws AUTOCDC_SCD2_NOT_SUPPORTED. + // =========================================================================================== + + test( + "SPARK-58313: 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( + "SPARK-58313: 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( + "SPARK-58313: 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( + "SPARK-58313: an SCD2 flow with a resolvable track-history selection passes the check" + ) { + // `name` is an eligible tracking column, so the construction-time check passes; construction + // then forces `schema`, which throws AUTOCDC_SCD2_NOT_SUPPORTED. Observing that error (rather + // than AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA) confirms the track-history check did NOT fire. + val ex = intercept[AnalysisException] { + newAutoCdcMergeFlow( + sourceDf = threeColumnSourceDf(), + storedAsScdType = ScdType.Type2, + trackHistorySelection = Some( + ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))) + ) + ) + } + assert(ex.getCondition == "AUTOCDC_SCD2_NOT_SUPPORTED") + } + + test( + "SPARK-58313: track-history validation respects case-insensitive analysis" + ) { + // With caseSensitive=false, `NAME` resolves to the eligible `name`, so the check passes and + // construction proceeds to the SCD2-not-supported gate. + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val ex = intercept[AnalysisException] { + newAutoCdcMergeFlow( + sourceDf = threeColumnSourceDf(), + storedAsScdType = ScdType.Type2, + trackHistorySelection = Some( + ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME"))) + ) + ) + } + assert(ex.getCondition == "AUTOCDC_SCD2_NOT_SUPPORTED") + } + } + + test( + "SPARK-58313: 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" + ) + ) + } + } } From bfddfc1010313278504dae92297ce6966437883f Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 02:00:43 +0000 Subject: [PATCH 2/2] [SPARK-58313][SDP] Address review: drop JIRA references from comments and test names Remove the SPARK-58313 references from the source comments and test names/section header, consistent with the other AutoCDC PRs (the merged PR records provenance). Threading conf.resolver through the shared ColumnSelection.applyToSchema (rather than deriving a resolver from a caseSensitive boolean) is left as a follow-up refactor, SPARK-58347, since it reworks a boolean/getFieldIndex-based API shared across the SCD1 and SCD2 code paths and is out of scope here. Co-authored-by: Opus 4.8 --- .../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 3 +-- .../apache/spark/sql/pipelines/graph/Flow.scala | 2 +- .../sql/pipelines/autocdc/AutoCdcFlowSuite.scala | 14 +++++++------- 3 files changed, 9 insertions(+), 10 deletions(-) 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 40fb6bd0feadf..6669ee35a2870 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 @@ -1404,8 +1404,7 @@ object Scd2BatchProcessor { * 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 - * (SPARK-58313). + * 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). diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala index 388a3c3e53bda..41f5adf33871a 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala @@ -404,7 +404,7 @@ 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 (SPARK-58313). + * 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 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 24608ae725ab6..9cb0b6dd4aa59 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 @@ -569,7 +569,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } // =========================================================================================== - // AutoCdcMergeFlow track-history validation tests (SPARK-58313) + // 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 @@ -579,7 +579,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { // =========================================================================================== test( - "SPARK-58313: an SCD2 flow tracking a non-existent column is rejected at construction" + "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. @@ -604,7 +604,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-58313: an SCD2 flow tracking a key column is rejected at construction (ineligible)" + "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. @@ -629,7 +629,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-58313: an SCD2 flow tracking a column dropped by columnSelection is rejected" + "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}. @@ -657,7 +657,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-58313: an SCD2 flow with a resolvable track-history selection passes the check" + "an SCD2 flow with a resolvable track-history selection passes the check" ) { // `name` is an eligible tracking column, so the construction-time check passes; construction // then forces `schema`, which throws AUTOCDC_SCD2_NOT_SUPPORTED. Observing that error (rather @@ -675,7 +675,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-58313: track-history validation respects case-insensitive analysis" + "track-history validation respects case-insensitive analysis" ) { // With caseSensitive=false, `NAME` resolves to the eligible `name`, so the check passes and // construction proceeds to the SCD2-not-supported gate. @@ -694,7 +694,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-58313: track-history validation respects case-sensitive analysis" + "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.