szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3715961610
##########
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:
Correcting two things in my comments above, having gone back through the
precedent.
First, two of my three objections do not hold up, and I withdraw them.
Conf-derived `lazy val`s are common in Spark (`HistogramNumeric.scala:84`,
`RuleExecutor.scala:223`), and so is capturing the ambient session in a field
(`FileDataSourceV2.scala:54`, `StateDataSource.scala:58`,
`SparkPlan.scala:66`). The throwing behavior of `SparkSession.active` also adds
no new failure mode on this path, since `elements.scala:46` already does
`SparkSession.getActiveSession.get` a few lines from the call site.
Second, I argued against `SQLConf.get` in my reply above. That was wrong for
`SchemaInferenceUtils.scala:55`, and it obscured the more important point.
`inferSchemaFromFlows` is reached from `VirtualTableInput.load`
(`elements.scala:189`), which runs inside the
`SQLConf.withExistingConf(ctx.flowConf)` scope that `FlowAnalysis` installs per
flow -- deliberately, per the comment at `FlowAnalysis.scala:48-53` and the
`analyze` scaladoc at `81-86`, so that flows resolving concurrently on a shared
session do not race on conf. Every other conf read inside that scope honors it:
Catalyst analysis via `SQLConf.get`, and `ctx.flowConf.getConfString` at
`FlowAnalysis.scala:242`. The new default argument is the one read that goes
around it to the shared session, so a flow that sets `spark.sql.caseSensitive`
in its own confs would be analyzed under one value and have its schema merged
under another.
So for that line the fix is just `caseSensitive: Boolean =
SQLConf.get.caseSensitiveAnalysis`, and specifically not threading it from the
caller as I suggested earlier -- `elements.scala:189` only has the shared
session, so passing the value from there would reproduce the same bypass.
For `DataflowGraph.scala:181` the earlier suggestion stands. That one is
forced outside any `withExistingConf` scope, so the flow's own session is the
right source -- `resolvedFlow(flow.identifier).df.sparkSession` -- matching
`AutoCdcAuxiliaryTable.scala:146` in the same package.
For what it is worth on the choice between the two: `SparkSession.active` as
a default parameter value does not appear anywhere else in Spark's main Scala,
and of its usages there, the two added by this PR are the only ones that read a
`SQLConf` value -- the rest need the session itself or something hanging off it
(its `SparkContext`, its `sessionState`, a derived Hadoop conf).
--
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]
