szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3723197809
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala:
##########
@@ -170,15 +171,27 @@ 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 rather than
+ * both, which would otherwise produce a target schema the engine's own
resolver cannot
+ * disambiguate. Which of the two spellings survives follows the order the
flows are merged in,
+ * which this map does not define, so callers should not depend on a
particular casing.
Review Comment:
You're right, and the mechanism is worse than "not the source order" -- I
was reading the wrong `flows`.
`GraphRegistrationContext` builds `flows` in declaration order, but
`DataflowGraphTransformer.transformDownNodes` replaces it: each worker pushes
its flow onto a `ConcurrentLinkedQueue` as resolution *completes*, and the
graph is then rebuilt from `resolvedFlows.asScala.toSeq`
(`DataflowGraphTransformer.scala:315-335`). Two flows writing to the same table
have no dependency on each other, so they are scheduled in the same batch and
their relative order is whichever thread finishes first. The "we maintain the
topological sort order" comment holds between levels, but says nothing about
siblings -- and siblings are exactly the flows merged here.
So the sentence as written is accurate, and I withdraw the suggestion to
restore the old wording. What it does mean is that the surviving spelling can
differ between two runs of an unchanged pipeline, not just after a source edit.
On the non-merging paths that reaches `diffSchemas` keyed on exact names, so a
run-to-run flip emits `deleteColumn` plus `addColumn` for a column that only
changed case.
If you want a rule you can state, the cheap version is to take the scheduler
out of it: fold over `flows.sortBy(_.identifier.unquotedString)` (or sort
inside `flowsTo`) and document that the lowest flow identifier supplies the
spelling. That keeps the doc honest and removes the run-to-run variance.
Open question, since it would make the ordering moot for the consumer: why
not emit `TableChange.renameColumn` when the only difference between the
current and target field is case? It is in the DSv2 API and pipelines never
uses it today. It keeps the column along with its comment and metadata instead
of dropping and re-adding it, while still adopting the declared casing -- which
is the property you wanted when you rejected normalizing inside `diffSchemas`.
The obvious cost is that it needs the target catalog to support rename; is that
the reason, or just scope?
--
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]
