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/28 19:03:05 UTC
[PR] [SPARK-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
anew opened a new pull request, #57610:
URL: https://github.com/apache/spark/pull/57610
### What changes were proposed in this pull request?
This is a follow-up refactor for SDP AutoCDC. It reworks `ColumnSelection.applyToSchema` (and its private helper `lookupFieldIndices`) to accept a `Resolver` instead of a `caseSensitive: Boolean`:
- `ColumnSelection.applyToSchema` / `lookupFieldIndices` now take a `Resolver`. `lookupFieldIndices` resolves each requested column via `schema.fieldNames.indexWhere(resolver(_, name))` rather than switching between `StructType.getFieldIndex` / `getFieldIndexCaseInsensitive` on a boolean.
- `Scd2BatchProcessor.computeTrackedHistoryColumns` now takes a `Resolver` directly instead of deriving `caseSensitiveResolution` / `caseInsensitiveResolution` from a boolean.
- All callers pass `spark.sessionState.conf.resolver` instead of `conf.caseSensitiveAnalysis`: `Scd1BatchProcessor.projectTargetColumnsOntoMicrobatch` (2 sites), `Scd2BatchProcessor.projectTargetColumnsOntoMicrobatch` (2 sites) and its `computeTrackedHistoryColumns` instance method, and `AutoCdcMergeFlow` (the `userSelectedSchema` projection and the construction-time track-history validation).
- The `caseSensitivity` message parameter of `AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA` is preserved via a new `CaseSensitivityLabels.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). `applyToSchema` was the odd one out, threading a raw `caseSensitive` boolean and re-deriving a resolver at each layer. Passing the `Resolver` directly makes column matching consistent with the rest of the pipeline, removes the boolean-to-resolver round-trips, and lets `computeTrackedHistoryColumns` take 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 `caseSensitivity` label in the `AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA` error is preserved.
### How was this patch tested?
Existing tests, updated to pass resolvers instead of booleans:
- `ChangeArgsSuite` and `AutoCdcFlowSuite` (65 tests) — including the case-sensitive/insensitive selection and missing-column error-message cases.
- `Scd2BatchProcessor*` and `Scd1BatchProcessorMergeSuite` (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)
--
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-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57610:
URL: https://github.com/apache/spark/pull/57610#discussion_r3670933846
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala:
##########
@@ -127,8 +131,13 @@ private[pipelines] object CaseSensitivityLabels {
val CaseSensitive: String = "case-sensitive"
val CaseInsensitive: String = "case-insensitive"
- def of(caseSensitive: Boolean): String =
- 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`;
Review 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.
--
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-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #57610:
URL: https://github.com/apache/spark/pull/57610#issuecomment-5122126113
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/8df89f20fc5c800ef4935c1b49a7525ec9df8921
- merged into branch-4.x https://github.com/apache/spark/commit/cddeec4151e9d3474a83c835919a6f259ddf0ccc
*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-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57610: [SPARK-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema
URL: https://github.com/apache/spark/pull/57610
--
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-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #57610:
URL: https://github.com/apache/spark/pull/57610#issuecomment-5113649225
Thank you @anew and @szehon-ho! Let's wait for CI
--
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-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57610:
URL: https://github.com/apache/spark/pull/57610#discussion_r3669944995
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala:
##########
@@ -129,6 +133,13 @@ private[pipelines] object CaseSensitivityLabels {
def of(caseSensitive: Boolean): String =
if (caseSensitive) CaseSensitive else CaseInsensitive
+
+ /**
+ * Maps a [[Resolver]] to its user-facing label. A resolver is case-insensitive iff it treats
+ * 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"))
Review 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:
```scala
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.
--
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-58347][SDP] Thread conf.resolver through ColumnSelection.applyToSchema [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57610:
URL: https://github.com/apache/spark/pull/57610#discussion_r3670534146
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala:
##########
@@ -127,8 +131,13 @@ private[pipelines] object CaseSensitivityLabels {
val CaseSensitive: String = "case-sensitive"
val CaseInsensitive: String = "case-insensitive"
- def of(caseSensitive: Boolean): String =
- 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`;
Review Comment:
nit: it doesnt match SchemaUtils.isCaseSensitiveAnalysis which throws for unknown resolvers. should we just remove such 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