szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3715659760
##########
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]
