szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3715637039
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -170,14 +171,20 @@ 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 (the first
+ * flow's spelling wins) rather than both, which would otherwise produce a
target schema the
+ * engine's own resolver cannot disambiguate.
*/
lazy val inferredSchema: Map[TableIdentifier, StructType] = {
+ val caseSensitive =
SparkSession.active.sessionState.conf.caseSensitiveAnalysis
Review Comment:
Would it be worth passing this in rather than reading the ambient session
here?
Two things make `SparkSession.active` awkward at this spot. It is read
inside a `lazy val`, so the value is captured whenever `inferredSchema` is
first forced -- a `DataflowGraph`'s contents then depend on the ambient
session's conf at an arbitrary time and thread rather than on the graph itself.
And `SparkSession.active` throws `SparkException.internalError` when there is
no active or default session (`SparkSession.scala:1034`), so forcing this val
from a thread without one now fails where it previously did not.
The same applies to the `caseSensitive` default on `inferSchemaFromFlows`:
as a default parameter value it is evaluated at each call site, so
`VirtualTableInput.load` and `GraphValidations` silently inherit whatever
session happens to be ambient -- and `VirtualTableInput` already carries its
own `spark`, so there would be two session-access paths in the same area.
There is also a subtler mismatch: flow schemas are analyzed under the flow's
own `sqlConf` (`FlowAnalysis.scala:260`), so a flow can produce its schema
under one case-sensitivity and have it merged here under another. Same concern
as my comment on `DatasetManager.scala:401`.
Threading `caseSensitive` down from a caller that already holds a session or
context would avoid all three.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -170,14 +171,20 @@ 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 (the first
+ * flow's spelling wins) rather than both, which would otherwise produce a
target schema the
+ * engine's own resolver cannot disambiguate.
*/
lazy val inferredSchema: Map[TableIdentifier, StructType] = {
+ val caseSensitive =
SparkSession.active.sessionState.conf.caseSensitiveAnalysis
Review Comment:
On the shorter-form question: `SQLConf.get.caseSensitiveAnalysis` is the
usual shorthand for this, but I would avoid it at this spot -- with no active
session it falls back to a thread-local default `SQLConf` rather than failing
(documented on `SQLConf.get`, `SQLConf.scala:251`), and since
`spark.sql.caseSensitive` defaults to false that turns a missing session into
silently case-insensitive merging.
The flow's own session looks like the better source, and there is precedent
in this package: `AutoCdcAuxiliaryTable.scala:146` reads
`inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver`. Here
`inferredSchema` already has the resolved flow in hand and `ResolvedFlow`
exposes `df` (`Flow.scala:214`), so the flag can come from the session that
actually produced the schema being merged.
That also closes the per-flow gap rather than relocating it. Per the
`analyze` scaladoc (`FlowAnalysis.scala:83-86`), a flow's SQL confs are
installed on the analyzing thread via `SQLConf.withExistingConf`, so a flow's
schema is produced under the flow's own conf -- but `inferredSchema` runs
outside that scope, so `SparkSession.active` (and `SQLConf.get` equally) see
the ambient session's conf instead.
Threading `caseSensitive` in as a parameter from a caller that already holds
a session would be cleaner still -- materialization already has `context.spark`
at `DatasetManager.scala:401`.
--
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]
