szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3677364925
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -227,13 +238,28 @@ object AutoCdcAuxiliaryTable {
StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType,
nullable = true)
val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+
deletedByBatchIdField)
+ // Resolve the effective track-history columns (an explicit TRACK HISTORY
selection, or the
+ // default of every eligible non-key/non-framework column) against the
target schema, using the
+ // same single source of truth the reconciler uses. A change in this set
reinterprets which
+ // transitions open a new historical record, so it is drift-checked.
+ val caseSensitive =
+ inputAutoCdcFlow.df.sparkSession.sessionState.conf.caseSensitiveAnalysis
+ val trackHistoryColumnNames =
Scd2BatchProcessor.computeTrackedHistoryColumns(
+ schema = targetTableSchema,
+ changeArgs = inputAutoCdcFlow.changeArgs,
+ caseSensitive = caseSensitive
+ )
Review Comment:
**Default and `EXCEPT`-based track-history sets make additive schema
evolution a permanent failure.**
`targetTableSchema` here is `inferredSchema(destinationTableIdentifier)`
(`DataflowGraph.scala:206`), i.e. the merge of *this run's* flow output
schemas, not the persisted target schema. With `trackHistorySelection = None`,
`computeTrackedHistoryColumns` returns every eligible column of that schema, so
the recorded value moves whenever the source moves:
- Run 1: source `(id, name, amount, seq)`, key `id`, no `TRACK HISTORY ON`
-> recorded `[name, amount, seq]`.
- Run 2: source gains a nullable `city` -> resolved set is `[name, amount,
seq, city]` -> arity differs -> `TRACK_HISTORY_DRIFT`.
`ColumnSelection.ExcludeColumns` (`TRACK HISTORY ON * EXCEPT (...)`) has the
same property; only an explicit `IncludeColumns` list is stable. Broadening or
narrowing the flow's `columnSelection` trips it too, which contradicts the
design note in the commit message that column-selection changes are absorbed by
additive evolution.
Two things make this worse than one failed run. The property is only
rewritten by `evolveTable` *after* validation passes, so every later run fails
identically and the pipeline is wedged until a full refresh. And
`materializeTable` runs before `materializeAuxiliaryTable`, so the target has
already been altered to add `city` by the time the run fails.
Adding a top-level column and broadening/narrowing the column selection are
documented supported behaviors for SCD1 -
`AutoCdcScd1SchemaEvolutionSuite:176`, `:275`, `:323`. There is no SCD2
schema-evolution suite, and `AutoCdcConfigDriftSuite` never adds or removes a
source column between runs, so nothing catches this.
If requiring a full refresh on any SCD2 source-schema change is intended,
that is a deliberate divergence from SCD1 that should be stated in the PR
description and pinned by a test. Otherwise the check needs to tolerate
additive evolution - only drift-check flows that gave an explicit selection,
compare only over columns present in both the recorded and current schema, or
persist the user's selection rather than its resolution.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -373,6 +401,100 @@ object AutoCdcAuxiliaryTable {
}
}
+ /**
+ * Reject an incremental update to an existing AutoCDC target table whose
sequencing type has
+ * drifted. The AutoCDC sequencing *expression* may legitimately change
across runs (e.g. a new
+ * timestamp parse format), but its resolved result type must not: the
target persists the
+ * sequencing type inside its `_cdc_metadata` struct (and, for SCD2, in the
interval columns), so
+ * a changed type would make new events incomparable with the persisted
history and would
+ * otherwise surface only as a generic CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE
during schema
+ * evolution. Runs against the target table (before its schema is evolved),
not the auxiliary
+ * table. The remedy is a full refresh.
+ *
+ * @param existingTargetSchema the schema of the already-materialized target
table.
+ * @param expectedSequencingType the resolved sequencing type of the
incoming AutoCDC flow.
+ */
+ private[graph] def validateNoTargetSequencingTypeDrift(
+ existingTargetSchema: StructType,
+ targetTableIdentifier: TableIdentifier,
+ expectedSequencingType: DataType): Unit = {
+ // The sequencing type is embedded as the inner field type(s) of the
reserved _cdc_metadata
+ // struct, for both SCD1 (delete/upsert sequence fields) and SCD2
(recordStartAt field). Read it
+ // from the first inner field. If the metadata column is absent or not a
struct, this is not a
+ // recognizable AutoCDC target state; skip rather than misreport (schema
evolution will surface
+ // any genuine incompatibility).
+ val recordedSequencingType: Option[DataType] = existingTargetSchema.fields
+ .find(_.name == AutoCdcReservedNames.cdcMetadataColName)
+ .map(_.dataType)
+ .collect { case s: StructType if s.nonEmpty => s.fields.head.dataType }
Review Comment:
Two silent-failure modes in this lookup.
`fields.head` happens to be the sequencing-typed field for both SCD types
today - `Scd1BatchProcessor.cdcMetadataColSchema` puts `deleteSequence` first,
`Scd2BatchProcessor.cdcMetadataColSchema` has only `recordStartAt` - but
nothing enforces it. A metadata field added at position 0 later (an
operation-type string, say) would silently compare an unrelated type, either
rejecting every valid run or missing real drift, with no test pointing at the
cause. The spec already carries `expectedScdType`, so dispatching on it and
looking up by `Scd1BatchProcessor.cdcDeleteSequenceFieldName` /
`Scd2BatchProcessor.recordStartAtFieldName` would pin the invariant to a name
rather than a position.
Separately, `.find(_.name == ...)` uses exact equality while every other
schema lookup in this file goes through the session `Resolver`
(`findFieldInTargetSchema`, line 291). Users hand-write the target DDL
including the metadata column - that is what `scd1MetadataDdl` /
`scd2MetadataDdl` model in the test mixin - so a case-differing declaration
resolves fine everywhere else but makes this check silently skip.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -227,13 +238,28 @@ object AutoCdcAuxiliaryTable {
StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType,
nullable = true)
val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+
deletedByBatchIdField)
+ // Resolve the effective track-history columns (an explicit TRACK HISTORY
selection, or the
+ // default of every eligible non-key/non-framework column) against the
target schema, using the
+ // same single source of truth the reconciler uses. A change in this set
reinterprets which
+ // transitions open a new historical record, so it is drift-checked.
+ val caseSensitive =
+ inputAutoCdcFlow.df.sparkSession.sessionState.conf.caseSensitiveAnalysis
Review Comment:
Minor: `resolver` on line 219 is already derived from this same conf -
`SQLConf.resolver` is literally `if (caseSensitiveAnalysis)
caseSensitiveResolution else caseInsensitiveResolution` (`SQLConf.scala:8529`)
- and `computeTrackedHistoryColumns` converts the boolean straight back into a
resolver. The method now reads one fact twice in two shapes; binding `val conf
= inputAutoCdcFlow.df.sparkSession.sessionState.conf` once and using
`conf.resolver` / `conf.caseSensitiveAnalysis` would cover it.
Worth noting that `AutoCdcMergeFlow` resolves this same selection at
construction time from a *different* session:
`requireTrackHistoryColumnsResolvableInSelectedSchema` (`Flow.scala:481`) uses
`spark`, which is `SparkSession.getActiveSession.get` (`elements.scala:46`),
not `df.sparkSession`. Same session in practice, but it undercuts the "single
source of truth" claim in the comment above.
The cleaner shape is a `val` on `AutoCdcMergeFlow` next to `sequencingType`
(`Flow.scala:278`):
```scala
private[graph] val trackHistoryColumnNames: Option[Seq[String]] =
changeArgs.storedAsScdType match {
case ScdType.Type2 =>
Some(Scd2BatchProcessor.computeTrackedHistoryColumns(
schema = userSelectedSchema,
changeArgs = changeArgs,
caseSensitive = spark.sessionState.conf.caseSensitiveAnalysis))
case ScdType.Type1 => None
}
```
The spec builder then reads `inputAutoCdcFlow.trackHistoryColumnNames`,
mirroring the `inputAutoCdcFlow.sequencingType` line below, and it subsumes
`requireTrackHistoryColumnsResolvableInSelectedSchema`. Result-preserving
today, since one AutoCDC flow per target means `inferredSchema(target)` is just
`AutoCdcMergeFlow.schema` = `userSelectedSchema` plus the framework columns,
which `computeTrackedHistoryColumns` filters out anyway.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -453,6 +468,12 @@ object DatasetManager extends Logging {
targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
expectedScdType = autoCdcSpec.expectedScdType
)
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = existingAuxiliaryTable,
+ targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
+ expectedTrackHistoryColumnNames =
autoCdcSpec.expectedTrackHistoryColumnNames,
+ resolver = context.spark.sessionState.conf.resolver
+ )
Review Comment:
The resolver used to *validate* comes from `context.spark`, while the case
sensitivity used to *record* the set comes from
`inputAutoCdcFlow.df.sparkSession` (`AutoCdcAuxiliaryTable.scala:245-246`) -
two session lookups for the same decision within one operation.
`validateNoKeyColumnDrift` above already has the same split so this is not new,
but it is worth converging on one source now that a third validator reads it.
There is a second-order effect worth being explicit about: both sides of the
track-history comparison are already normalized to actual schema field names.
`ColumnSelection.applyToSchema` returns `StructField`s taken from the schema,
so `computeTrackedHistoryColumns` emits exact schema names on both the
recording run and the checking run. The `resolver` parameter therefore never
does any work in the end-to-end path - a plain set comparison would behave
identically. That matches what the third commit found when it had to add
session-less unit tests to exercise the resolver at all. Not a bug, but it
means the argument documents an intent this call site cannot reach, and it is
why the end-to-end tests for reordering and for case-only differences cannot
fail.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTableSuite.scala:
##########
@@ -156,4 +157,129 @@ class AutoCdcAuxiliaryTableSuite extends SparkFunSuite {
"tableName" -> TableIdentifier("target", Some("ns"),
Some("cat")).unquotedString,
"propertyName" -> AutoCdcAuxiliaryTable.scdTypePropertyKey))
}
+
+ private val targetIdent = TableIdentifier("target", Some("ns"), Some("cat"))
+
+ /** An auxiliary table stub recording the given track-history column names
as JSON. */
+ private def auxTableWithTrackHistory(names: Seq[String]): Table =
+ auxTableWithProperties(Map(
+ AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty ->
+ AutoCdcAuxiliaryTable.serializeKeyColumnNames(names)))
+
+ test("validateNoTrackHistoryDrift is a no-op when the expected column set is
None") {
+ // A None expected set means the flow does not constrain track-history
(SCD1, or an SCD2 flow
+ // whose default resolution has not been computed here); the validator
must not even read the
+ // property. Passing an empty-properties table proves nothing is
dereferenced.
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = auxTableWithProperties(Map.empty),
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = None,
+ resolver = caseInsensitiveResolution)
+ }
+
+ test("validateNoTrackHistoryDrift accepts a recorded set that matches
regardless of order") {
+ val existing = auxTableWithTrackHistory(Seq("name", "amount", "seq"))
+ // Same set, different order: must not throw.
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = Some(Seq("seq", "name", "amount")),
+ resolver = caseInsensitiveResolution)
+ }
+
+ test("validateNoTrackHistoryDrift compares case-insensitively under the
default resolver, " +
+ "even when the stored property names differ only in case") {
+ // Isolates the resolver-aware comparison: the stored property holds
`Name`/`AMOUNT` while the
+ // expected set holds `name`/`amount`. In the end-to-end path both sides
are normalized to
+ // actual schema field names before comparison, so only a direct unit test
can exercise a
+ // genuine case difference reaching the resolver. Under the default
resolver, no drift.
+ val existing = auxTableWithTrackHistory(Seq("Name", "AMOUNT"))
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = Some(Seq("name", "amount")),
+ resolver = caseInsensitiveResolution)
+ }
+
+ test("validateNoTrackHistoryDrift throws TRACK_HISTORY_DRIFT under the
case-sensitive resolver " +
+ "when the stored property names differ only in case") {
+ // The mirror of the case-insensitive test: with the case-sensitive
resolver, `Name` and
+ // `name` are distinct, so the same-cardinality sets do not match and the
validator drifts.
+ val existing = auxTableWithTrackHistory(Seq("Name", "amount"))
+ val ex = intercept[AnalysisException] {
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = Some(Seq("name", "amount")),
+ resolver = caseSensitiveResolution)
+ }
+ checkError(
+ exception = ex,
+ condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+ sqlState = "42000",
+ parameters = Map(
+ "tableName" -> targetIdent.unquotedString,
+ "expectedTrackHistoryColumns" -> "name, amount",
+ "recordedTrackHistoryColumns" -> "Name, amount"))
+ }
+
+ test("validateNoTrackHistoryDrift throws TRACK_HISTORY_DRIFT when the
recorded set differs") {
+ val existing = auxTableWithTrackHistory(Seq("name"))
+ val ex = intercept[AnalysisException] {
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = Some(Seq("amount")),
+ resolver = caseInsensitiveResolution)
+ }
+ checkError(
+ exception = ex,
+ condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+ sqlState = "42000",
+ parameters = Map(
+ "tableName" -> targetIdent.unquotedString,
+ "expectedTrackHistoryColumns" -> "amount",
+ "recordedTrackHistoryColumns" -> "name"))
+ }
+
+ test("validateNoTrackHistoryDrift throws AUXILIARY_TABLE_PROPERTY_MISSING
when the " +
+ "track-history property is absent") {
+ // An SCD2 aux table created before this property existed: the validator
must surface a
+ // structured error (remedy: full refresh) rather than skipping the check.
+ val ex = intercept[AnalysisException] {
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = auxTableWithProperties(Map.empty),
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = Some(Seq("name")),
+ resolver = caseInsensitiveResolution)
+ }
+ checkError(
+ exception = ex,
+ condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+ sqlState = "42000",
+ parameters = Map(
+ "tableName" -> targetIdent.unquotedString,
+ "propertyName" ->
AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty))
+ }
+
+ test("validateNoTrackHistoryDrift throws AUXILIARY_TABLE_PROPERTY_MALFORMED
when the " +
+ "track-history property is not a JSON array of strings") {
+ val existing = auxTableWithProperties(Map(
+ AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty ->
"not-a-json-array"))
+ val ex = intercept[AnalysisException] {
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = targetIdent,
+ expectedTrackHistoryColumnNames = Some(Seq("name")),
+ resolver = caseInsensitiveResolution)
+ }
+ checkError(
+ exception = ex,
+ condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MALFORMED",
+ sqlState = "42000",
+ parameters = Map(
+ "tableName" -> targetIdent.unquotedString,
+ "propertyName" ->
AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty,
+ "rawValue" -> "not-a-json-array"))
+ }
Review Comment:
`validateNoTrackHistoryDrift` got six direct unit tests here;
`validateNoTargetSequencingTypeDrift` got none, only end-to-end coverage in
`AutoCdcConfigDriftSuite`. The uncovered branches are the ones that pass
silently: `_cdc_metadata` absent, present but not a `StructType`, and present
but an empty struct. That is where a regression would go unnoticed, and they
are cheap tests since the validator takes a plain `StructType` and needs no
session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
