From fd8622cb059d2fb41321a96ed3c3f3fd25d94339 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Fri, 24 Jul 2026 04:24:14 +0000 Subject: [PATCH 1/3] [SPARK-57251][SQL] Validate SCD2 reserved framework columns at AutoCDC flow construction AutoCdcMergeFlow only rejected source columns using the reserved AutoCDC prefix (`__spark_autocdc_`). SCD2 additionally persists the framework columns `__START_AT` and `__END_AT`, which do not carry that prefix, so a colliding source column would be silently overwritten during preprocessing. Add a constructor-time check that rejects source columns colliding (by exact name, resolver-aware) with SCD2's non-prefixed reserved framework columns. It runs before the flow schema is forced, so it surfaces this actionable error ahead of the temporary AUTOCDC_SCD2_NOT_SUPPORTED gate and remains correct once SCD2 support lands. SCD1 targets carry no non-prefixed framework columns, so the check is a no-op for SCD1. Adds the AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT error condition and tests in AutoCdcFlowSuite. Co-authored-by: Isaac --- .../resources/error/error-conditions.json | 6 + .../autocdc/Scd2BatchProcessor.scala | 10 +- .../spark/sql/pipelines/graph/Flow.scala | 47 +++++++ .../pipelines/autocdc/AutoCdcFlowSuite.scala | 121 ++++++++++++++++++ 4 files changed, 180 insertions(+), 4 deletions(-) diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 69577c1ddcfe3..9b578e9fe6bbd 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -331,6 +331,12 @@ ], "sqlState" : "22023" }, + "AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT" : { + "message" : [ + "The column `` in the schema collides with a reserved AutoCDC column name (using column name comparison). The following column names are reserved by AutoCDC and cannot appear in the source: . Rename or remove the column." + ], + "sqlState" : "42710" + }, "AUTOCDC_RESERVED_COLUMN_NAME_PREFIX_CONFLICT" : { "message" : [ "The column `` in the schema collides with the reserved AutoCDC column name prefix `` (using column name comparison). Rename or remove the column." 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..938d143a5db55 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 @@ -1400,11 +1400,13 @@ object Scd2BatchProcessor { * eventually persisted in the target table. If the user's source dataframe contains any of * these columns, SCD2 reconciliation will fail. * - * 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. + * Note [[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 this set at construction time (SPARK-57251) to fail fast with a + * user-actionable error 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 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..a172333e76bfe 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 @@ -253,6 +253,7 @@ class AutoCdcMergeFlow( val funcResult: FlowFunctionResult ) extends ResolvedFlow { requireReservedPrefixAbsentInSourceColumns() + requireReservedFrameworkColumnsAbsentInSourceColumns() def changeArgs: ChangeArgs = flow.changeArgs @@ -397,6 +398,52 @@ class AutoCdcMergeFlow( } } + /** + * Validate that the resolved source dataframe does not contain any column whose name collides + * (by exact name, resolver-aware) with an SCD-type-specific reserved framework column that is + * NOT covered by [[requireReservedPrefixAbsentInSourceColumns]]. + * + * The prefix guard above only rejects names starting with [[AutoCdcReservedNames.prefix]]. + * SCD2 additionally persists the framework columns [[Scd2BatchProcessor.startAtColName]] and + * [[Scd2BatchProcessor.endAtColName]], which do NOT carry that prefix, so a colliding source + * column would otherwise be silently overwritten during preprocessing (SPARK-57251). SCD1 + * targets carry no such non-prefixed framework columns, so this guard is a no-op for SCD1. + * + * Runs in the constructor before [[schema]] is forced, so it surfaces this actionable error + * ahead of the (temporary) [[AUTOCDC_SCD2_NOT_SUPPORTED]] gate, and remains correct once SCD2 + * support lands. + */ + 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. */ 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..1c893f83c60fe 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 @@ -701,6 +701,127 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } } + // =========================================================================================== + // AutoCdcMergeFlow reserved framework-column (non-prefixed) validation tests (SPARK-57251) + // + // 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("SPARK-57251: 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( + "SPARK-57251: 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( + "SPARK-57251: 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( + "SPARK-57251: 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( + "SPARK-57251: 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( + "SPARK-57251: 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) // =========================================================================================== From 003bb6f5783fd765a770d4219e42952d396ae139 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 00:21:31 +0000 Subject: [PATCH 2/3] [SPARK-57251][SDP] Address review: trim comments and drop JIRA references Per review feedback: - Remove the SPARK-57251 references from source comments and test names/section header (the merged PR already records provenance), and condense the surrounding comments to reduce bloat. - Clarify the reservedFrameworkColNames scaladoc to say the flow validates the source schema against the *non-prefixed* names in the set (the prefixed member is handled by requireReservedPrefixAbsentInSourceColumns), for precision. No behavior change; the reserved framework-column check still uses `find` (reporting the first collision), consistent with the sibling prefix guard. Co-authored-by: Opus 4.8 --- .../autocdc/Scd2BatchProcessor.scala | 11 +++++----- .../spark/sql/pipelines/graph/Flow.scala | 21 +++++++------------ .../pipelines/autocdc/AutoCdcFlowSuite.scala | 20 ++++++++---------- 3 files changed, 21 insertions(+), 31 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 938d143a5db55..73855f284baa8 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 @@ -1396,15 +1396,14 @@ 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. * - * Note [[startAtColName]] and [[endAtColName]] do NOT carry the reserved + * [[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 this set at construction time (SPARK-57251) to fail fast with a - * user-actionable error instead of silently overwriting them at preprocess time. + * source schema against the non-prefixed names in this set at construction time, failing fast + * instead of silently overwriting them at preprocess time. */ private[pipelines] val reservedFrameworkColNames: Set[String] = Set( startAtColName, 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 a172333e76bfe..048569cca1ed0 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 @@ -399,26 +399,19 @@ class AutoCdcMergeFlow( } /** - * Validate that the resolved source dataframe does not contain any column whose name collides - * (by exact name, resolver-aware) with an SCD-type-specific reserved framework column that is - * NOT covered by [[requireReservedPrefixAbsentInSourceColumns]]. - * - * The prefix guard above only rejects names starting with [[AutoCdcReservedNames.prefix]]. - * SCD2 additionally persists the framework columns [[Scd2BatchProcessor.startAtColName]] and - * [[Scd2BatchProcessor.endAtColName]], which do NOT carry that prefix, so a colliding source - * column would otherwise be silently overwritten during preprocessing (SPARK-57251). SCD1 - * targets carry no such non-prefixed framework columns, so this guard is a no-op for SCD1. - * - * Runs in the constructor before [[schema]] is forced, so it surfaces this actionable error - * ahead of the (temporary) [[AUTOCDC_SCD2_NOT_SUPPORTED]] gate, and remains correct once SCD2 - * support lands. + * 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]]. + // by requireReservedPrefixAbsentInSourceColumns. val reservedNames: Set[String] = changeArgs.storedAsScdType match { case ScdType.Type2 => Scd2BatchProcessor.reservedFrameworkColNames.filterNot(_.startsWith(reservedPrefix)) 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 1c893f83c60fe..96f920897edf9 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 @@ -702,7 +702,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } // =========================================================================================== - // AutoCdcMergeFlow reserved framework-column (non-prefixed) validation tests (SPARK-57251) + // 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 @@ -717,7 +717,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { .toSeq .sorted - test("SPARK-57251: non-prefixed reserved names exist and are covered by this suite") { + 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. @@ -728,8 +728,8 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-57251: an SCD2 flow with a source column colliding with a reserved framework " + - "column is rejected at construction" + "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) @@ -752,7 +752,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-57251: the reserved framework-column check runs before the SCD2-not-supported gate" + "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 @@ -765,8 +765,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-57251: an SCD1 flow with a source column matching an SCD2-only reserved name is " + - "allowed" + "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. @@ -778,8 +777,7 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-57251: an uppercase reserved framework-column name is rejected for SCD2 when " + - "caseSensitive=false" + "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) @@ -803,8 +801,8 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } test( - "SPARK-57251: a differently-cased reserved framework-column name does not trip the reserved " + - "check for SCD2 when caseSensitive=true" + "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. From c0508236d1ca81a2e2fe9a78615a40ab13a90fd6 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 07:11:05 +0000 Subject: [PATCH 3/3] [SPARK-57251][SDP] Reconcile tests with merged SCD2 schema support (SPARK-58319) SPARK-58319 (now on master) added SCD2 target-schema derivation and removed the AUTOCDC_SCD2_NOT_SUPPORTED construction gate, so two tests need updating: - "differently-cased reserved framework-column name ... caseSensitive=true": previously relied on the gate throwing after the reserved-name check passed. It now asserts the flow constructs successfully and keeps the differently-cased column as an ordinary user column. - Behavior change: SPARK-58319 shipped a test asserting a source may contain __START_AT / __END_AT and exclude them via the column selection. This PR's requireReservedFrameworkColumnsAbsentInSourceColumns guard validates the RAW source schema (pre-selection), so such a source is now rejected outright -- an ExcludeColumns cannot rescue it. That test is replaced with one asserting the rejection. Allowing an explicit opt-out by validating post-selection is tracked separately by SPARK-58325. Co-authored-by: Opus 4.8 --- .../pipelines/autocdc/AutoCdcFlowSuite.scala | 100 +++++++----------- 1 file changed, 38 insertions(+), 62 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 96f920897edf9..4fa4566f6632d 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 @@ -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(", ") ) ) } @@ -806,17 +785,14 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { ) { // 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. + // 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 ex = intercept[AnalysisException] { - newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2) - } - assert(ex.getCondition == "AUTOCDC_SCD2_NOT_SUPPORTED") + val resolvedFlow = newAutoCdcMergeFlow(sourceDf, storedAsScdType = ScdType.Type2) + assert(resolvedFlow.schema.fieldNames.contains(nonConflictingName)) } }