diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index c0046b7c7ab7..2a28b1c83d3b 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 6e5af84216b5..d0d7fa25f2c1 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,16 @@ 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. * - * 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. + * [[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 the non-prefixed names in this set at construction time, failing fast + * 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 dd4d1556afbf..1b107f05ed02 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,45 @@ class AutoCdcMergeFlow( } } + /** + * 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. + 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 32374f8ecb04..06086ff9ebe1 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,125 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { } } + // =========================================================================================== + // 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 + // 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("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( + "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( + "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( + "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( + "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( + "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) // ===========================================================================================