szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3723145870
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -305,8 +313,9 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
+ val sessionCaseSensitive =
context.spark.sessionState.conf.caseSensitiveAnalysis
val outputSchema = table.specifiedSchema.getOrElse(
- resolvedDataflowGraph.inferredSchema(table.identifier).asNullable
+
resolvedDataflowGraph.inferredSchema(sessionCaseSensitive)(table.identifier).asNullable
Review Comment:
`inferredSchema` was a memoized `lazy val`; as a `def` it rebuilds the whole
graph's schema map on every call, and this call site sits inside
`materializeTable`, which `transformTables` invokes once per table. An N-table
pipeline therefore runs N full inference passes over all flows where it
previously ran one -- N+1 counting the map `auxiliaryTableSpecs` builds. It is
pure CPU with no catalog I/O so nothing breaks, but it is quadratic in table
count for no gain.
Lines 103-104 already solve exactly this by hoisting `sessionCaseSensitive`
and `auxiliaryTableSpecs` out of the loop. Suggest the same here: compute
`inferredSchema(sessionCaseSensitive)` alongside them and thread the map (or
just this table's schema) into `materializeTable`.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##########
@@ -252,6 +255,12 @@ class AutoCdcMergeFlow(
val flow: AutoCdcFlow,
val funcResult: FlowFunctionResult
) extends ResolvedFlow {
+ private[graph] val effectiveResolver: Resolver =
SchemaInferenceUtils.resolverFor(
+ sqlConf
+ .get(SQLConf.CASE_SENSITIVE.key)
+ .map(_.trim.toBoolean)
+ .getOrElse(spark.sessionState.conf.caseSensitiveAnalysis))
Review Comment:
This reads `spark.sql.caseSensitive` by hand instead of going through
`SchemaInferenceUtils.effectiveCaseSensitivity`, duplicating the
`_.trim.toBoolean` parsing this PR just centralized and skipping the conflict
check that helper exists to provide. For a single flow the two are exactly
equivalent, so this is pure duplication today; it stays correct only because an
AutoCDC target is single-flow by validation, and if that ever changes this
silently picks one flow's value rather than raising
`CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY`.
`effectiveCaseSensitivity` takes `Seq[Flow]` and reads only `sqlConf` and
`identifier`, both from the constructor args, so passing the flow itself here
is safe:
```scala
private[graph] val effectiveResolver: Resolver =
SchemaInferenceUtils.resolverFor(
SchemaInferenceUtils.effectiveCaseSensitivity(
tableIdentifier = destinationIdentifier,
flows = Seq(this),
sessionCaseSensitive = spark.sessionState.conf.caseSensitiveAnalysis))
```
--
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]
