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..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 @@ -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,43 @@ 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. + * + * 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..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 @@ -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. + * + * 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..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 @@ -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,151 @@ class AutoCdcFlowSuite extends QueryTest with SharedSparkSession { ) ) } + + // =========================================================================================== + // 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 + // flow construction rather than deferring to the first microbatch's reconciliation, mirroring + // the keys-presence validator above. A resolvable selection passes the check and the flow + // constructs successfully. + // =========================================================================================== + + test( + "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( + "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( + "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( + "an SCD2 flow with a resolvable track-history selection passes the check" + ) { + // `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(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 + // the flow constructs successfully. + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val resolvedFlow = newAutoCdcMergeFlow( + sourceDf = threeColumnSourceDf(), + storedAsScdType = ScdType.Type2, + trackHistorySelection = Some( + ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME"))) + ) + ) + assert(resolvedFlow.schema.fieldNames.contains(AutoCdcReservedNames.cdcMetadataColName)) + } + } + + test( + "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" + ) + ) + } + } }