szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3717384097
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -170,15 +171,27 @@ case class DataflowGraph(
/**
* A map of the inferred schema of each table, computed by merging the
analyzed schemas
* of all flows writing to that table.
+ *
+ * The merge honors the session's `spark.sql.caseSensitive`: under
case-insensitive analysis two
+ * flows emitting column names that differ only in case contribute a single
column rather than
+ * both, which would otherwise produce a target schema the engine's own
resolver cannot
+ * disambiguate. Which of the two spellings survives follows the order the
flows are merged in,
+ * which this map does not define, so callers should not depend on a
particular casing.
Review Comment:
This gives up a guarantee the code still provides, and there is now a
consumer that depends on it.
`flowsTo` is `flows.groupBy(_.destinationIdentifier)`, which preserves
encounter order within each group, and the fold below is a left `reduce` -- so
the first flow's spelling does win deterministically, which is what this
sentence said before this commit.
The consumer: for a table with no declared schema `outputSchema` is this
map's value (`DatasetManager.scala:315`), and on the non-merging paths
(materialized views and any full refresh) it reaches `diffSchemas` unchanged,
where column identity is keyed on the exact field name. So if the surviving
spelling ever changes -- reordering flow definitions in source is enough -- the
diff emits `deleteColumn` plus `addColumn` for a column that only changed case.
Those paths replace the data anyway, so this is schema churn rather than data
loss, but it is churn driven by something this doc declares undefined.
Suggest restoring the "the first flow's spelling wins" wording, and putting
the inference assertion back to `toSeq` from `toSet`, rather than documenting
the casing as arbitrary while a consumer keys on it.
Separately, line 176 above still says the merge honors "the session's
`spark.sql.caseSensitive`" -- that is the part this commit changed, since it
now honors `effectiveCaseSensitivity` with the session only as a fallback.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -626,20 +656,36 @@ object DatasetManager extends Logging {
* @param mergeWithExistingSchema whether the effective schema is the merge
of the existing and
* desired schemas (additive evolution)
rather than the desired
* schema as-is.
+ * @param caseSensitive whether the additive schema merge treats
field names differing
+ * only in case as distinct columns. Threaded
from the session's
Review Comment:
Stale after this commit: line 405 now passes
`effectiveCaseSensitivityFor(...)`, which reads the flows' value and falls back
to the session only when no flow declares one, so this is no longer "threaded
from the session's `spark.sql.caseSensitive`". The `effectiveCaseSensitivity`
scaladoc already describes the new behavior accurately.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala:
##########
@@ -110,6 +184,19 @@ object SchemaInferenceUtils {
* 1. New columns that need to be added
* 2. Existing columns that need type updates
*
+ * Column identity is keyed on the exact field name, deliberately NOT on a
case-normalized one.
Review Comment:
The "since both were considered and rejected" framing here, and the `NOTE:
caseSensitive deliberately does not reach diffSchemas` in `evolveTable`, are
review context rather than invariants the next reader needs -- `AGENTS.md` asks
for the latter specifically. The substance is worth keeping: identity is keyed
on exact names, and case folding belongs to the merge. The record of what was
rejected isn't.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -170,15 +171,27 @@ case class DataflowGraph(
/**
* A map of the inferred schema of each table, computed by merging the
analyzed schemas
* of all flows writing to that table.
+ *
+ * The merge honors the session's `spark.sql.caseSensitive`: under
case-insensitive analysis two
+ * flows emitting column names that differ only in case contribute a single
column rather than
+ * both, which would otherwise produce a target schema the engine's own
resolver cannot
+ * disambiguate. Which of the two spellings survives follows the order the
flows are merged in,
+ * which this map does not define, so callers should not depend on a
particular casing.
*/
lazy val inferredSchema: Map[TableIdentifier, StructType] = {
- flowsTo.view.mapValues { flows =>
- flows
+ val sessionCaseSensitive =
SparkSession.active.sessionState.conf.caseSensitiveAnalysis
Review Comment:
This reads `SparkSession.active`, while the other caller of
`effectiveCaseSensitivity` reads `context.spark.sessionState.conf`
(`DatasetManager.scala:598`) -- the same fallback, for the same table, from two
different sources. `PipelineUpdateContextImpl.spark` is captured from
`SparkSession.getActiveSession` at construction so they normally agree, but
nothing enforces it, and a divergence would have inference and evolution fold
case differently for one table. Deriving both from a single source removes the
possibility entirely.
The `caseSensitive` default on `inferSchemaFromFlows` is still
`SparkSession.active` as well.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
