[SPARK-58313][SDP] Validate SCD2 track-history columns at AutoCDC flow construction - #57490
Closed
anew wants to merge 3 commits into
Closed
[SPARK-58313][SDP] Validate SCD2 track-history columns at AutoCDC flow construction#57490anew wants to merge 3 commits into
anew wants to merge 3 commits into
Conversation
…w construction An SCD2 AutoCDC flow's history-tracking columns (`TRACK HISTORY ON ...`, i.e. ChangeArgs.trackHistorySelection) were only validated when the first microbatch ran reconciliation. An unresolvable or ineligible tracking column (one that is absent, a key, a framework column, or dropped by the column selection) therefore surfaced mid-stream, deep inside Scd2BatchProcessor, rather than eagerly at flow construction. Validate the selection at AutoCdcMergeFlow construction time, mirroring the existing key-presence check. The eligibility + resolution logic is extracted into a schema-based Scd2BatchProcessor.computeTrackedHistoryColumns helper that both the runtime path and the new construction-time validator call, so the two can never diverge. An unresolvable selection fails with the existing AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA error; the check is a no-op when trackHistorySelection is None (all SCD1 flows and unrestricted SCD2 flows). Adds tests in AutoCdcFlowSuite. Co-authored-by: Isaac
…e-track-history-columns # Conflicts: # sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
anew
changed the title
jose-torres
reviewed
Jul 25, 2026
jose-torres
left a comment
jose-torres
left a comment
Contributor
There was a problem hiding this comment.
Please try to refactor to pass conf.resolver all the way through for futureproofing, but if it ends up being hard for some reason it's probably not an issue.
| Scd2BatchProcessor.computeTrackedHistoryColumns( | ||
| schema = df.schema, | ||
| changeArgs = changeArgs, | ||
| caseSensitive = df.sparkSession.sessionState.conf.caseSensitiveAnalysis |
Contributor
There was a problem hiding this comment.
I guess I'm OK with this, it's unlikely we'll introduce some third resolution mode all of the sudden.
| * [[org.apache.spark.sql.pipelines.graph.AutoCdcMergeFlow]] construction time (against the | ||
| * user-selected source schema), so an unresolvable or ineligible selection fails fast with a | ||
| * user-actionable [[org.apache.spark.sql.AnalysisException]] instead of surfacing mid-stream | ||
| * (SPARK-58313). |
Contributor
There was a problem hiding this comment.
nit: spark ticket comment as per other PR
Contributor
Author
There was a problem hiding this comment.
Removed all SPARK-58313 references from the changed files
… and test names Remove the SPARK-58313 references from the source comments and test names/section header, consistent with the other AutoCDC PRs (the merged PR records provenance). Threading conf.resolver through the shared ColumnSelection.applyToSchema (rather than deriving a resolver from a caseSensitive boolean) is left as a follow-up refactor, SPARK-58347, since it reworks a boolean/getFieldIndex-based API shared across the SCD1 and SCD2 code paths and is out of scope here. Co-authored-by: Opus 4.8
Contributor
Author
|
I kept the caseSensitive boolean here and filed SPARK-58347 (https://issues.apache.org/jira/browse/SPARK-58347) for the full refactor. The blocker is that the shared ColumnSelection.applyToSchema / lookupFieldIndices are boolean/getFieldIndex-based and used by 6 call sites across SCD1 + SCD2 + Flow, so this ia a large refactor beyond the scope of this PR |
Contributor
Author
|
This is replaced by #57524 |
jose-torres
pushed a commit
that referenced
this pull request
Jul 27, 2026
…w construction ### What changes were proposed in this pull request? This PR replaces #57490, which had to be abandoned due to severe merge conflicts. It is an exact cherry-pick of the previous commits, plus a fix for the test cases that failed due to the merge conflicts. An SCD2 AutoCDC flow can restrict which columns define a "run" via TRACK HISTORY ON (...), which populate ChangeArgs.trackHistorySelection. Until now, that selection was only resolved when the first microbatch ran, inside Scd2BatchProcessor.computeTrackedHistoryColumns during reconciliation. An unresolvable or ineligible tracking column — one that is absent from the source, is a key, is a reserved framework column, or was dropped by the flow's column_list — therefore surfaced mid-stream rather than at flow construction, unlike every other AutoCDC misconfiguration (keys, column selection, reserved names), which fail eagerly. This PR validates trackHistorySelection at AutoCdcMergeFlow construction time, mirroring the existing requireKeysPresentInSelectedSchema check: - The eligibility + resolution logic is extracted from Scd2BatchProcessor.computeTrackedHistoryColumns into a schema-based companion helper Scd2BatchProcessor.computeTrackedHistoryColumns(schema, changeArgs, caseSensitive). Both the per-microbatch runtime path and the new construction-time validator call it, so the two can never diverge. The refactor is behavior-preserving. - AutoCdcMergeFlow gains requireTrackHistoryColumnsResolvableInSelectedSchema, invoked when deriving the user-selected schema (right after the key-presence check). It runs before the flow's schema is forced, so the actionable error surfaces ahead of the temporary AUTOCDC_SCD2_NOT_SUPPORTED gate and remains correct once SCD2 support lands. - No new error condition: an unresolvable selection reuses the existing AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA (schema name trackHistorySelection). - The check is a no-op when trackHistorySelection is None, which covers all SCD1 flows (enforced by ChangeArgs) and SCD2 flows that do not restrict tracking. ### Why are the changes needed? Deferring this validation to reconciliation means a simple typo or misconfiguration (TRACK HISTORY ON (typo), or tracking a key/excluded column) is not reported at graph analysis time; it only fails once data flows, with an error raised deep in the SCD2 batch processor. Validating at flow construction gives a fail-fast, user-actionable error consistent with the rest of the AutoCDC configuration surface (keys, column_list, reserved names). ### Does this PR introduce any user-facing change? Yes. An AutoCDC SCD2 flow whose TRACK HISTORY ON (...) references a column that is not an eligible history-tracking column (absent, a key, a framework column, or excluded by column_list) now fails at flow construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA instead of failing when the first microbatch runs. Valid selections are unaffected, and there is no change for SCD1 flows. (Note: SCD2 AutoCDC flows are not yet generally supported on master — still gated by AUTOCDC_SCD2_NOT_SUPPORTED — so no released behavior changes.) ### How was this patch tested? New unit tests in AutoCdcFlowSuite covering: an SCD2 flow tracking a non-existent column, a key column (ineligible), and a column dropped by columnSelection are each rejected at construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA; a resolvable selection passes the check (falling through to the SCD2-not-supported gate); and case-sensitive/insensitive resolution behavior. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes #57524 from anew/spark-58313-track-history-v2. Authored-by: Andreas Neumann <andreas.neumann@databricks.com> Signed-off-by: Jose Torres <jtorres@apache.org>
jose-torres
pushed a commit
that referenced
this pull request
Jul 27, 2026
…w construction ### What changes were proposed in this pull request? This PR replaces #57490, which had to be abandoned due to severe merge conflicts. It is an exact cherry-pick of the previous commits, plus a fix for the test cases that failed due to the merge conflicts. An SCD2 AutoCDC flow can restrict which columns define a "run" via TRACK HISTORY ON (...), which populate ChangeArgs.trackHistorySelection. Until now, that selection was only resolved when the first microbatch ran, inside Scd2BatchProcessor.computeTrackedHistoryColumns during reconciliation. An unresolvable or ineligible tracking column — one that is absent from the source, is a key, is a reserved framework column, or was dropped by the flow's column_list — therefore surfaced mid-stream rather than at flow construction, unlike every other AutoCDC misconfiguration (keys, column selection, reserved names), which fail eagerly. This PR validates trackHistorySelection at AutoCdcMergeFlow construction time, mirroring the existing requireKeysPresentInSelectedSchema check: - The eligibility + resolution logic is extracted from Scd2BatchProcessor.computeTrackedHistoryColumns into a schema-based companion helper Scd2BatchProcessor.computeTrackedHistoryColumns(schema, changeArgs, caseSensitive). Both the per-microbatch runtime path and the new construction-time validator call it, so the two can never diverge. The refactor is behavior-preserving. - AutoCdcMergeFlow gains requireTrackHistoryColumnsResolvableInSelectedSchema, invoked when deriving the user-selected schema (right after the key-presence check). It runs before the flow's schema is forced, so the actionable error surfaces ahead of the temporary AUTOCDC_SCD2_NOT_SUPPORTED gate and remains correct once SCD2 support lands. - No new error condition: an unresolvable selection reuses the existing AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA (schema name trackHistorySelection). - The check is a no-op when trackHistorySelection is None, which covers all SCD1 flows (enforced by ChangeArgs) and SCD2 flows that do not restrict tracking. ### Why are the changes needed? Deferring this validation to reconciliation means a simple typo or misconfiguration (TRACK HISTORY ON (typo), or tracking a key/excluded column) is not reported at graph analysis time; it only fails once data flows, with an error raised deep in the SCD2 batch processor. Validating at flow construction gives a fail-fast, user-actionable error consistent with the rest of the AutoCDC configuration surface (keys, column_list, reserved names). ### Does this PR introduce any user-facing change? Yes. An AutoCDC SCD2 flow whose TRACK HISTORY ON (...) references a column that is not an eligible history-tracking column (absent, a key, a framework column, or excluded by column_list) now fails at flow construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA instead of failing when the first microbatch runs. Valid selections are unaffected, and there is no change for SCD1 flows. (Note: SCD2 AutoCDC flows are not yet generally supported on master — still gated by AUTOCDC_SCD2_NOT_SUPPORTED — so no released behavior changes.) ### How was this patch tested? New unit tests in AutoCdcFlowSuite covering: an SCD2 flow tracking a non-existent column, a key column (ineligible), and a column dropped by columnSelection are each rejected at construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA; a resolvable selection passes the check (falling through to the SCD2-not-supported gate); and case-sensitive/insensitive resolution behavior. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes #57524 from anew/spark-58313-track-history-v2. Authored-by: Andreas Neumann <andreas.neumann@databricks.com> Signed-off-by: Jose Torres <jtorres@apache.org> (cherry picked from commit 90b1f89) Signed-off-by: Jose Torres <jtorres@apache.org>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
An SCD2 AutoCDC flow can restrict which columns define a "run" via TRACK HISTORY ON (...), which populate ChangeArgs.trackHistorySelection. Until now, that selection was only resolved when the first microbatch ran, inside Scd2BatchProcessor.computeTrackedHistoryColumns during reconciliation. An unresolvable or ineligible tracking column — one that is absent from the source, is a key, is a reserved framework column, or was dropped by the flow's column_list — therefore surfaced mid-stream rather than at flow construction, unlike every other AutoCDC misconfiguration (keys, column selection, reserved names), which fail eagerly.
This PR validates trackHistorySelection at AutoCdcMergeFlow construction time, mirroring the existing requireKeysPresentInSelectedSchema check:
Why are the changes needed?
Deferring this validation to reconciliation means a simple typo or misconfiguration (TRACK HISTORY ON (typo), or tracking a key/excluded column) is not reported at graph analysis time; it only fails once data flows, with an error raised deep in the SCD2 batch processor. Validating at flow construction gives a fail-fast, user-actionable error consistent with the rest of the AutoCDC configuration surface (keys, column_list, reserved names).
Does this PR introduce any user-facing change?
Yes. An AutoCDC SCD2 flow whose TRACK HISTORY ON (...) references a column that is not an eligible history-tracking column (absent, a key, a framework column, or excluded by column_list) now fails at flow construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA instead of failing when the first microbatch runs. Valid selections are unaffected, and there is no change for SCD1 flows. (Note: SCD2 AutoCDC flows are not yet generally supported on master — still gated by AUTOCDC_SCD2_NOT_SUPPORTED — so no released behavior changes.)
How was this patch tested?
New unit tests in AutoCdcFlowSuite covering: an SCD2 flow tracking a non-existent column, a key column (ineligible), and a column dropped by columnSelection are each rejected at construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA; a resolvable selection passes the check (falling through to the SCD2-not-supported gate); and case-sensitive/insensitive resolution behavior.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)