From a246dd20a7dc7e1c2a1527891d119c681e7d6beb Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Fri, 24 Jul 2026 04:56:16 +0000 Subject: [PATCH 1/3] [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 | 25 +++ .../pipelines/autocdc/AutoCdcFlowSuite.scala | 158 +++++++++++++++++- 3 files changed, 226 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 157d2d5788b6d..b0cefc74ef112 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 @@ -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,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 31444f0c15f3d..3d65042843872 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 @@ -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 (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 98c31a3020ef5..7b6d255b3b23c 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)) @@ -767,4 +769,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 7cbcdac00810e0e2e744e903fea981f4cce02495 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 02:00:43 +0000 Subject: [PATCH 2/3] [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 b0cefc74ef112..fbf9bc15127ec 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 3d65042843872..410b0486c787b 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 @@ -425,7 +425,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 7b6d255b3b23c..37eb6597f81c5 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 @@ -771,7 +771,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 @@ -781,7 +781,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. @@ -806,7 +806,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. @@ -831,7 +831,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}. @@ -859,7 +859,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 @@ -877,7 +877,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. @@ -896,7 +896,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. From 34975fbbb9d13b9dae7d32a437d007b6e3afeef3 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 04:01:25 +0000 Subject: [PATCH 3/3] [SPARK-58313][SDP] Reconcile track-history tests with merged SCD2 schema support SPARK-58319 (now on master) removed the AUTOCDC_SCD2_NOT_SUPPORTED gate, so an SCD2 flow with a resolvable track-history selection now constructs successfully instead of throwing. Update the two tests that expected that gate to assert the flow builds and its SCD2 schema resolves, and refresh the stale section comment. The case-sensitive rejection test is unchanged: it still expects AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA from the track-history check itself. Co-authored-by: Opus 4.8 --- .../pipelines/autocdc/AutoCdcFlowSuite.scala | 43 ++++++++----------- 1 file changed, 19 insertions(+), 24 deletions(-) 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 37eb6597f81c5..5769b264993bf 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 @@ -776,8 +776,8 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { // 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. + // the keys-presence validator above. A resolvable selection passes the check and the flow + // constructs successfully. // =========================================================================================== test( @@ -861,37 +861,32 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { 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; 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"))) - ) + // `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(ex.getCondition == "AUTOCDC_SCD2_NOT_SUPPORTED") + ) + 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 - // construction proceeds to the SCD2-not-supported gate. + // the flow constructs successfully. withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { - val ex = intercept[AnalysisException] { - newAutoCdcMergeFlow( - sourceDf = threeColumnSourceDf(), - storedAsScdType = ScdType.Type2, - trackHistorySelection = Some( - ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME"))) - ) + val resolvedFlow = newAutoCdcMergeFlow( + sourceDf = threeColumnSourceDf(), + storedAsScdType = ScdType.Type2, + trackHistorySelection = Some( + ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME"))) ) - } - assert(ex.getCondition == "AUTOCDC_SCD2_NOT_SUPPORTED") + ) + assert(resolvedFlow.schema.fieldNames.contains(AutoCdcReservedNames.cdcMetadataColName)) } }