szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3715636886
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -177,7 +177,7 @@ case class DataflowGraph(
.map { flow =>
resolvedFlow(flow.identifier).schema
}
- .reduce(SchemaMergingUtils.mergeSchemas)
+ .reduce(SchemaMergingUtils.mergeSchemas(_, _))
Review Comment:
Withdrawing this one -- I was reading 2bcacab and missed 80862d650a, which
threads the conf through `inferredSchema` and `inferSchemaFromFlows` (including
the declared-vs-inferred merge) and adds the multi-flow tests. The example
above now fails at validation with `unableToInferSchema` on the STRING/INT
conflict, which is the right outcome. Sorry for the noise.
The `diffSchemas` half of that comment is independent of the inference fix,
so I moved it to its own comment on `getFieldMap`.
##########
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.
--
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]
