From 075e1fba3429a861f282ed0ca2207e5e4d4ceae9 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Fri, 24 Jul 2026 04:24:14 +0000 Subject: [PATCH 1/2] [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 | 48 +++++++ .../pipelines/autocdc/AutoCdcFlowSuite.scala | 121 ++++++++++++++++++ 4 files changed, 181 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 c0046b7c7ab7f..2a28b1c83d3bc 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 6e5af84216b58..32140965a9911 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 @@ -1089,11 +1089,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 dd4d1556afbf8..c307aeb8d04e2 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} @@ -252,6 +253,7 @@ class AutoCdcMergeFlow( val funcResult: FlowFunctionResult ) extends ResolvedFlow { requireReservedPrefixAbsentInSourceColumns() + requireReservedFrameworkColumnsAbsentInSourceColumns() def changeArgs: ChangeArgs = flow.changeArgs @@ -375,6 +377,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 32374f8ecb048..737abf49ed89e 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 @@ -499,6 +499,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 d7d11bfbca909f9f06e1cfbbc4452e27304d8e02 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 00:21:31 +0000 Subject: [PATCH 2/2] [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 32140965a9911..d0d7fa25f2c1b 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 @@ -1085,15 +1085,14 @@ object Scd2BatchProcessor { private[autocdc] 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 c307aeb8d04e2..1b107f05ed02e 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 @@ -378,26 +378,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 737abf49ed89e..06086ff9ebe12 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 @@ -500,7 +500,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 @@ -515,7 +515,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. @@ -526,8 +526,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) @@ -550,7 +550,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 @@ -563,8 +563,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. @@ -576,8 +575,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) @@ -601,8 +599,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.