szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3723376467
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -259,10 +271,12 @@ case class DataflowGraph(
def resolved: Boolean =
flows.forall(f => resolvedFlow.contains(f.identifier))
- def resolve(): DataflowGraph =
+ def resolve(sessionCaseSensitive: Boolean): DataflowGraph =
Review Comment:
Suggestion: either update the Connect server suites for the new signature,
or make `sessionCaseSensitive` a field on `DataflowGraph` set where the graph
is constructed, so `resolve`/`validate`/`inferSchemas` don't need the parameter.
`build/sbt connect/Test/compile` on this ref fails with 28 errors, 23 of
them `not enough arguments for method resolve: (sessionCaseSensitive: Boolean)`
-- 21 in `PythonPipelineSuite.scala`, 2 in
`SparkDeclarativePipelinesServerSuite.scala` (lines 268, 315) -- plus 4
cascading `value map is not a member of Any` at
`PythonPipelineSuite.scala:396,401,440,445` and one fatal unused-import at
`:46`. No `validate` error is reported yet, because every `.validate()` there
is chained onto a `.resolve()` that already failed, so more sites need updating
than the error count suggests. CI will show this as a Connect build failure
rather than a test failure.
On the shape: I argued earlier for making this explicit rather than
inherited, and threading it is a faithful reading of that. The one thing I'd
weigh is that all 23 of those sites, plus the five in `FlowExecution`, would
pass the same session-derived value -- so if you'd rather not carry the
parameter, a field set from `context.spark` keeps the single authoritative
source while removing the per-hop plumbing. Either way is fine by me; the
compile break is the actionable part.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala:
##########
@@ -186,7 +186,11 @@ case class VirtualTableInput(
// Otherwise infer the schema from a combination of the incoming flows
and the
// user-specified schema, if provided.
case _ =>
- SchemaInferenceUtils.inferSchemaFromFlows(availableFlows,
specifiedSchema)
+ SchemaInferenceUtils.inferSchemaFromFlows(
+ tableIdentifier = identifier,
+ flows = availableFlows,
+ userSpecifiedSchema = specifiedSchema,
+ sessionCaseSensitive = spark.sessionState.conf.caseSensitiveAnalysis)
Review Comment:
Suggestion: give `VirtualTableInput` a `sessionCaseSensitive` field and use
it here rather than reading the active session. `CoreDataflowNodeProcessor`
constructs it at line 85 and already takes the value as a constructor
parameter, so it's a one-field thread-through.
This is the last `inferSchemaFromFlows` caller still reading the ambient
session -- `spark` here is `SparkSession.getActiveSession.get` (line 46). The
rest of this commit removed exactly that pattern, so leaving one inference path
on the active session means this table's schema can be inferred under a
different `spark.sql.caseSensitive` than the graph's other tables whenever the
thread's active session isn't the update context's.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -232,7 +232,9 @@ class StreamingTableWrite(
override def getOrigin: QueryOrigin = flow.origin
def startStream(): StreamingQuery = {
- val data = graph.reanalyzeFlow(flow).df
+ val data = graph.reanalyzeFlow(
+ flow,
+ updateContext.spark.sessionState.conf.caseSensitiveAnalysis).df
Review Comment:
Suggestion: hoist this into a single val on `FlowExecution` -- `private val
sessionCaseSensitive =
updateContext.spark.sessionState.conf.caseSensitiveAnalysis` -- and use it at
all five `reanalyzeFlow` calls (237, 267, 307, 339, 386).
Nit, readability only. The five copies also read as though they might not
agree: the `BatchTableWrite` one at 267 evaluates inside
`SparkSessionUtils.withSqlConf(spark, sqlConf.toList: _*)`, so it sees the
session after the flow's confs are applied, while the other four see it before.
They can't actually diverge -- if the flow sets the key,
`effectiveCaseSensitivity` uses the flow's own value and never reaches the
fallback; if it doesn't, `withSqlConf` leaves the value untouched -- but a
single val saves the next reader from working that out.
--
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]
