[SPARK-58517][SDP] Honor spark.sql.caseSensitive in table schema evolution

common/utils/src/main/resources/error/error-conditions.json

@@ -1240,6 +1240,14 @@
     ],
     "sqlState" : "KD009"
   },
+  "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY" : {
+    "message" : [
+      "The flows writing to table <tableName> do not agree on '<configKey>': <flowConfigurations>.",
+      "The effective value determines whether column names that differ only in case identify the same column, so the table's schema would depend on the order the flows are evaluated in.",
+      "Set '<configKey>' to the same value for every flow writing to this table."
+    ],
+    "sqlState" : "42KD9"
+  },
   "CONNECT" : {
     "message" : [
       "Generic Spark Connect error."

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala

@@ -74,9 +74,8 @@ object ColumnSelection {
    * @param columnSelection The user-provided selection. `None` is a no-op and returns `schema`
    *                        unchanged.
    * @param resolver        Determines whether two column names are considered equal. Callers
-   *                        should pass the session resolver, e.g.
-   *                        `session.sessionState.conf.resolver`, so column matching stays
-   *                        consistent with `spark.sql.caseSensitive`.
+   *                        should pass the resolver for the effective `spark.sql.caseSensitive`
+   *                        of the operation being validated.
    */
   def applyToSchema(
       schemaName: String,

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala

@@ -143,7 +143,7 @@ object AutoCdcAuxiliaryTable {
   ): AuxiliaryTableSpec = {
     val scd1AuxiliaryTableIdentifier = identifier(targetTable.identifier)
 
-    val resolver = inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver
+    val resolver = inputAutoCdcFlow.effectiveResolver
     val autoCdcKeyColumnNames = inputAutoCdcFlow.changeArgs.keys.map(_.name)
 
     // The auxiliary table should derive its schema from the exact same key/CDC metadata column
@@ -217,7 +217,7 @@ object AutoCdcAuxiliaryTable {
   ): AuxiliaryTableSpec = {
     val scd2AuxiliaryTableIdentifier = identifier(targetTable.identifier)
 
-    val resolver = inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver
+    val resolver = inputAutoCdcFlow.effectiveResolver
     val autoCdcKeyColumnNames = inputAutoCdcFlow.changeArgs.keys.map(_.name)
 
     // Resolve the key fields from the (evolved) target schema, exactly as SCD1 does, so the
@@ -283,7 +283,7 @@ object AutoCdcAuxiliaryTable {
    *
    * @param targetTableSchema the AutoCDC target's evolved schema to resolve against
    * @param fieldName the column name to resolve
-   * @param resolver the session resolver used for case-sensitivity-aware field lookups
+   * @param resolver the effective resolver used for case-sensitivity-aware field lookups
    * @param targetTableIdentifier the AutoCDC target's identifier, named in the error message
    * @param autoCdcFlowIdentifier the AutoCDC flow writing to the target, named in the error message
    * @return the matching field
@@ -415,7 +415,7 @@ object AutoCdcAuxiliaryTable {
    * @param expectedScdType the SCD type of the incoming AutoCDC flow, which determines which inner
    *                        `_cdc_metadata` field carries the recorded sequencing type.
    * @param expectedSequencingType the resolved sequencing type of the incoming AutoCDC flow.
-   * @param resolver the session resolver, used to match the reserved column and inner field names
+   * @param resolver the effective resolver, used to match the reserved column and inner field names
    *                 the same case-aware way as every other schema lookup in this file.
    */
   private[graph] def validateNoTargetSequencingTypeDrift(

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala

@@ -20,9 +20,10 @@ package org.apache.spark.sql.pipelines.graph
 import scala.util.Try
 
 import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.pipelines.graph.DataflowGraph.mapUnique
-import org.apache.spark.sql.pipelines.util.SchemaMergingUtils
+import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils
 import org.apache.spark.sql.types.StructType
 
 /**
@@ -170,15 +171,26 @@ 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 effective `spark.sql.caseSensitive` of the flows writing to each table:
+   * 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.
    */
   lazy val inferredSchema: Map[TableIdentifier, StructType] = {
-    flowsTo.view.mapValues { flows =>
-      flows
-        .map { flow =>
-          resolvedFlow(flow.identifier).schema
-        }
-        .reduce(SchemaMergingUtils.mergeSchemas)
-    }.toMap
+    val sessionCaseSensitive = SparkSession.active.sessionState.conf.caseSensitiveAnalysis
+    flowsTo.map { case (destinationIdentifier, flows) =>
+      val resolvedFlows = flows.map { flow =>
+        resolvedFlow(flow.identifier)
+      }
+      destinationIdentifier -> SchemaInferenceUtils.inferSchemaFromFlows(
+        tableIdentifier = destinationIdentifier,
+        flows = resolvedFlows,
+        userSpecifiedSchema = None,
+        sessionCaseSensitive = sessionCaseSensitive)
+    }
   }
 
   /**

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala

@@ -24,7 +24,7 @@ import org.apache.spark.SparkException
 import org.apache.spark.internal.{Logging, LogKeys}
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, PersistedView}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, PersistedView, Resolver}
 import org.apache.spark.sql.classic.SparkSession
 import org.apache.spark.sql.connector.catalog.{
   CatalogV2Util,
@@ -39,9 +39,12 @@ import org.apache.spark.sql.connector.catalog.CatalogV2Util.v2ColumnsToStructTyp
 import org.apache.spark.sql.connector.expressions.{ClusterByTransform, Expressions, Transform}
 import org.apache.spark.sql.execution.command.CreateViewCommand
 import org.apache.spark.sql.pipelines.graph.QueryOrigin.ExceptionHelpers
-import org.apache.spark.sql.pipelines.util.PipelinesCatalogUtils
+import org.apache.spark.sql.pipelines.util.{
+  PipelinesCatalogUtils,
+  SchemaInferenceUtils,
+  SchemaMergingUtils
+}
 import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils.diffSchemas
-import org.apache.spark.sql.pipelines.util.SchemaMergingUtils
 import org.apache.spark.sql.types.StructType
 
 /**
@@ -149,6 +152,10 @@ object DatasetManager extends Logging {
                       auxiliaryTableSpec = auxiliaryTableSpec,
                       isFullRefresh = isFullRefresh,
                       existingAuxiliaryTable = existingAuxiliaryTable,
+                      // The auxiliary schema is derived from its target's, so it evolves under the
+                      // target's effective case sensitivity.
+                      caseSensitive = effectiveCaseSensitivityFor(
+                        resolvedDataflowGraph, table.identifier, context),
                       context = context
                     )
                 }
@@ -354,6 +361,9 @@ object DatasetManager extends Logging {
     val autoCdcAuxTableSpecOpt = auxiliaryTableSpecOpt.collect {
         case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec
       }
+    val effectiveCaseSensitive = effectiveCaseSensitivityFor(
+      resolvedDataflowGraph, table.identifier, context)
+    val effectiveResolver = SchemaInferenceUtils.resolverFor(effectiveCaseSensitive)
 
     // For an incrementally-updated AutoCDC target, validate that the AutoCDC configuration recorded
     // on the auxiliary table has not drifted, BEFORE anything is created or evolved this run. These
@@ -365,7 +375,7 @@ object DatasetManager extends Logging {
     // Running here turns that into one clear drift error (remedy: full refresh).
     if (isTableIncrementallyUpdated) {
       autoCdcAuxTableSpecOpt.foreach {
-        validateNoAutoCdcAuxConfigDrift(_, existingAuxiliaryTable, context)
+        validateNoAutoCdcAuxConfigDrift(_, existingAuxiliaryTable, effectiveResolver)
       }
     }
 
@@ -387,7 +397,7 @@ object DatasetManager extends Logging {
               targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
               expectedScdType = autoCdcSpec.expectedScdType,
               expectedSequencingType = autoCdcSpec.expectedSequencingType,
-              resolver = context.spark.sessionState.conf.resolver
+              resolver = effectiveResolver
             )
           }
         }
@@ -397,7 +407,8 @@ object DatasetManager extends Logging {
           existingTable = existingTable,
           desiredSchema = outputSchema,
           properties = mergedProperties,
-          mergeWithExistingSchema = isTableIncrementallyUpdated
+          mergeWithExistingSchema = isTableIncrementallyUpdated,
+          caseSensitive = effectiveCaseSensitive
         )
       case None =>
         createTable(
@@ -455,12 +466,15 @@ object DatasetManager extends Logging {
    * @param existingAuxiliaryTable the already-loaded auxiliary table (if it exists), loaded once by
    *                               the caller and shared with the config-drift validation in
    *                               [[materializeTable]] rather than re-loaded here.
+   * @param caseSensitive the effective case sensitivity of the flows writing to the auxiliary
+   *                      table's TARGET, whose schema the auxiliary schema is derived from.
    * @param context the context for the pipeline update.
    */
   private def materializeAuxiliaryTable(
       auxiliaryTableSpec: AuxiliaryTableSpec,
       isFullRefresh: Boolean,
       existingAuxiliaryTable: Option[V2Table],
+      caseSensitive: Boolean,
       context: PipelineUpdateContext): Unit = {
     // Get the DSv2 catalog handler and identifier for the aux table.
     val (catalog, auxiliaryTableIdentifier) =
@@ -511,7 +525,8 @@ object DatasetManager extends Logging {
             existingTable = existingAuxTable,
             desiredSchema = auxiliaryTableSpec.schema,
             properties = auxiliaryTableSpec.properties,
-            mergeWithExistingSchema = true
+            mergeWithExistingSchema = true,
+            caseSensitive = caseSensitive
           )
         case None =>
           createTable(
@@ -541,13 +556,12 @@ object DatasetManager extends Logging {
    * @param existingAuxiliaryTableOpt the already-loaded auxiliary table (if it exists), shared with
    *                                  the caller and [[materializeAuxiliaryTable]] to avoid a
    *                                  redundant load.
-   * @param context the context for the pipeline update.
+   * @param resolver the effective resolver of the flows writing to the AutoCDC target.
    */
   private def validateNoAutoCdcAuxConfigDrift(
       autoCdcSpec: AutoCdcAuxiliaryTableSpec,
       existingAuxiliaryTableOpt: Option[V2Table],
-      context: PipelineUpdateContext): Unit = {
-    val resolver = context.spark.sessionState.conf.resolver
+      resolver: Resolver): Unit = {
     existingAuxiliaryTableOpt.foreach { existingAuxiliaryTable =>
       AutoCdcAuxiliaryTable.validateNoKeyColumnDrift(
         existingAuxiliaryTable = existingAuxiliaryTable,
@@ -569,6 +583,23 @@ object DatasetManager extends Logging {
     }
   }
 
+  /**
+   * The effective `spark.sql.caseSensitive` for schema evolution of `tableIdentifier`, read from
+   * the flows writing to it rather than from the session, so evolution stays consistent with the
+   * flows whose schemas it is evolving (a pipeline-level `SET` never reaches the session). Fails if
+   * those flows disagree; see [[SchemaInferenceUtils.effectiveCaseSensitivity]].
+   */
+  private def effectiveCaseSensitivityFor(
+      resolvedDataflowGraph: DataflowGraph,
+      tableIdentifier: TableIdentifier,
+      context: PipelineUpdateContext): Boolean = {
+    SchemaInferenceUtils.effectiveCaseSensitivity(
+      tableIdentifier = tableIdentifier,
+      flows = resolvedDataflowGraph.flowsTo.getOrElse(tableIdentifier, Seq.empty),
+      sessionCaseSensitive = context.spark.sessionState.conf.caseSensitiveAnalysis
+    )
+  }
+
   /**
    * Loads the table at `identifier` from `catalog`, or `None` if it does not exist. A single
    * `loadTable` guarded by a `NoSuchTableException` catch, rather than a `tableExists` +
@@ -626,20 +657,36 @@ 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 the additive schema merge treats field names differing
+   *                                only in case as distinct columns. Threaded from the flows'
+   *                                effective `spark.sql.caseSensitive` so the merge matches how
+   *                                the flows resolve 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. Only affects the merge
+   *                                (i.e. `mergeWithExistingSchema = true`); the subsequent diff
+   *                                always keys columns on their exact names, so a case-only rename
+   *                                on a non-merging path stays an explicit drop-then-add.
    */
   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
     }
+    // NOTE: `caseSensitive` deliberately does not reach `diffSchemas`. On the incremental path the
+    // merge above has already folded a case-only-differing incoming field onto the persisted one,
+    // so there is nothing left for `diffSchemas` to match case-insensitively; on the non-merging
+    // paths (materialized views, full refresh) `targetSchema` is the declared schema as-is, where
+    // matching case-insensitively would make a case-only rename invisible and freeze the persisted
+    // spelling.
     val columnChanges = diffSchemas(currentSchema, targetSchema)
 
     val existingProperties = existingTable.properties()

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala

@@ -23,7 +23,9 @@ import org.apache.spark.SparkException
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{functions => F, AnalysisException, Column}
 import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.Resolver
 import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.pipelines.autocdc.{
   AutoCdcReservedNames,
   CaseSensitivityLabels,
@@ -33,6 +35,7 @@ import org.apache.spark.sql.pipelines.autocdc.{
   Scd2BatchProcessor,
   ScdType
 }
+import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils
 import org.apache.spark.sql.types.{DataType, StructField, StructType}
 
 /**
@@ -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))
+
   requireReservedPrefixAbsentInSourceColumns()
   requireReservedFrameworkColumnsAbsentInSourceColumns()
 
@@ -263,7 +272,7 @@ class AutoCdcMergeFlow(
       schemaName = "changeDataFeed",
       schema = df.schema,
       columnSelection = changeArgs.columnSelection,
-      resolver = spark.sessionState.conf.resolver
+      resolver = effectiveResolver
     )
     // AutoCDC flows require all key columns to be present in the user-selected source schema,
     // so that they survive into the target table where SCD reconciliation needs them.
@@ -303,7 +312,7 @@ class AutoCdcMergeFlow(
         Some(Scd2BatchProcessor.computeTrackedHistoryColumns(
           schema = userSelectedSchema,
           changeArgs = changeArgs,
-          resolver = spark.sessionState.conf.resolver))
+          resolver = effectiveResolver))
       case ScdType.Type1 => None
     }
 
@@ -405,7 +414,7 @@ class AutoCdcMergeFlow(
    * names that use the reserved Spark AutoCDC prefix.
    */
   private def requireReservedPrefixAbsentInSourceColumns(): Unit = {
-    val resolver = spark.sessionState.conf.resolver
+    val resolver = effectiveResolver
     val reservedPrefix = AutoCdcReservedNames.prefix
 
     def nameContainsReservedPrefix(name: String): Boolean = {
@@ -437,7 +446,7 @@ class AutoCdcMergeFlow(
    * during preprocessing. No-op for SCD1, which has no such columns.
    */
   private def requireReservedFrameworkColumnsAbsentInSourceColumns(): Unit = {
-    val resolver = spark.sessionState.conf.resolver
+    val resolver = effectiveResolver
     val reservedPrefix = AutoCdcReservedNames.prefix
 
     // Only the non-prefixed reserved names need checking here; prefixed ones are already rejected
@@ -469,7 +478,7 @@ class AutoCdcMergeFlow(
    * Validate all keys specified in changeArgs are actually present in the user-selected schema.
    */
   private def requireKeysPresentInSelectedSchema(selectedSchema: StructType): Unit = {
-    val resolver = spark.sessionState.conf.resolver
+    val resolver = effectiveResolver
 
     changeArgs.keys
       .find(key => !selectedSchema.fieldNames.exists(name => resolver(name, key.name)))

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphErrors.scala

@@ -109,4 +109,33 @@ object GraphErrors {
       cause = Option(cause.orNull)
     )
   }
+
+  /**
+   * Throws if the flows writing to one table disagree on a configuration whose value determines
+   * how the table's schema is derived, so that the resulting schema would otherwise depend on the
+   * order the flows happen to be evaluated in.
+   *
+   * @param tableIdentifier the destination table the conflicting flows write to
+   * @param configKey the configuration the flows disagree on
+   * @param valuesByFlow the distinct values, each with the flows that declared it
+   */
+  def conflictingFlowConfigurationError(
+      tableIdentifier: TableIdentifier,
+      configKey: String,
+      valuesByFlow: Map[String, Seq[TableIdentifier]]): AnalysisException = {
+    val rendered = valuesByFlow.toSeq
+      .sortBy(_._1)
+      .map { case (value, flows) =>
+        s"$value (${flows.map(_.unquotedString).sorted.mkString(", ")})"
+      }
+      .mkString("; ")
+    new AnalysisException(
+      errorClass = "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY",
+      messageParameters = Map(
+        "tableName" -> tableIdentifier.unquotedString,
+        "configKey" -> configKey,
+        "flowConfigurations" -> rendered
+      )
+    )
+  }
 }