szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3715502969
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -626,21 +628,28 @@ object DatasetManager extends Logging {
* @param mergeWithExistingSchema whether the effective schema is the merge
of the existing and
* desired schemas (additive evolution)
rather than the desired
* schema as-is.
+ * @param caseSensitive whether schema evolution treats field
names differing only in
+ * case as distinct columns. Threaded from
the session's
+ * `spark.sql.caseSensitive` so evolution
matches how the rest of
+ * the engine resolves the same names; when
`false`, an incoming
+ * column differing from an existing one only
in case is folded
+ * onto it rather than added as a duplicate.
*/
private def evolveTable(
catalog: TableCatalog,
tableIdentifier: Identifier,
existingTable: V2Table,
desiredSchema: StructType,
properties: Map[String, String],
- mergeWithExistingSchema: Boolean): Unit = {
+ mergeWithExistingSchema: Boolean,
+ caseSensitive: Boolean): Unit = {
val currentSchema = v2ColumnsToStructType(existingTable.columns())
val targetSchema = if (mergeWithExistingSchema) {
- SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema,
caseSensitive)
} else {
desiredSchema
}
- val columnChanges = diffSchemas(currentSchema, targetSchema)
+ val columnChanges = diffSchemas(currentSchema, targetSchema, caseSensitive)
Review Comment:
I think passing `caseSensitive` here changes behavior for materialized views
in a way the PR doesn't mention. Concretely:
```sql
-- pipeline source, v1
CREATE MATERIALIZED VIEW daily_totals AS
SELECT id, total FROM events;
```
Materializes `daily_totals(id INT, total BIGINT)`. The author then renames
the column's casing:
```sql
-- pipeline source, v2
CREATE MATERIALIZED VIEW daily_totals AS
SELECT id, total AS Total FROM events;
```
In 4.1.0 the next update dropped `total` and added `Total`, so the table
matched the definition. With this change, no `ALTER` is emitted at all and the
table keeps `total` -- permanently, since nothing on this path ever rebuilds
it. The definition says `Total`, `DESCRIBE TABLE` says `total`, and there's no
error pointing at the discrepancy. A colleague who runs the same definition
against a fresh table gets `Total`, because `createTable` persists the declared
casing verbatim -- so the same source yields different column names depending
on whether the table already existed.
The detail: for an MV (and for any full refresh) `targetSchema` is the run's
schema as-is, with no merge, so case-insensitive matching makes a case-only
rename invisible. That's also why it can't help the case this PR fixes -- on
the incremental path `StructType.merge` has already replaced the incoming name
with the persisted one, so a case-differing pair never reaches this call.
Dropping the third argument leaves your `mergeSchemas` fix fully intact.
Is the frozen casing intended? If so, could we make the create path agree,
say so in the `evolveTable` doc, and add a test? If not, dropping the argument
here restores the previous behavior.
##########
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:
You already flagged that multi-flow inference still merges case-sensitively.
Here's the end-to-end it produces, which also turns up a second problem.
Two teams append to a shared events table. An earlier version of the source
selected only `id`, so the table materialized as `events(id INT)`. Then each
team adds a field, and they spell it differently:
```sql
CREATE STREAMING TABLE events AS
SELECT id, value FROM STREAM src_a; -- value is STRING
CREATE FLOW events_extra AS INSERT INTO events BY NAME
SELECT id, Value FROM STREAM src_b; -- Value is INT
```
The next update leaves the table as `events(id INT, Value INT)`: one INT
column, and the STRING column silently never created. `events_extra` writes
fine, while the other flow's `value` STRING resolves case-insensitively onto
the INT column and fails on the type. Which of the two survives depends on
field order, so swapping which query defines the table flips which flow breaks.
And once `Value INT` is persisted, correcting both flows to `value STRING`
makes the merge reject INT against STRING on every later update, so recovery
needs a full refresh.
Worth noting that if both teams had written `value`, inference would have
caught this immediately with a clear `unableToInferSchema` naming the
STRING/INT conflict. Only the case difference gets through.
The detail: inference merges case-sensitively, so both spellings reach the
output schema; `mergeSchemas` appends both, since neither case-matches the
existing `id`; then `diffSchemas` collapses them, because `getFieldMap` keys on
the lower-cased name and `.toMap` lets the last field win. Threading the
session conf into `inferredSchema` / `inferSchemaFromFlows` fixes the root
cause -- the two fold, the type conflict surfaces at validation before the
catalog is touched, and the same-type variant simply works.
Separately, should `diffSchemas` ever silently pick one of two colliding
columns? That stays reachable even with inference fixed:
`CreateStreamingTableHandler` takes `cst.columns` verbatim and nothing checks
for duplicate column names, so a declared `(id INT, value STRING, Value
STRING)` reaches it too. Either `getFieldMap` should fail on a duplicate
normalized name, or the `caseSensitive` parameter should come off `diffSchemas`
entirely -- which would also address my other comment.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -397,7 +397,8 @@ object DatasetManager extends Logging {
existingTable = existingTable,
desiredSchema = outputSchema,
properties = mergedProperties,
- mergeWithExistingSchema = isTableIncrementallyUpdated
+ mergeWithExistingSchema = isTableIncrementallyUpdated,
+ caseSensitive = context.spark.sessionState.conf.caseSensitiveAnalysis
Review Comment:
Suggestion: take this from the pipeline's conf rather than the session's.
A pipeline can set `spark.sql.caseSensitive` for itself, and `SET` in
pipeline source doesn't touch the session -- `SqlGraphRegistrationContext`
records it, `GraphRegistrationContext.registerFlow` folds it into each flow's
`sqlConf` (`defaultSqlConf ++ flowDef.sqlConf`), and it's applied when the flow
is analyzed (`FlowAnalysis.scala:260`) and executed
(`FlowExecution.scala:214`). So the two can disagree:
```sql
SET spark.sql.caseSensitive = true;
CREATE STREAMING TABLE events AS
SELECT id, Value FROM STREAM src;
```
With `events(id INT, value STRING)` already materialized, the pipeline has
explicitly asked for `Value` to be a distinct column, and that's what happened
before this PR. Now evolution reads the session's default `false`, folds
`Value` onto the existing `value`, and emits no change -- after which the flow,
which does resolve case-sensitively, can't find `Value` and the write fails.
The flows writing to this table already carry the effective value in
`flow.sqlConf`, so reading it from there (or from the pipeline's default conf)
would keep evolution consistent with the schemas it is evolving.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala:
##########
@@ -112,25 +114,46 @@ object SchemaInferenceUtils {
*
* @param currentSchema The current schema of the table
* @param targetSchema The target schema that we want the table to have
+ * @param caseSensitive Whether two field names that differ only in case
identify distinct
+ * columns. When `false` (mirroring a case-insensitive
session), a target
+ * field is matched to the current field it differs
from only in case -- so
+ * it is treated as the same column (an in-place update
against the current
+ * column's name) rather than a spurious drop-then-add.
Callers on a
+ * schema-evolution path should pass the session's
`spark.sql.caseSensitive`;
+ * the default `true` preserves the historical
case-sensitive behavior.
* @return A sequence of TableChange objects representing the necessary
changes
*/
- def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] = {
+ def diffSchemas(
+ currentSchema: StructType,
+ targetSchema: StructType,
+ caseSensitive: Boolean = true): Seq[TableChange] = {
val changes = scala.collection.mutable.ArrayBuffer.empty[TableChange]
- // Helper function to get a map of field name to field
+ // Normalize a field name to its lookup key: identity when case-sensitive,
lower-cased when not,
+ // so that a target field is matched to the current field it differs from
only in case. Lower-
+ // case with Locale.ROOT to match StructType.merge and Spark's analyzer
resolver; a locale-
Review Comment:
Nit: suggest dropping "and Spark's analyzer resolver" here -- matching
`StructType.merge` / `fieldsMap` is the accurate claim, and the right one.
The resolver is `equalsIgnoreCase` (`analysis/package.scala:40`), which
isn't the same predicate as folding both sides with `toLowerCase(Locale.ROOT)`;
they disagree where case mapping isn't 1:1, e.g. `equalsIgnoreCase("İ", "i")`
is true but the folded forms aren't equal. No code change needed -- just so a
future reader doesn't rely on the parity.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaMergingUtils.scala:
##########
@@ -20,7 +20,24 @@ package org.apache.spark.sql.pipelines.util
import org.apache.spark.sql.types.StructType
object SchemaMergingUtils {
- def mergeSchemas(tableSchema: StructType, dataSchema: StructType):
StructType = {
- StructType.merge(tableSchema, dataSchema).asInstanceOf[StructType]
+
+ /**
+ * Additively merges `dataSchema` into `tableSchema`, returning a schema
that is the union of the
+ * two (recursing into nested structs/arrays). On a field present in both,
`tableSchema`'s name
+ * and position win; `dataSchema` only contributes fields absent from
`tableSchema`.
+ *
+ * @param caseSensitive whether two field names that differ only in case are
considered distinct.
+ * When `false` (mirroring a case-insensitive session),
`dataSchema`'s field
+ * is folded onto the matching `tableSchema` field
rather than added as a
+ * separate, case-differing column. Callers on a
schema-evolution path should
+ * pass the session's `spark.sql.caseSensitive`; the
default `true` preserves
+ * the historical behavior for callers that
intentionally merge case
+ * sensitively.
+ */
+ def mergeSchemas(
+ tableSchema: StructType,
+ dataSchema: StructType,
+ caseSensitive: Boolean = true): StructType = {
Review Comment:
Would it be worth dropping the `= true` default here and on `diffSchemas`,
and passing the flag explicitly at every call site?
The two places that still merge case-sensitively are exactly the ones that
silently inherited this default rather than choosing it. Without a default
they'd have been compile errors, and each site's choice would be visible in
review. The `mergeSchemas(_, _)` eta-expansion this PR had to add at
`DataflowGraph.scala:180` exists only to preserve the default, which points the
same way.
--
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]
