szehon-ho commented on code in PR #57722:
URL: https://github.com/apache/spark/pull/57722#discussion_r3716284338
##########
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:
On whether case-insensitive mode is just expected to collapse: Spark rejects
a single schema holding both spellings rather than picking one, and there is an
error class whose message names the case verbatim -- `Found duplicate field(s)
"<requiredFieldName>": <matchedOrcFields> in case-insensitive mode.`
(`ParquetReadSupport.scala:468`, `OrcUtils.scala:321`). Resolution raises
`AMBIGUOUS_COLUMN_OR_FIELD`, and the DDL/write paths raise
`COLUMN_ALREADY_EXISTS` through `SchemaUtils.checkColumnNameDuplication(...,
conf.resolver)` (`CheckAnalysis.scala:1216`, `rules.scala:318`).
The distinction is that case-insensitive matching is a collapse *across* two
schemas -- a target `value` matching an existing `Value` is the same column,
which is what this PR wants and what `StructType.merge` does. `getFieldMap`
applies that rule *within* each schema, which turns a duplicate into a silent
pick.
Pipelines is the one path that can actually reach that state:
`DatasetManager` calls `catalog.createTable` / `catalog.alterTable` directly
rather than going through logical plans, so none of the guards above run for
pipeline tables.
--
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]
