szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3715614515
##########
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-
+ // sensitive fold (e.g. Turkish dotless-i) would diverge from how the rest
of the engine
+ // compares the same names.
+ def normalize(name: String): String = {
+ if (caseSensitive) name else name.toLowerCase(Locale.ROOT)
+ }
+
+ // Map each schema by its normalized name. Column identity (add vs. delete
vs. update) is keyed
+ // off the normalized name, while the current column's original-cased name
is what we emit in
+ // the change so we address the column as it actually exists in the
catalog.
def getFieldMap(schema: StructType): Map[String, StructField] = {
- schema.fields.map(field => field.name -> field).toMap
+ schema.fields.map(field => normalize(field.name) -> field).toMap
Review Comment:
`getFieldMap` keys on the normalized name and builds the map with `.toMap`,
so two fields differing only in case collapse into a single entry and the later
one silently wins.
A single flow is enough to reach it. With `events(id INT)` already
materialized:
```sql
CREATE STREAMING TABLE events AS
SELECT id, a AS value, b AS Value FROM STREAM src; -- value STRING,
Value INT
```
A DataFrame can carry two case-differing columns under a case-insensitive
session -- `DataFrameToSchemaSuite`'s "negative: ambiguous column" builds
`toDF("i", "I")` on the default conf and only fails later, once something
resolves the name -- and nothing checks a projection for duplicate output
names. So this flow's schema has both spellings, `mergeSchemas` appends both
(neither case-matches the existing `id`), and `diffSchemas` then emits a single
`addColumn("Value", IntegerType)`: `value STRING` is never created, with no
error and no log line. Which of the two survives depends on field order, so
reordering the SELECT list changes the resulting table.
This one is independent of the inference gap in my other comment:
`inferSchemaFromFlows` folds from an empty `StructType`, and `StructType.merge`
appends every right field absent from the left regardless of `caseSensitive`,
so both spellings survive inference either way.
Suggest failing on a duplicate normalized name here rather than picking a
winner -- or dropping the `caseSensitive` parameter from `diffSchemas`
altogether, which would also address my comment on line 652.
--
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]
