[SPARK-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema - #57610
Open
anew wants to merge 5 commits into
Open
[SPARK-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema#57610anew wants to merge 5 commits into
anew wants to merge 5 commits into
Conversation
…ToSchema Rework `ColumnSelection.applyToSchema` and its private `lookupFieldIndices` to accept a `Resolver` instead of a `caseSensitive: Boolean`. Callers now pass `conf.resolver` directly, matching how the rest of the AutoCDC code resolves identifiers, and `Scd2BatchProcessor.computeTrackedHistoryColumns` takes a `Resolver` too instead of deriving one from a boolean. The `caseSensitivity` label in `AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA` is preserved via a new `CaseSensitivityLabels.of(resolver)` overload that classifies the resolver by probing it, so error messages are unchanged. This is a pure refactor with no behavior change. Co-authored-by: Opus 4.8
szehon-ho
reviewed
Jul 28, 2026
| * two identifiers differing only in case as equal, so we classify it by probing with such a | ||
| * pair rather than by reference identity; the label then stays correct for any resolver. | ||
| */ | ||
| def of(resolver: Resolver): String = of(caseSensitive = !resolver("a", "A")) |
Member
There was a problem hiding this comment.
Nit: probing with "a" / "A" feels a bit awkward for what is only an error-message label.
Spark already classifies resolvers by reference identity against the two singletons — see SchemaUtils.isCaseSensitiveAnalysis — and SQLConf.resolver only ever returns those. Something like:
def of(resolver: Resolver): String =
of(caseSensitive = resolver == caseSensitiveResolution)(or the same check at the one call site, without a new overload) would match that pattern. The "works for any resolver" rationale seems oversold here: AutoCDC callers pass conf.resolver / the test singletons, and SchemaUtils treats anything else as unreachable. Also Flow.scala still uses of(caseSensitiveAnalysis), so this invents a third classification strategy alongside the boolean overload.
…sitivityLabels.of Address review: classify the resolver for the error-message label by reference identity against the two session resolver singletons, matching the existing SchemaUtils.isCaseSensitiveAnalysis pattern, instead of probing with an "a"/"A" pair. conf.resolver only ever returns one of these two singletons. Co-authored-by: Opus 4.8
…solver overload Remove the boolean `of(caseSensitive)` overload and keep only `of(resolver)`, which classifies by reference identity against the session resolver singletons. Update the three Flow.scala call sites to pass the local `resolver` they already hold, instead of re-reading `conf.caseSensitiveAnalysis`. All AutoCDC error labels now derive from a single resolver value per method. Co-authored-by: Opus 4.8
szehon-ho
reviewed
Jul 29, 2026
| if (caseSensitive) CaseSensitive else CaseInsensitive | ||
| /** | ||
| * Maps a [[Resolver]] to its user-facing label. Classifies by reference identity against the | ||
| * two session resolver singletons, matching `SchemaUtils.isCaseSensitiveAnalysis`; |
Member
There was a problem hiding this comment.
nit: it doesnt match SchemaUtils.isCaseSensitiveAnalysis which throws for unknown resolvers. should we just remove such comment?
Contributor
Author
There was a problem hiding this comment.
I kept the comment but made it throw as well. I think that is more future-proof just in case there is ever a third resolver in the future.
…esolver case The unreachable branch in CaseSensitivityLabels.of threw java.lang.InternalError, a fatal JVM Error that bypasses Spark's NonFatal handling and error-class framework. Switch to SparkException.internalError (INTERNAL_ERROR), consistent with the existing internalError call in this file and with SchemaUtils.isCaseSensitiveAnalysis, which this classification mirrors. Co-authored-by: Opus 4.8
…anch Relocate the "conf.resolver only returns one of these two" note from the method scaladoc to the unreachable else branch it explains, where the reader encounters the internalError throw. Co-authored-by: Opus 4.8
uros-b
approved these changes
Jul 29, 2026
Member
|
Thank you @anew and @szehon-ho! Let's wait for CI |
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
3 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?
This is a follow-up refactor for SDP AutoCDC. It reworks
ColumnSelection.applyToSchema(and its private helperlookupFieldIndices) to accept aResolverinstead of acaseSensitive: Boolean:ColumnSelection.applyToSchema/lookupFieldIndicesnow take aResolver.lookupFieldIndicesresolves each requested column viaschema.fieldNames.indexWhere(resolver(_, name))rather than switching betweenStructType.getFieldIndex/getFieldIndexCaseInsensitiveon a boolean.Scd2BatchProcessor.computeTrackedHistoryColumnsnow takes aResolverdirectly instead of derivingcaseSensitiveResolution/caseInsensitiveResolutionfrom a boolean.spark.sessionState.conf.resolverinstead ofconf.caseSensitiveAnalysis:Scd1BatchProcessor.projectTargetColumnsOntoMicrobatch(2 sites),Scd2BatchProcessor.projectTargetColumnsOntoMicrobatch(2 sites) and itscomputeTrackedHistoryColumnsinstance method, andAutoCdcMergeFlow(theuserSelectedSchemaprojection and the construction-time track-history validation).caseSensitivitymessage parameter ofAUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMAis preserved via a newCaseSensitivityLabels.of(resolver)overload that classifies the resolver by probing it (!resolver("a", "A")), so error messages are byte-for-byte unchanged.Why are the changes needed?
The AutoCDC code already resolves identifiers everywhere else through
conf.resolver(the canonical Spark abstraction for case-aware identifier comparison).applyToSchemawas the odd one out, threading a rawcaseSensitiveboolean and re-deriving a resolver at each layer. Passing theResolverdirectly makes column matching consistent with the rest of the pipeline, removes the boolean-to-resolver round-trips, and letscomputeTrackedHistoryColumnstake the resolver it actually needs rather than reconstructing one.Does this PR introduce any user-facing change?
No. This is a pure refactor with no behavior change; the
caseSensitivitylabel in theAUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMAerror is preserved.How was this patch tested?
Existing tests, updated to pass resolvers instead of booleans:
ChangeArgsSuiteandAutoCdcFlowSuite(65 tests) — including the case-sensitive/insensitive selection and missing-column error-message cases.Scd2BatchProcessor*andScd1BatchProcessorMergeSuite(150 tests) — exercising the microbatch projection and history-tracking paths under both case-sensitivity settings.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)