You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anew (via GitHub)" <gi...@apache.org> on 2026/07/29 07:38:42 UTC

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

anew opened a new pull request, #57625:
URL: https://github.com/apache/spark/pull/57625

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


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3677364948


##########
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:
   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.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3693060975


##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -453,6 +470,12 @@ object DatasetManager extends Logging {
                 targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
                 expectedScdType = autoCdcSpec.expectedScdType
               )
+              AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(

Review Comment:
   done and added a test assertion for it



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on PR #57625:
URL: https://github.com/apache/spark/pull/57625#issuecomment-5114759096

   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.
   
   


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57625: [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift
URL: https://github.com/apache/spark/pull/57625


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3693574074


##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -333,6 +333,26 @@ object DatasetManager extends Logging {
     // Create the table if absent, otherwise evolve it (schema + properties).
     existingTableOpt match {
       case Some(existingTable) =>
+        // For an incrementally-updated AutoCDC target, validate that the AutoCDC configuration has
+        // not drifted from what the auxiliary table recorded, BEFORE evolving the target's schema.
+        // This ordering is load-bearing: `evolveTable` below ALTERs the target (additively) in
+        // place, so a check that ran afterwards would leave the target already mutated by a run it
+        // then rejects -- and the drift remedy ("correct the flow") could not undo that. Running
+        // first means a rejected run leaves the target untouched. It also surfaces a
+        // sequencing-type change as an actionable SEQUENCING_TYPE_DRIFT rather than a generic
+        // CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE from the schema merge.
+        if (isTableIncrementallyUpdated) {
+          resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).collect {
+            case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec
+          }.foreach { autoCdcSpec =>
+            validateNoAutoCdcConfigDrift(

Review Comment:
   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.



##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -474,6 +481,61 @@ object DatasetManager extends Logging {
     }
   }
 
+  /**
+   * Validate that an incrementally-updated AutoCDC target's configuration has not drifted from
+   * what its auxiliary table recorded. Called from [[materializeTable]] BEFORE the target's
+   * `evolveTable`, so a rejected run leaves the target's schema untouched (a check that ran after
+   * the target evolve would leave it already ALTERed by a run it then rejects, and the drift
+   * remedy could not undo that).
+   *
+   * The sequencing-type check reads the existing target schema directly; the key-column, SCD-type,
+   * and track-history checks read the recorded properties off the existing auxiliary table, so this
+   * loads it. If the auxiliary table does not exist yet (first AutoCDC run over a pre-existing
+   * target), those three checks are skipped -- there is no recorded configuration to drift from.
+   *
+   * @param existingTargetSchema the schema of the already-materialized target table.
+   * @param autoCdcSpec the auxiliary-table spec carrying this run's expected AutoCDC configuration.
+   * @param context the context for the pipeline update.
+   */
+  private def validateNoAutoCdcConfigDrift(
+      existingTargetSchema: StructType,
+      autoCdcSpec: AutoCdcAuxiliaryTableSpec,
+      context: PipelineUpdateContext): Unit = {
+    val resolver = context.spark.sessionState.conf.resolver
+
+    // Independent of the auxiliary table: the sequencing type is embedded in the target's schema.
+    AutoCdcAuxiliaryTable.validateNoTargetSequencingTypeDrift(
+      existingTargetSchema = existingTargetSchema,
+      targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
+      expectedScdType = autoCdcSpec.expectedScdType,
+      expectedSequencingType = autoCdcSpec.expectedSequencingType,
+      resolver = resolver
+    )
+
+    // The remaining checks read the recorded configuration off the existing auxiliary table.
+    val (auxCatalog, auxIdentifier) =
+      PipelinesCatalogUtils.resolveTableCatalog(context.spark, autoCdcSpec.identifier)
+    loadTableIfExists(auxCatalog, auxIdentifier).foreach { existingAuxiliaryTable =>

Review Comment:
   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.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688722225


##########
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 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.
    



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688689860


##########
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:
   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. 
   



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3692501362


##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -453,6 +470,12 @@ object DatasetManager extends Logging {
                 targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
                 expectedScdType = autoCdcSpec.expectedScdType
               )
+              AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(

Review Comment:
   excellent catch! will address this shortly



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #57625:
URL: https://github.com/apache/spark/pull/57625#issuecomment-5150007781

   **Merge Summary:**
   - merged into master https://github.com/apache/spark/commit/9209369ed28a4bff80bb30ce9bed022cc8dccc7a
   - merged into branch-4.x https://github.com/apache/spark/commit/cd3549a0517225d5b607fac89f6d4e54f3a98965
   
   *Posted by `merge_spark_pr.py`*


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688733812


##########
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:
    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.
     
   
   



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3694118102


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcConfigDriftSuite.scala:
##########
@@ -0,0 +1,593 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end tests covering AutoCDC configuration-drift validation for the sequencing result type
+ * (SCD1 and SCD2) and the SCD2 track-history column set, validated at flow execution-init time
+ * against the auxiliary table's recorded configuration (mirroring
+ * [[AutoCdcScd1KeyDriftSuite]] for keys).
+ *
+ * Guiding principle: guard the invariants that keep already-persisted state coherent, not the
+ * expressions themselves. The sequencing expression and delete condition may change across runs;
+ * the sequencing result *type* and the SCD2 track-history column *set* may not.
+ */
+class AutoCdcConfigDriftSuite
+    extends ExecutionTest
+    with SharedSparkSession
+    with AutoCdcGraphExecutionTestMixin {
+
+  import testImplicits._
+
+  private def targetName: String =
+    fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString
+
+  /** SCD2 target DDL: user columns + the SCD2 framework columns (sequencing type long). */
+  private def createScd2Target(userCols: String): Unit = {
+    spark.sql(
+      s"CREATE TABLE $catalog.$namespace.target ($userCols, $scd2MetadataDdl)"
+    )
+  }
+
+  // ===========================================================================================
+  // Sequencing type drift
+  // ===========================================================================================
+
+  test("an SCD1 flow whose sequencing type differs from the recorded type triggers " +
+    "SEQUENCING_TYPE_DRIFT") {
+    spark.sql(
+      s"CREATE TABLE $catalog.$namespace.target " +
+      s"(id INT NOT NULL, seq_long BIGINT, seq_int INT, $scd1MetadataDdl)"
+    )
+
+    // Pipeline #1 sequences by a BIGINT column; aux records sequencingType = long.
+    val stream1 = MemoryStream[(Int, Long, Int)]
+    stream1.addData((1, 1L, 1))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_long"))
+
+    // Pipeline #2 sequences by an INT column - type drift (int vs long), even though the
+    // expression (a different column) is otherwise a legal change.
+    val stream2 = MemoryStream[(Int, Long, Int)]
+    stream2.addData((1, 2L, 2))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_int")
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.SEQUENCING_TYPE_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedSequencingType" -> "INT",
+        "recordedSequencingType" -> "BIGINT"
+      )
+    )
+  }
+
+  test("an SCD1 flow that changes the sequencing expression but keeps the same type does NOT " +
+    "trigger drift") {
+    spark.sql(
+      s"CREATE TABLE $catalog.$namespace.target " +
+      s"(id INT NOT NULL, seq BIGINT, $scd1MetadataDdl)"
+    )
+
+    val stream1 = MemoryStream[(Int, Long)]
+    stream1.addData((1, 10L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq"))
+
+    // A different expression over the same column, still yielding BIGINT: legal, no drift.
+    val stream2 = MemoryStream[(Int, Long)]
+    stream2.addData((1, 20L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq" + 1L))
+  }
+
+  test("an SCD2 flow whose sequencing type differs from the recorded type triggers " +
+    "SEQUENCING_TYPE_DRIFT") {
+    createScd2Target("id INT NOT NULL, seq_long BIGINT, seq_int INT")
+
+    val stream1 = MemoryStream[(Int, Long, Int)]
+    stream1.addData((1, 1L, 1))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_long",
+      scdType = ScdType.Type2))
+
+    val stream2 = MemoryStream[(Int, Long, Int)]
+    stream2.addData((1, 2L, 2))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_int",
+      scdType = ScdType.Type2)
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.SEQUENCING_TYPE_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedSequencingType" -> "INT",
+        "recordedSequencingType" -> "BIGINT"
+      )
+    )
+  }
+
+  // ===========================================================================================
+  // Track-history drift (SCD2 only)
+  // ===========================================================================================
+
+  test("an SCD2 flow that changes its explicit TRACK HISTORY column set triggers " +
+    "TRACK_HISTORY_DRIFT") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 tracks history on `name` only.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+    // Pipeline #2 tracks history on `amount` - a different set.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("amount")))))
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "amount",
+        "recordedTrackHistoryColumns" -> "name"
+      )
+    )
+  }
+
+  test("an SCD2 flow that reorders the same TRACK HISTORY columns does NOT trigger drift") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(
+        Seq(UnqualifiedColumnName("name"), UnqualifiedColumnName("amount"))))))
+
+    // Same set, reversed order: run semantics are order-insensitive, so no drift.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(
+        Seq(UnqualifiedColumnName("amount"), UnqualifiedColumnName("name"))))))
+  }
+
+  test("an SCD2 flow with no TRACK HISTORY (default = all eligible columns) followed by an " +
+    "explicit selection of that same set does NOT trigger drift") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 omits trackHistorySelection: the recorded set is the default, every eligible
+    // (non-key, non-framework) column, i.e. name, amount, seq.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Pipeline #2 explicitly lists that same default set: the drift check compares resolved sets,
+    // not user syntax, so an explicit restatement of the default must NOT drift.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(Seq(
+        UnqualifiedColumnName("name"),
+        UnqualifiedColumnName("amount"),
+        UnqualifiedColumnName("seq"))))))
+  }
+
+  test("an SCD2 flow with no TRACK HISTORY (default = all eligible columns) followed by an " +
+    "explicit subset triggers TRACK_HISTORY_DRIFT") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 records the default set (name, amount, seq).
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Pipeline #2 narrows to an explicit subset (name only): a real change to which transitions
+    // open a new record, so it must drift against the recorded default set.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))))
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "name",
+        "recordedTrackHistoryColumns" -> "name, amount, seq"
+      )
+    )
+  }
+
+  test("an SCD2 flow's EXCEPT-based TRACK HISTORY followed by an equivalent explicit include " +
+    "set does NOT trigger drift") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 uses TRACK HISTORY ON * EXCEPT (amount): the resolved set is the eligible
+    // columns minus `amount`, i.e. name, seq.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("amount"))))))
+
+    // Pipeline #2 states the same set as an explicit include list: EXCLUDE and INCLUDE that
+    // resolve to the same set must not drift, since only the resolved set is recorded.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(
+        Seq(UnqualifiedColumnName("name"), UnqualifiedColumnName("seq"))))))
+  }
+
+  test("SCD2 track-history drift validation is resolver-aware: a case-only difference does NOT " +
+    "trigger drift under the default (case-insensitive) resolver") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 records track-history on `name`.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+    // Pipeline #2 selects `NAME` (different case). The source DF column is still lowercase `name`
+    // so it resolves against the schema; only the tracking-column casing differs. Under the
+    // default case-insensitive resolver the two sets are equal, so there must be no drift.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME"))))))
+  }
+
+  test("an existing SCD2 aux table missing the trackHistoryColumnNames property requires a " +
+    "full refresh (AUXILIARY_TABLE_PROPERTY_MISSING)") {
+    // Back-compat guard: an SCD2 auxiliary table created before this change carries no
+    // trackHistoryColumnNames property. Track-history drift validation surfaces this as a
+    // structured AUXILIARY_TABLE_PROPERTY_MISSING (remedy: full refresh) rather than silently
+    // skipping the check. Simulate the pre-existing table by unsetting the property after the
+    // first run, then run again.
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    val stream = MemoryStream[(Int, String, Int, Long)]
+    def buildCtx(): TestGraphRegistrationContext =
+      singleAutoCdcFlowPipeline(
+        flowName = "auto_cdc_flow",
+        target = "target",
+        sourceDf = stream.toDF().toDF("id", "name", "amount", "seq"),
+        keys = Seq("id"),
+        sequencing = $"seq",
+        scdType = ScdType.Type2)
+
+    stream.addData((1, "a", 10, 1L))
+    runPipeline(buildCtx())
+
+    // Drop the property to mimic an aux table materialized before this change.
+    spark.sql(
+      s"ALTER TABLE ${auxTableNameFor("target")} " +
+      s"UNSET TBLPROPERTIES ('${AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty}')"
+    )
+
+    stream.addData((1, "a", 20, 2L))
+    val ex = intercept[RuntimeException] { runPipeline(buildCtx()) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "propertyName" -> AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty
+      )
+    )
+  }
+
+  // ===========================================================================================
+  // Sequencing type: SCD2 expression-change symmetry with the SCD1 case above
+  // ===========================================================================================
+
+  test("an SCD2 flow that changes the sequencing expression but keeps the same type does NOT " +
+    "trigger drift") {
+    createScd2Target("id INT NOT NULL, seq BIGINT")
+
+    val stream1 = MemoryStream[(Int, Long)]
+    stream1.addData((1, 10L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // A different expression over the same column, still yielding BIGINT: legal, no drift.
+    val stream2 = MemoryStream[(Int, Long)]
+    stream2.addData((1, 20L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq" + 1L,
+      scdType = ScdType.Type2))
+  }
+
+  // ===========================================================================================
+  // Intended divergence from SCD1: additive source-schema evolution under default / EXCEPT
+  // tracking changes the effective tracked set, and so requires a full refresh.
+  // ===========================================================================================
+
+  test("adding a source column under default (all-column) tracking triggers TRACK_HISTORY_DRIFT") {
+    // The effective tracked set is derived from the flow's selected source schema, so under default
+    // tracking every selected non-key column is tracked. Adding a source column therefore changes
+    // the tracked set, which reinterprets which transitions open a new SCD2 record and cannot be
+    // applied to already-reconciled history. Unlike SCD1 (where a new nullable column is absorbed
+    // by schema evolution), SCD2 requires a full refresh. Pins that intended divergence.
+    createScd2Target("id INT NOT NULL, name STRING, seq BIGINT")
+
+    // Run #1: source (id, name, seq); recorded tracked set = {name, seq}.
+    val stream1 = MemoryStream[(Int, String, Long)]
+    stream1.addData((1, "a", 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Run #2: source gains a nullable `city`; default tracking now resolves to {name, city, seq}.
+    val stream2 = MemoryStream[(Int, String, String, Long)]
+    stream2.addData((1, "a", "nyc", 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "city", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2)
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "name, city, seq",
+        "recordedTrackHistoryColumns" -> "name, seq"))
+
+    // The drift check runs before the target's schema is evolved, so the rejected run must leave
+    // the target untouched: `city` must NOT have been added. (Were it added, the "correct the
+    // flow" remedy would then wedge the pipeline on a column-count mismatch during reconciliation.)
+    assert(
+      !spark.table(s"$catalog.$namespace.target").schema.fieldNames.contains("city"),
+      "rejected run must not have evolved the target schema to add `city`")
+  }
+
+  test("dropping a source column under default (all-column) tracking triggers " +
+    "TRACK_HISTORY_DRIFT") {
+    // The mirror of the additive case: removing a selected column shrinks the default tracked set,
+    // which is likewise a tracked-set change requiring a full refresh.
+    createScd2Target("id INT NOT NULL, name STRING, city STRING, seq BIGINT")
+
+    // Run #1: source (id, name, city, seq); recorded tracked set = {name, city, seq}.
+    val stream1 = MemoryStream[(Int, String, String, Long)]
+    stream1.addData((1, "a", "nyc", 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "city", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Run #2: `city` dropped from the source; default tracking now resolves to {name, seq}.
+    val stream2 = MemoryStream[(Int, String, Long)]
+    stream2.addData((1, "a", 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2)
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "name, seq",
+        "recordedTrackHistoryColumns" -> "name, city, seq"))
+  }
+
+  test("dropping the target (but not the auxiliary table) between runs still detects drift") {
+    // A user drops and recreates the target to reset it, but does not know to also drop the
+    // internal auxiliary table. On the next run the target is absent (so it is re-created), but the
+    // stale auxiliary table survives with the old recorded configuration. Drift validation reads
+    // the auxiliary table, so it must still fire regardless of the target's existence -- otherwise
+    // the aux table's additive evolve would silently overwrite the recorded track-history property
+    // with the new run's value. Mirrors AutoCdcScd1AuxiliaryTableDurabilitySuite's
+    // "auxiliary table is dropped between runs" case, with the two tables swapped.
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Run #1: track history on `name`; records trackHistoryColumnNames = [name] on the aux table.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection =
+        Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+    // Drop ONLY the target; the auxiliary table survives with its recorded config.
+    spark.sql(s"DROP TABLE $catalog.$namespace.target")
+    assert(spark.catalog.tableExists(auxTableNameFor("target")),
+      "auxiliary table should survive dropping the target")
+
+    // Run #2: recreate the target and track history on `amount` instead -- a changed tracked set.
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")

Review Comment:
   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.



##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -330,9 +348,46 @@ object DatasetManager extends Logging {
       context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
     }
 
+    // For an incrementally-updated AutoCDC target, validate that the AutoCDC configuration recorded
+    // on the auxiliary table has not drifted, BEFORE anything is created or evolved this run. These
+    // checks read the auxiliary table, so they run whenever IT exists -- independent of whether the
+    // target exists. That matters when a user drops and recreates the target without dropping the
+    // internal auxiliary table: the target is then absent (so it is re-created below) but the stale
+    // auxiliary table survives, and `materializeAuxiliaryTable`'s additive evolve would otherwise
+    // silently overwrite the recorded key/SCD-type/track-history properties with this run's values.
+    // Running here turns that into one clear drift error (remedy: full refresh).
+    if (isTableIncrementallyUpdated) {
+      resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).collect {
+        case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec
+      }.foreach { autoCdcSpec =>
+        validateNoAutoCdcAuxConfigDrift(autoCdcSpec, existingAuxiliaryTable, context)
+      }
+    }
+
     // Create the table if absent, otherwise evolve it (schema + properties).
     existingTableOpt match {
       case Some(existingTable) =>
+        // The sequencing-type check needs the existing target schema (the type is embedded in the
+        // target's `_cdc_metadata`), so it runs here, and BEFORE `evolveTable`: `evolveTable`
+        // ALTERs the target (additively) in place, so a check that ran afterwards would leave the
+        // target already mutated by a run it then rejects -- and the drift remedy could not undo
+        // that. Running first means a rejected run leaves the target untouched, and surfaces the
+        // change as an actionable SEQUENCING_TYPE_DRIFT rather than a generic
+        // CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE from the schema merge.
+        if (isTableIncrementallyUpdated) {
+          resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).collect {

Review Comment:
   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.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688709867


##########
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:
   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.
     



##########
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:
   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.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688694738


##########
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:
   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.
   



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3692319590


##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -453,6 +470,12 @@ object DatasetManager extends Logging {
                 targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
                 expectedScdType = autoCdcSpec.expectedScdType
               )
+              AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(

Review Comment:
   **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.



##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -227,13 +239,27 @@ object AutoCdcAuxiliaryTable {
       StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
     val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+ deletedByBatchIdField)
 
+    // The effective track-history column set, resolved by the flow from its user-selected source
+    // schema (see [[AutoCdcMergeFlow.trackHistoryColumnNames]]) -- NOT recomputed here from the
+    // evolved target schema, which would keep tracking columns the flow no longer selects and would
+    // miss implicit (default / `* EXCEPT`) tracked-set changes. A change in this set reinterprets
+    // which transitions open a new historical record, so it is drift-checked.
+    val trackHistoryColumnNames = inputAutoCdcFlow.trackHistoryColumnNames.getOrElse(
+      throw SparkException.internalError(
+        "SCD2 AutoCDC flow is missing its resolved track-history column set."
+      )
+    )
+
     val scd2AuxiliaryTableProperties =
       // Record which SCD strategy this auxiliary table serves so downstream readers can identify it
       // without inspecting the schema.
       Map(scdTypePropertyKey -> ScdType.Type2.label) ++
       // Persist the AutoCDC key column names as a JSON list; immutable post-creation (full-refresh
       // is the only way to change it).
       Map(keyColumnNamesProperty -> serializeKeyColumnNames(keyFields.map(_.name))) ++
+      // Persist the resolved track-history column names; a change reinterprets already-reconciled
+      // history, so it is immutable post-creation (full-refresh is the only way to change it).
+      Map(trackHistoryColumnNamesProperty -> serializeKeyColumnNames(trackHistoryColumnNames)) ++

Review Comment:
   `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.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3694157382


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcConfigDriftSuite.scala:
##########
@@ -0,0 +1,593 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end tests covering AutoCDC configuration-drift validation for the sequencing result type
+ * (SCD1 and SCD2) and the SCD2 track-history column set, validated at flow execution-init time
+ * against the auxiliary table's recorded configuration (mirroring
+ * [[AutoCdcScd1KeyDriftSuite]] for keys).
+ *
+ * Guiding principle: guard the invariants that keep already-persisted state coherent, not the
+ * expressions themselves. The sequencing expression and delete condition may change across runs;
+ * the sequencing result *type* and the SCD2 track-history column *set* may not.
+ */
+class AutoCdcConfigDriftSuite
+    extends ExecutionTest
+    with SharedSparkSession
+    with AutoCdcGraphExecutionTestMixin {
+
+  import testImplicits._
+
+  private def targetName: String =
+    fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString
+
+  /** SCD2 target DDL: user columns + the SCD2 framework columns (sequencing type long). */
+  private def createScd2Target(userCols: String): Unit = {
+    spark.sql(
+      s"CREATE TABLE $catalog.$namespace.target ($userCols, $scd2MetadataDdl)"
+    )
+  }
+
+  // ===========================================================================================
+  // Sequencing type drift
+  // ===========================================================================================
+
+  test("an SCD1 flow whose sequencing type differs from the recorded type triggers " +
+    "SEQUENCING_TYPE_DRIFT") {
+    spark.sql(
+      s"CREATE TABLE $catalog.$namespace.target " +
+      s"(id INT NOT NULL, seq_long BIGINT, seq_int INT, $scd1MetadataDdl)"
+    )
+
+    // Pipeline #1 sequences by a BIGINT column; aux records sequencingType = long.
+    val stream1 = MemoryStream[(Int, Long, Int)]
+    stream1.addData((1, 1L, 1))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_long"))
+
+    // Pipeline #2 sequences by an INT column - type drift (int vs long), even though the
+    // expression (a different column) is otherwise a legal change.
+    val stream2 = MemoryStream[(Int, Long, Int)]
+    stream2.addData((1, 2L, 2))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_int")
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.SEQUENCING_TYPE_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedSequencingType" -> "INT",
+        "recordedSequencingType" -> "BIGINT"
+      )
+    )
+  }
+
+  test("an SCD1 flow that changes the sequencing expression but keeps the same type does NOT " +
+    "trigger drift") {
+    spark.sql(
+      s"CREATE TABLE $catalog.$namespace.target " +
+      s"(id INT NOT NULL, seq BIGINT, $scd1MetadataDdl)"
+    )
+
+    val stream1 = MemoryStream[(Int, Long)]
+    stream1.addData((1, 10L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq"))
+
+    // A different expression over the same column, still yielding BIGINT: legal, no drift.
+    val stream2 = MemoryStream[(Int, Long)]
+    stream2.addData((1, 20L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq" + 1L))
+  }
+
+  test("an SCD2 flow whose sequencing type differs from the recorded type triggers " +
+    "SEQUENCING_TYPE_DRIFT") {
+    createScd2Target("id INT NOT NULL, seq_long BIGINT, seq_int INT")
+
+    val stream1 = MemoryStream[(Int, Long, Int)]
+    stream1.addData((1, 1L, 1))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_long",
+      scdType = ScdType.Type2))
+
+    val stream2 = MemoryStream[(Int, Long, Int)]
+    stream2.addData((1, 2L, 2))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq_long", "seq_int"),
+      keys = Seq("id"),
+      sequencing = $"seq_int",
+      scdType = ScdType.Type2)
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.SEQUENCING_TYPE_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedSequencingType" -> "INT",
+        "recordedSequencingType" -> "BIGINT"
+      )
+    )
+  }
+
+  // ===========================================================================================
+  // Track-history drift (SCD2 only)
+  // ===========================================================================================
+
+  test("an SCD2 flow that changes its explicit TRACK HISTORY column set triggers " +
+    "TRACK_HISTORY_DRIFT") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 tracks history on `name` only.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+    // Pipeline #2 tracks history on `amount` - a different set.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("amount")))))
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "amount",
+        "recordedTrackHistoryColumns" -> "name"
+      )
+    )
+  }
+
+  test("an SCD2 flow that reorders the same TRACK HISTORY columns does NOT trigger drift") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(
+        Seq(UnqualifiedColumnName("name"), UnqualifiedColumnName("amount"))))))
+
+    // Same set, reversed order: run semantics are order-insensitive, so no drift.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(
+        Seq(UnqualifiedColumnName("amount"), UnqualifiedColumnName("name"))))))
+  }
+
+  test("an SCD2 flow with no TRACK HISTORY (default = all eligible columns) followed by an " +
+    "explicit selection of that same set does NOT trigger drift") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 omits trackHistorySelection: the recorded set is the default, every eligible
+    // (non-key, non-framework) column, i.e. name, amount, seq.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Pipeline #2 explicitly lists that same default set: the drift check compares resolved sets,
+    // not user syntax, so an explicit restatement of the default must NOT drift.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(Seq(
+        UnqualifiedColumnName("name"),
+        UnqualifiedColumnName("amount"),
+        UnqualifiedColumnName("seq"))))))
+  }
+
+  test("an SCD2 flow with no TRACK HISTORY (default = all eligible columns) followed by an " +
+    "explicit subset triggers TRACK_HISTORY_DRIFT") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 records the default set (name, amount, seq).
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Pipeline #2 narrows to an explicit subset (name only): a real change to which transitions
+    // open a new record, so it must drift against the recorded default set.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))))
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "name",
+        "recordedTrackHistoryColumns" -> "name, amount, seq"
+      )
+    )
+  }
+
+  test("an SCD2 flow's EXCEPT-based TRACK HISTORY followed by an equivalent explicit include " +
+    "set does NOT trigger drift") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 uses TRACK HISTORY ON * EXCEPT (amount): the resolved set is the eligible
+    // columns minus `amount`, i.e. name, seq.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("amount"))))))
+
+    // Pipeline #2 states the same set as an explicit include list: EXCLUDE and INCLUDE that
+    // resolve to the same set must not drift, since only the resolved set is recorded.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(ColumnSelection.IncludeColumns(
+        Seq(UnqualifiedColumnName("name"), UnqualifiedColumnName("seq"))))))
+  }
+
+  test("SCD2 track-history drift validation is resolver-aware: a case-only difference does NOT " +
+    "trigger drift under the default (case-insensitive) resolver") {
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Pipeline #1 records track-history on `name`.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+    // Pipeline #2 selects `NAME` (different case). The source DF column is still lowercase `name`
+    // so it resolves against the schema; only the tracking-column casing differs. Under the
+    // default case-insensitive resolver the two sets are equal, so there must be no drift.
+    val stream2 = MemoryStream[(Int, String, Int, Long)]
+    stream2.addData((1, "a", 20, 2L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection = Some(
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("NAME"))))))
+  }
+
+  test("an existing SCD2 aux table missing the trackHistoryColumnNames property requires a " +
+    "full refresh (AUXILIARY_TABLE_PROPERTY_MISSING)") {
+    // Back-compat guard: an SCD2 auxiliary table created before this change carries no
+    // trackHistoryColumnNames property. Track-history drift validation surfaces this as a
+    // structured AUXILIARY_TABLE_PROPERTY_MISSING (remedy: full refresh) rather than silently
+    // skipping the check. Simulate the pre-existing table by unsetting the property after the
+    // first run, then run again.
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    val stream = MemoryStream[(Int, String, Int, Long)]
+    def buildCtx(): TestGraphRegistrationContext =
+      singleAutoCdcFlowPipeline(
+        flowName = "auto_cdc_flow",
+        target = "target",
+        sourceDf = stream.toDF().toDF("id", "name", "amount", "seq"),
+        keys = Seq("id"),
+        sequencing = $"seq",
+        scdType = ScdType.Type2)
+
+    stream.addData((1, "a", 10, 1L))
+    runPipeline(buildCtx())
+
+    // Drop the property to mimic an aux table materialized before this change.
+    spark.sql(
+      s"ALTER TABLE ${auxTableNameFor("target")} " +
+      s"UNSET TBLPROPERTIES ('${AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty}')"
+    )
+
+    stream.addData((1, "a", 20, 2L))
+    val ex = intercept[RuntimeException] { runPipeline(buildCtx()) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "propertyName" -> AutoCdcAuxiliaryTable.trackHistoryColumnNamesProperty
+      )
+    )
+  }
+
+  // ===========================================================================================
+  // Sequencing type: SCD2 expression-change symmetry with the SCD1 case above
+  // ===========================================================================================
+
+  test("an SCD2 flow that changes the sequencing expression but keeps the same type does NOT " +
+    "trigger drift") {
+    createScd2Target("id INT NOT NULL, seq BIGINT")
+
+    val stream1 = MemoryStream[(Int, Long)]
+    stream1.addData((1, 10L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // A different expression over the same column, still yielding BIGINT: legal, no drift.
+    val stream2 = MemoryStream[(Int, Long)]
+    stream2.addData((1, 20L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq" + 1L,
+      scdType = ScdType.Type2))
+  }
+
+  // ===========================================================================================
+  // Intended divergence from SCD1: additive source-schema evolution under default / EXCEPT
+  // tracking changes the effective tracked set, and so requires a full refresh.
+  // ===========================================================================================
+
+  test("adding a source column under default (all-column) tracking triggers TRACK_HISTORY_DRIFT") {
+    // The effective tracked set is derived from the flow's selected source schema, so under default
+    // tracking every selected non-key column is tracked. Adding a source column therefore changes
+    // the tracked set, which reinterprets which transitions open a new SCD2 record and cannot be
+    // applied to already-reconciled history. Unlike SCD1 (where a new nullable column is absorbed
+    // by schema evolution), SCD2 requires a full refresh. Pins that intended divergence.
+    createScd2Target("id INT NOT NULL, name STRING, seq BIGINT")
+
+    // Run #1: source (id, name, seq); recorded tracked set = {name, seq}.
+    val stream1 = MemoryStream[(Int, String, Long)]
+    stream1.addData((1, "a", 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Run #2: source gains a nullable `city`; default tracking now resolves to {name, city, seq}.
+    val stream2 = MemoryStream[(Int, String, String, Long)]
+    stream2.addData((1, "a", "nyc", 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "city", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2)
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "name, city, seq",
+        "recordedTrackHistoryColumns" -> "name, seq"))
+
+    // The drift check runs before the target's schema is evolved, so the rejected run must leave
+    // the target untouched: `city` must NOT have been added. (Were it added, the "correct the
+    // flow" remedy would then wedge the pipeline on a column-count mismatch during reconciliation.)
+    assert(
+      !spark.table(s"$catalog.$namespace.target").schema.fieldNames.contains("city"),
+      "rejected run must not have evolved the target schema to add `city`")
+  }
+
+  test("dropping a source column under default (all-column) tracking triggers " +
+    "TRACK_HISTORY_DRIFT") {
+    // The mirror of the additive case: removing a selected column shrinks the default tracked set,
+    // which is likewise a tracked-set change requiring a full refresh.
+    createScd2Target("id INT NOT NULL, name STRING, city STRING, seq BIGINT")
+
+    // Run #1: source (id, name, city, seq); recorded tracked set = {name, city, seq}.
+    val stream1 = MemoryStream[(Int, String, String, Long)]
+    stream1.addData((1, "a", "nyc", 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "city", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2))
+
+    // Run #2: `city` dropped from the source; default tracking now resolves to {name, seq}.
+    val stream2 = MemoryStream[(Int, String, Long)]
+    stream2.addData((1, "a", 2L))
+    val ctx2 = singleAutoCdcFlowPipeline(
+      flowName = "flow_v2",
+      target = "target",
+      sourceDf = stream2.toDF().toDF("id", "name", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2)
+
+    val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+    checkErrorInPipelineFailure(
+      failure = ex,
+      condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+      sqlState = Some("42000"),
+      parameters = Map(
+        "tableName" -> targetName,
+        "expectedTrackHistoryColumns" -> "name, seq",
+        "recordedTrackHistoryColumns" -> "name, city, seq"))
+  }
+
+  test("dropping the target (but not the auxiliary table) between runs still detects drift") {
+    // A user drops and recreates the target to reset it, but does not know to also drop the
+    // internal auxiliary table. On the next run the target is absent (so it is re-created), but the
+    // stale auxiliary table survives with the old recorded configuration. Drift validation reads
+    // the auxiliary table, so it must still fire regardless of the target's existence -- otherwise
+    // the aux table's additive evolve would silently overwrite the recorded track-history property
+    // with the new run's value. Mirrors AutoCdcScd1AuxiliaryTableDurabilitySuite's
+    // "auxiliary table is dropped between runs" case, with the two tables swapped.
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")
+
+    // Run #1: track history on `name`; records trackHistoryColumnNames = [name] on the aux table.
+    val stream1 = MemoryStream[(Int, String, Int, Long)]
+    stream1.addData((1, "a", 10, 1L))
+    runPipeline(singleAutoCdcFlowPipeline(
+      flowName = "flow_v1",
+      target = "target",
+      sourceDf = stream1.toDF().toDF("id", "name", "amount", "seq"),
+      keys = Seq("id"),
+      sequencing = $"seq",
+      scdType = ScdType.Type2,
+      trackHistorySelection =
+        Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+    // Drop ONLY the target; the auxiliary table survives with its recorded config.
+    spark.sql(s"DROP TABLE $catalog.$namespace.target")
+    assert(spark.catalog.tableExists(auxTableNameFor("target")),
+      "auxiliary table should survive dropping the target")
+
+    // Run #2: recreate the target and track history on `amount` instead -- a changed tracked set.
+    createScd2Target("id INT NOT NULL, name STRING, amount INT, seq BIGINT")

Review Comment:
   you are so right, fixing



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3693059428


##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -227,13 +239,27 @@ object AutoCdcAuxiliaryTable {
       StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
     val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+ deletedByBatchIdField)
 
+    // The effective track-history column set, resolved by the flow from its user-selected source
+    // schema (see [[AutoCdcMergeFlow.trackHistoryColumnNames]]) -- NOT recomputed here from the
+    // evolved target schema, which would keep tracking columns the flow no longer selects and would
+    // miss implicit (default / `* EXCEPT`) tracked-set changes. A change in this set reinterprets
+    // which transitions open a new historical record, so it is drift-checked.
+    val trackHistoryColumnNames = inputAutoCdcFlow.trackHistoryColumnNames.getOrElse(
+      throw SparkException.internalError(
+        "SCD2 AutoCDC flow is missing its resolved track-history column set."
+      )
+    )
+
     val scd2AuxiliaryTableProperties =
       // Record which SCD strategy this auxiliary table serves so downstream readers can identify it
       // without inspecting the schema.
       Map(scdTypePropertyKey -> ScdType.Type2.label) ++
       // Persist the AutoCDC key column names as a JSON list; immutable post-creation (full-refresh
       // is the only way to change it).
       Map(keyColumnNamesProperty -> serializeKeyColumnNames(keyFields.map(_.name))) ++
+      // Persist the resolved track-history column names; a change reinterprets already-reconciled
+      // history, so it is immutable post-creation (full-refresh is the only way to change it).
+      Map(trackHistoryColumnNamesProperty -> serializeKeyColumnNames(trackHistoryColumnNames)) ++

Review Comment:
   renamed!



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58391][SDP] Validate AutoCDC sequencing-type and SCD2 track-history drift [spark]

Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688709867


##########
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:
   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.
     



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org