[SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift

[anew]

What changes were proposed in this pull request?

Follow-up for AutoCDC SCD2 (SPARK-56249). Adds two configuration-drift validations for AutoCDC targets, closing gaps left by the existing key-column and SCD-type drift checks. Both are checked on an incremental run against values persisted when the auxiliary table was created, and both direct the user to a full refresh as the remedy.

• Sequencing-type drift (AUTOCDC_INVALID_STATE.SEQUENCING_TYPE_DRIFT): the 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
  _cdc_metadata (and, for SCD2, in the __START_AT / __END_AT interval columns), so a changed type would make new events incomparable with the persisted history. Validated at the target level, before the target-table merge, for both SCD1 and SCD2.
• Track-history drift (AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT, SCD2 only): the resolved SCD2 track-history column set defines what constitutes a run (a change in any tracked column opens a new historical record), so changing the set would reinterpret already-reconciled history. The effective set (an explicit TRACK HISTORY ON selection, or the default of every eligible non-key/non-framework column) is resolved via Scd2BatchProcessor.computeTrackedHistoryColumns — the same source of truth the reconciler uses — persisted as a new auxiliary-table property trackHistoryColumnNames, and compared order-insensitively on later runs.

Supporting changes:

• New auxiliary-table property pipelines.autocdc.trackHistoryColumnNames (JSON array; SCD2-only, absent for SCD1).
• AutoCdcAuxiliaryTableSpec gains expectedSequencingType and expectedTrackHistoryColumnNames.
• DatasetManager runs the target-level sequencing-type check before materializeTable evolves the target; the aux-level checks run in materializeAuxiliaryTable.
• Reworded AUXILIARY_TABLE_PROPERTY_MISSING to speak of "the AutoCDC configuration" generally rather than "key columns" specifically, since more properties are now validated.

Why are the changes needed?

AutoCDC already rejects key-column and SCD-type drift on incremental runs, because those silently corrupt the persisted state. Sequencing type and (for SCD2) the track-history column set have the same property: keeping them constant is an invariant the reconciled on-disk state depends on, but neither was validated. Without these checks a changed sequencing type surfaces as an opaque type-mismatch mid-merge, and a changed track-history set silently reinterprets history. Both now fail fast with an actionable error pointing at full refresh.

Does this PR introduce any user-facing change?

Yes. An incremental AutoCDC run whose sequencing result type differs from the recorded type now fails with AUTOCDC_INVALID_STATE.SEQUENCING_TYPE_DRIFT (SQLSTATE 42000); an SCD2 run that changes its effective track-history column set now fails with AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT. Both messages name the conflicting vs recorded values and recommend a full refresh. Reordering the same track-history columns, or changing the sequencing expression while keeping its type, is unaffected. This is within the unreleased AutoCDC feature on master.

How was this patch tested?

New AutoCdcConfigDriftSuite (5 tests):

• SCD1 and SCD2 sequencing-type drift each trigger SEQUENCING_TYPE_DRIFT.
• Changing the sequencing expression while keeping the same type is allowed.
• Changing the SCD2 TRACK HISTORY ON column set triggers TRACK_HISTORY_DRIFT.
• Reordering the same track-history columns does not trigger drift.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

[anew]

Here is codex' finding on test coverage in this PR:

1. Coverage is not quite enough for upgrade/backcompat.
   The PR makes SCD2 validation require the new auxiliary-table property
   pipelines.autocdc.trackHistoryColumnNames. Existing SCD2 AutoCDC auxiliary tables created before this
   PR will not have that property, and validateNoTrackHistoryDrift will fail with
   AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING. I don’t see a test for “existing SCD2 aux
   table without the new property”. That is the main gap, and possibly a product compatibility issue
   unless full refresh is explicitly intended for all existing SCD2 AutoCDC targets.

2. Track-history coverage is too narrow.
   The new suite covers explicit TRACK HISTORY name changing to explicit TRACK HISTORY amount, plus
   reorder of the same explicit set. It does not cover:

   • default track-history set vs explicit same set
   • default all tracked columns vs explicit subset
   • EXCEPT/exclude-based track-history selection
   • resolver/case-sensitivity behavior

   Those are the riskier paths because the implementation stores the result of
   Scd2BatchProcessor.computeTrackedHistoryColumns(...), not merely the user’s explicit include list.

3. Sequencing coverage is mostly good, with one small gap.
   The tests cover SCD1 drift, SCD1 same-type expression change, and SCD2 drift. I’d add one SCD2 same-
   type expression-change test for symmetry, but this is lower priority than the SCD2 track-history/
   backcompat gaps.

[szehon-ho]

Reviewed the drift-validation logic against the surrounding sql/pipelines code. The two invariants being guarded are the right ones and the sequencing-type check is sound, but I think deriving the SCD2 track-history set from the per-run inferred schema couples it to source-schema evolution in a way that will hard-fail the most common SCD2 configuration. Details inline, plus a few smaller notes on the _cdc_metadata lookup and test coverage.

Leaving this as a comment rather than a change request since the main finding hinges on an intent question you may already have an answer for.

[szehon-ho]

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.

[anew]

Good catch, and you've pinned exactly the right tension. Two parts to the response:

The divergence fix: I've moved the tracked-set resolution off the evolved target schema and onto the flow's user-selected source schema, via a new AutoCdcMergeFlow.trackHistoryColumnNames val (the shape you suggested in the other comment) that the aux-spec builder now reads. So the recorded set follows the flow's selection, not columns lingering in the sticky target — and the implicit default/* EXCEPT cases are now actually detected.

The semantic decision: requiring a full refresh on any SCD2 tracked-set change — including additive evolution under default/* EXCEPT — is intended, and I've documented it as a deliberate divergence from SCD1 (on the new val) and pinned it with end-to-end tests (add a column, drop a column, both → TRACK_HISTORY_DRIFT).
The reason it can't just tolerate additive evolution: changing which columns define an SCD2 run reinterprets where record boundaries fall, and that can't be applied to already-reconciled history. We verified that letting it through silently corrupts history when a late event drags pre-change records into the reconciliation window (a widened set fabricates spurious splits; a narrowed set destroys distinct records).

SCD1 has no run concept, so the same evolution is a no-op there.

We will evaluate whether we can support evolution of the tracked column set in SPARK-58452, but for now, this PR prevents any drift in the tracked columns.

[szehon-ho]

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):

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.

[anew]

Done — took the val on AutoCdcMergeFlow you sketched, computed from userSelectedSchema, mirroring the sequencingType line. The aux-spec builder reads inputAutoCdcFlow.trackHistoryColumnNames, and it subsumes requireTrackHistoryColumnsResolvableInSelectedSchema, which I removed. That also resolves the recording-side session concern from your DatasetManager comment: the set is now recorded once, by the flow, from a single session.

[szehon-ho]

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.

[anew]

Fixed both. validateNoTargetSequencingTypeDrift now takes expectedScdType and looks the sequencing-typed inner field up by name — Scd1BatchProcessor.cdcUpsertSequenceFieldName for SCD1, Scd2BatchProcessor.recordStartAtFieldName for SCD2 — instead of fields.head, so a metadata field added at position 0 later can't silently compare an unrelated type. And the _cdc_metadata column and the inner field are both matched through the session Resolver now, consistent with findFieldInTargetSchema, so a case-differing hand-written target DDL no longer skips the check.

[szehon-ho]

Nit: the caseSensitiveAnalysis used to record the set comes from inputAutoCdcFlow.df.sparkSession (AutoCdcAuxiliaryTable.scala:245), while the resolver used to validate it comes from context.spark here. Can we source both from one session? The val on AutoCdcMergeFlow suggested in the other comment would settle the recording side.

Also worth a test: a case-only rename of a source column across runs (value -> Value) is the case that actually exercises this resolver, and neither end-to-end test covers it - both vary the casing of the TRACK HISTORY ON selection, which is normalized away before the comparison.

[anew]

The recording side now comes from the AutoCdcMergeFlow val (per the other comment), and the sequencing-drift check's resolver is sourced from context.spark at the call site, so record and validate are consistent. On the case-only rename: I covered it as a direct unit test of validateNoTargetSequencingTypeDrift (a _cdc_metadata/inner-field schema whose casing differs from the constant, asserting the resolver still finds it and detects drift) rather than end-to-end — the streaming write always persists the canonical metadata casing, so a genuine case difference only reaches this resolver via a hand-written schema, which the unit test constructs directly.

[szehon-ho]

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.

[anew]

Added direct unit tests: matching type (SCD1 and SCD2), drift (SCD1 and SCD2), and the three silent-pass branches you flagged — _cdc_metadata absent, present-but-not-a-StructType, and present-but-empty-struct — plus a resolver-aware lookup case. All session-less, taking a plain StructType.

[szehon-ho]

By the time this fires, the target has already been altered - and the error's own remedy then wedges the pipeline.

materializeTable runs the target's evolveTable before materializeAuxiliaryTable runs these validators (DatasetManager.scala:110, then :134). So when an SCD2 flow adds a source column, the target is ALTERed to add city and then this throws TRACK_HISTORY_DRIFT. diffSchemas only adds columns, so the target keeps city for good; the aux table never gets it, since its evolveTable is below this throw.

The message says "Correct the conflicting flow(s) or perform a full refresh." Full refresh works. Correcting the flow does not: reverting the source makes this check pass (recorded [name, seq] == expected [name, seq]), but Scd2ForeachBatchHandler.reconcileMicrobatch unions the batch, the aux rows, and the target rows with unionByName, and findAffectedRowsFromTargetTable passes all target columns through. Six columns against seven, so the run dies on a column-count mismatch that mentions neither AutoCDC nor drift. The error steers users onto the one remedy that doesn't work.

Suggested fix, mirroring what you already do for sequencing type: run this check before the target's evolveTable, so a rejected run leaves the target untouched. validateNoKeyColumnDrift has the same ordering, so consider moving the group. Worth a test too - neither new end-to-end test asserts the target's schema after the rejected run.

[anew]

excellent catch! will address this shortly

[anew]

done and added a test assertion for it

[szehon-ho]

serializeKeyColumnNames / parseKeyColumnNames now serve two properties, but their scaladoc still describes only keys: "callers are expected to enforce a non-empty key set upstream" (AutoCdcAuxiliaryTable.scala:74).

That clause is wrong for this caller - an empty tracked set is legitimate when every eligible source column is a key, and computeTrackedHistoryColumns returns Seq.empty there. Someone trusting the comment could add a non-empty assertion and break it.

They're really just a JSON string-array codec, so serializeColumnNames / parseColumnNames with a property-neutral comment would fit. AUXILIARY_TABLE_PROPERTY_MALFORMED is already generic, so nothing else changes.

[anew]

renamed!

[szehon-ho]

Moving all four checks in here also narrowed when they run, and I think that drops a guard we had before.

This block sits inside case Some(existingTable) => (line 335), so every check now requires the target table to already exist. But the three aux-owned checks (key columns, SCD type, track history) used to run in materializeAuxiliaryTable whenever the auxiliary table existed, regardless of the target.

So: a user drops and recreates the target to reset it, not knowing to also drop the internal __aux_state_<target> table. On the next run:

• existingTableOpt is None, so we take the case None => createTable branch and no drift check runs at all.
• materializeAuxiliaryTable then finds the surviving aux table and goes straight to evolveTable(mergeWithExistingSchema = true), which merges the new schema in and overwrites keyColumnNamesProperty, scdTypePropertyKey, and trackHistoryColumnNamesProperty with this run's values.

That is the case validateNoKeyColumnDrift's own scaladoc says it exists to prevent: "a changed key set would otherwise be silently unioned into the schema by the additive evolve."

I traced the downstream impact, and to be fair it isn't target corruption. With changed keys the stale aux rows have NULL in the new key column so they never join - they just sit there accumulating. With a changed SCD type or track-history set, the merged aux schema stops lining up with the reconciled rows and the run fails later in unionByName or on the aux MERGE with an opaque schema error. Either way we lose what the guard was for: one clear error telling the user to full refresh.

Suggested fix: keep the sequencing-type check inside the Some(existingTable) branch, since it needs the target schema, and hoist the other three out of it so they run whenever the aux table exists, still gated on isTableIncrementallyUpdated. Worth a test in the durability suite mirroring AutoCdcScd1AuxiliaryTableDurabilitySuite:209 ("if the AutoCDC auxiliary table is dropped between runs, it is transparently recreated") - that's the same situation with the two tables swapped, and it's the half that now has neither a guard nor a test.

[szehon-ho]

Minor, and fine as a follow-up rather than something to block on: this loads the auxiliary table, and then materializeAuxiliaryTable resolves and loads it again at line 457. loadTableIfExists is itself two catalog calls (tableExists then loadTable), so that's four round trips per AutoCDC table per run where there used to be two. Fine for an in-JVM metastore, less so for a remote one.

Two independent ways to improve it whenever you get to it:

• Make loadTableIfExists one call instead of two. CatalogV2Util.loadTable already does exactly that (try / catch NoSuchTableException) and returns Option[Table], which is what V2Table aliases here. That helps all three call sites including the target load at line 306, and closes a small hole where the table disappearing between tableExists and loadTable throws an uncaught NoSuchTableException. It also swallows NoSuchDatabaseException, so if you'd rather not widen that for the target, a narrow local try / catch does the job.
• Or load the aux table once in the transformTables block and pass the handle into both materializeTable and materializeAuxiliaryTable. Nothing between them mutates the aux table, so a single snapshot is safe - and it happens to make the aux-owned checks independent of the target's existence, which is the other comment.

[szehon-ho]

This recreate means the test doesn't exercise the fix. By the time run #2 materializes, the target exists again, so existingTableOpt is Some and the drift check fires through the case Some(existingTable) branch that the previous revision already covered.

I confirmed it by running the test against the parent commit 5643ef3, which doesn't have the hoist: it passes there. Deleting this one line so the target really is absent flips that - on 5643ef3 it then fails with "Expected exception java.lang.RuntimeException to be thrown, but no exception was thrown", and passes on 563f9d8. That silent success is exactly the regression this test is meant to pin: the run takes the createTable branch for the target, then materializeAuxiliaryTable additively evolves the surviving aux table and overwrites trackHistoryColumnNames from [name] to [amount] with no error at all.

So dropping this line and letting materializeTable take its case None => createTable branch is the whole fix, and it's what the comment above already describes ("the target is absent (so it is re-created)"). I ran AutoCdcConfigDriftSuite, AutoCdcAuxiliaryTableSuite and AutoCdcScd1AuxiliaryTableDurabilitySuite with just that removed: 44 tests, all green. So the pipeline does create the target from the flow's output schema here, and nothing else in these suites depends on it being pre-created.

[anew]

you are so right, fixing

[szehon-ho]

Tiny readability thing: this auxiliaryTableSpecs.get(...).collect { case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec } is now verbatim identical to the one at line 360, and the caller already computed auxiliaryTableSpecOpt at line 115 before passing the loaded table down. A single val autoCdcSpecOpt at the top of materializeTable, or passing the spec in alongside existingAuxiliaryTable, would collapse all three into one.

Related and even smaller, so ignore if you'd rather not touch it: the caller at line 117 now loads the auxiliary table on every run including full refresh, where nothing reads it -- materializeAuxiliaryTable's full-refresh branch drops and recreates, and these checks are gated off by isTableIncrementallyUpdated. One wasted load on a path that already does DDL, so negligible; just noting it since this commit was about trimming round trips.

[szehon-ho]

Merge Summary:

• merged into master 9209369
• merged into branch-4.x cd3549a

Posted by merge_spark_pr.py