[SPARK-59015][SQL] Match rebinding names with the fold resolution uses

Rebinding compared captured and current names with the resolver, while name
resolution and refresh validation both key on the `toLowerCase` fold and only
filter resolver candidates afterwards. The two rules disagree: `equalsIgnoreCase`
equates U+017F LONG S with `s`, the fold does not. Capturing `s`, renaming it to
`S` and adding U+017F is therefore a change validation accepts and a fresh query
resolves, but rebinding saw two candidates for `s` and failed the query with
INTERNAL_ERROR.

Match on a shared `SchemaUtils.foldName` instead. This removes the exact-name
preference and the ambiguity branch rather than adding to them: the fold is
single-valued because the duplicate-name check rejects a schema holding two names
that fold alike, so the ambiguity error is now unreachable.

Also:
- Assert `assertNotCached` after the second refresh in the SPARK-54424 cache test.
  `numCachedEntries == 1` held whether the entry was usable or not, so it did not
  record that the entry stops being reused.
- Assert the read schema and the filters reaching the source in a nested-refresh
  test, pinning the nested pruning and pushdown the rebuilt struct currently costs.

Verified by reverting the match to the resolver: the new end-to-end and unit tests
fail with `captured name s matches multiple current names [S, ſ]`.

Co-authored-by: Isaac <no-reply@databricks.com>

Author: yyanyy

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V2TableUtil.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql.connector.catalog
 
-import java.util.Locale
-
 import org.apache.spark.sql.catalyst.SQLConfHelper
 import org.apache.spark.sql.catalyst.analysis.Resolver
 import org.apache.spark.sql.catalyst.expressions.MetadataAttributeWithLogicalName
@@ -198,7 +196,7 @@ private[sql] object V2TableUtil extends SQLConfHelper {
   }
 
   private def normalize(name: String): String = {
-    if (conf.caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
+    SchemaUtils.foldName(name, conf.caseSensitiveAnalysis)
   }
 
   private def resolver: Resolver = conf.resolver

sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala

@@ -529,12 +529,23 @@ private[spark] object SchemaUtils {
     if (field.nullable) s"$name $dataType" else s"$name $dataType NOT NULL"
   }
 
+  /**
+   * Folds a name to the key used to decide whether two names refer to the same column or field.
+   *
+   * This is the identity rule name resolution is built on: `AttributeSeq` looks attributes up by
+   * this key and only then filters the candidates with the resolver, and the duplicate-name checks
+   * above reject a schema that holds two names folding to one key. Matching by folded name
+   * therefore finds exactly the field resolution would, while comparing with the resolver alone
+   * can match several fields a schema is allowed to keep apart (`equalsIgnoreCase` equates U+017F
+   * LONG S with `s`, which this fold does not).
+   */
+  def foldName(name: String, caseSensitiveAnalysis: Boolean): String = {
+    if (caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
+  }
+
   private def index(fields: Array[StructField], resolver: Resolver): Map[String, StructField] = {
-    if (isCaseSensitiveAnalysis(resolver)) {
-      fields.map(field => field.name -> field).toMap
-    } else {
-      fields.map(field => field.name.toLowerCase(Locale.ROOT) -> field).toMap
-    }
+    val caseSensitive = isCaseSensitiveAnalysis(resolver)
+    fields.map(field => foldName(field.name, caseSensitive) -> field).toMap
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CapturedSchemaProjection.scala

@@ -19,11 +19,11 @@ package org.apache.spark.sql.execution.datasources.v2
 
 import org.apache.spark.SparkException
 import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.analysis.Resolver
 import org.apache.spark.sql.catalyst.expressions.{Alias, ArrayTransform, AttributeReference, CreateNamedStruct, Expression, GetStructField, If, IsNull, KnownNotNull, LambdaFunction, Literal, MetadataAttributeWithLogicalName, NamedLambdaVariable, TaggingExpression, TransformKeys, TransformValues, UnresolvedNamedLambdaVariable}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.util.MetadataColumnHelper
 import org.apache.spark.sql.types.{ArrayType, DataType, MapType, Metadata, StructType}
+import org.apache.spark.sql.util.SchemaUtils
 
 /**
  * Rebinds a relation that reads a current table schema to output attributes captured from an
@@ -46,7 +46,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
     // The relation still carries the output captured at analysis time; only its table has been
     // swapped for the current one.
     val capturedOutput = relation.output
-    val resolver = conf.resolver
+    val caseSensitive = conf.caseSensitiveAnalysis
     val current = DataSourceV2Relation.create(
       relation.table,
       relation.catalog,
@@ -56,7 +56,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
     val currentMetadataOutput = current.metadataOutput
     val currentMetadata = capturedOutput.filter(_.isMetadataCol).map { captured =>
       val logicalName = metadataLogicalName(captured)
-      matchName(currentMetadataOutput, logicalName, resolver)(metadataLogicalName)
+      matchName(currentMetadataOutput, logicalName, caseSensitive)(metadataLogicalName)
         .map(pos => currentMetadataOutput(pos))
         .getOrElse {
           // The connector still reports this metadata column, so it can only be absent here
@@ -81,13 +81,13 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
       return relation
     }
 
-    val capturedIndex = new AttributeIndex(capturedOutput, resolver)
+    val capturedIndex = new AttributeIndex(capturedOutput, caseSensitive)
     val reboundOutput = currentOutput.map { currentAttr =>
       capturedIndex.get(currentAttr).filter(canReuse(_, currentAttr)).getOrElse(currentAttr)
     }
     val reboundRelation = relation.copy(output = reboundOutput)
 
-    val reboundIndex = new AttributeIndex(reboundOutput, resolver)
+    val reboundIndex = new AttributeIndex(reboundOutput, caseSensitive)
     val projectList = capturedOutput.map { capturedAttr =>
       val currentAttr = reboundIndex.get(capturedAttr).getOrElse {
         unexpectedSchemaChange(
@@ -102,7 +102,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
             s"nullability changed for captured column ${capturedAttr.name} in ${relation.name}")
         }
         val projected = projectToType(
-          currentAttr, currentAttr.dataType, capturedAttr.dataType, resolver)
+          currentAttr, currentAttr.dataType, capturedAttr.dataType, caseSensitive)
         if (projected.dataType != capturedAttr.dataType ||
           projected.nullable != capturedAttr.nullable) {
           unexpectedSchemaChange(
@@ -122,7 +122,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
       input: Expression,
       from: DataType,
       to: DataType,
-      resolver: Resolver): Expression = {
+      caseSensitive: Boolean): Expression = {
     if (from == to) {
       return input
     }
@@ -131,7 +131,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
       case (fromStruct: StructType, toStruct: StructType) =>
         val structInput = if (input.nullable) KnownNotNull(input) else input
         val fields = toStruct.fields.iterator.flatMap { targetField =>
-          val index = matchName(fromStruct, targetField.name, resolver)(_.name).getOrElse {
+          val index = matchName(fromStruct, targetField.name, caseSensitive)(_.name).getOrElse {
             unexpectedSchemaChange(
               s"captured struct field ${targetField.name} is missing from $fromStruct")
           }
@@ -140,7 +140,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
             GetStructField(structInput, index, Some(sourceField.name)),
             sourceField.dataType,
             targetField.dataType,
-            resolver)
+            caseSensitive)
           val namedValue = if (targetField.metadata == Metadata.empty) {
             // An empty captured value is still an explicit instruction not to inherit metadata
             // from the current GetStructField. CleanupAliases removes an empty-metadata Alias, so
@@ -172,7 +172,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
         ArrayTransform(
           input,
           LambdaFunction(
-            projectToType(element, fromElement, toElement, resolver), Seq(element)))
+            projectToType(element, fromElement, toElement, caseSensitive), Seq(element)))
 
       case (
             MapType(fromKey, fromValue, fromValueContainsNull),
@@ -197,7 +197,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
           // duplicate keys.
           TransformKeys(
             input,
-            LambdaFunction(projectToType(key, fromKey, toKey, resolver), Seq(key, value)))
+            LambdaFunction(projectToType(key, fromKey, toKey, caseSensitive), Seq(key, value)))
         } else {
           input
         }
@@ -214,7 +214,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
           TransformValues(
             withProjectedKeys,
             LambdaFunction(
-              projectToType(value, fromValue, toValue, resolver), Seq(key, value)))
+              projectToType(value, fromValue, toValue, caseSensitive), Seq(key, value)))
         } else {
           withProjectedKeys
         }
@@ -231,37 +231,31 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
   }
 
   /**
-   * Returns the position of the entry whose name matches `target`, if any.
+   * Returns the position of the first entry whose folded name matches `target`, if any.
    *
-   * An exact match wins so that a name binds to itself even when the resolver cannot tell it apart
-   * from another name in the same schema. Duplicate names are rejected by folding with
-   * `toLowerCase` while resolution compares with `equalsIgnoreCase`, so a schema can legally hold
-   * several names the resolver considers equal. Without an exact match the resolver match has to be
-   * unique: nothing here can decide which of two indistinguishable names the captured plan read.
+   * Matching on [[SchemaUtils.foldName]] rather than comparing with the resolver keeps rebinding on
+   * the identity rule the rest of resolution uses, so a captured name binds to the field that
+   * validation matched it to and the field a fresh query would resolve it to. The fold is also
+   * single-valued where the resolver is not: refresh validation rejects a schema holding two names
+   * that fold alike, so at most one candidate can match here.
    */
   private def matchName[T](
       candidates: Seq[T],
       target: String,
-      resolver: Resolver)(name: T => String): Option[Int] = {
-    val exact = candidates.indexWhere(candidate => name(candidate) == target)
-    if (exact >= 0) {
-      return Some(exact)
+      caseSensitive: Boolean)(name: T => String): Option[Int] = {
+    val foldedTarget = SchemaUtils.foldName(target, caseSensitive)
+    val pos = candidates.indexWhere { candidate =>
+      SchemaUtils.foldName(name(candidate), caseSensitive) == foldedTarget
     }
-    val matches = candidates.indices.filter(pos => resolver(name(candidates(pos)), target))
-    if (matches.length > 1) {
-      unexpectedSchemaChange(
-        s"captured name $target matches multiple current names " +
-          matches.map(pos => name(candidates(pos))).mkString("[", ", ", "]"))
-    }
-    matches.headOption
+    if (pos >= 0) Some(pos) else None
   }
 
   /**
    * Indexes attributes by name for the rebinding lookups. Data and metadata attributes are indexed
    * separately because a metadata attribute matches on its logical name, which a data column may
    * also carry.
    */
-  private class AttributeIndex(attributes: Seq[AttributeReference], resolver: Resolver) {
+  private class AttributeIndex(attributes: Seq[AttributeReference], caseSensitive: Boolean) {
     private val dataAttrs = attributes.filterNot(_.isMetadataCol)
     private val metadataAttrs = attributes.filter(_.isMetadataCol)
 
@@ -275,7 +269,7 @@ private[sql] object CapturedSchemaProjection extends SQLConfHelper {
 
     private def find(attrs: Seq[AttributeReference], targetName: String)(
         name: AttributeReference => String): Option[AttributeReference] = {
-      matchName(attrs, targetName, resolver)(name).map(pos => attrs(pos))
+      matchName(attrs, targetName, caseSensitive)(name).map(pos => attrs(pos))
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala

@@ -1585,6 +1585,12 @@ class DataSourceV2DataFrameSuite
     stripAQEPlan(qe.executedPlan).asInstanceOf[T]
   }
 
+  private def scanRelationOf(df: DataFrame): DataSourceV2ScanRelation = {
+    df.queryExecution.optimizedPlan.collectFirst {
+      case scan: DataSourceV2ScanRelation => scan
+    }.get
+  }
+
   private def checkDefaultValues(
       columns: Array[Column],
       expectedDefaultValues: Array[ColumnDefaultValue],
@@ -1726,6 +1732,36 @@ class DataSourceV2DataFrameSuite
     }
   }
 
+  test("rebind a captured column renamed beside an addition the resolver cannot tell apart") {
+    // U+017F LONG S folds to itself under `toLowerCase`, so it is a distinct column name to the
+    // fold that name resolution and refresh validation key on, while `equalsIgnoreCase` equates it
+    // with `s`. Renaming `s` to `S` and adding U+017F is therefore a compatible change, and the
+    // captured `s` must keep reading the renamed column rather than becoming ambiguous.
+    val longS = new String(Character.toChars(0x17f))
+    val t = "testcat.ns1.ns2.tbl"
+    withTable(t) {
+      sql(s"CREATE TABLE $t (id INT, s INT) USING foo")
+
+      // Analyze the plan but do not execute it: a Dataset memoizes its optimized plan, so a collect
+      // before the change would refresh once and never see the change below.
+      val df = spark.table(t).filter("id > 0")
+      assert(df.queryExecution.analyzed.resolved)
+
+      // Spark's own DDL refuses both changes because it checks for an existing field with the
+      // resolver, so apply them through the catalog the way another engine would. Data is written
+      // afterwards because this fixture migrates rows by exact field name and so drops the values
+      // of a renamed column.
+      val cat = catalog("testcat")
+      cat.alterTable(testIdent, TableChange.renameColumn(Array("s"), "S"))
+      cat.alterTable(testIdent, TableChange.addColumn(Array(longS), IntegerType, true))
+      externalAppend(cat, testIdent, InternalRow(2, 20, 99))
+
+      // Reading 99 instead of 20 would mean the captured name bound to the addition.
+      checkAnswer(df, Seq(Row(2, 20)))
+      assert(df.queryExecution.optimizedPlan.output.map(_.name) == Seq("id", "s"))
+    }
+  }
+
   test("refresh reconciles a wider partially-pruned scan with stored temp view output") {
     val t = "testcat.ns1.ns2.tbl"
     withTable(t) {
@@ -1780,6 +1816,57 @@ class DataSourceV2DataFrameSuite
     }
   }
 
+  test("refresh recreates a captured nested schema without nested pruning or filter pushdown") {
+    val t = "testcat.ns1.ns2.tbl"
+    withTable(t) {
+      sql(s"CREATE TABLE $t (id INT, person STRUCT<name: STRING, city: STRING>) USING foo")
+      sql(s"INSERT INTO $t VALUES (1, named_struct('name', 'Alice', 'city', 'SF'))")
+
+      def nestedQuery(): DataFrame =
+        spark.table(t).where("person.name = 'Alice'").select("person.name")
+
+      def personFieldsRead(df: DataFrame): Seq[String] = {
+        val readSchema = scanRelationOf(df).scan.readSchema()
+        readSchema("person").dataType.asInstanceOf[StructType].fieldNames.toSeq
+      }
+
+      // Every filter this source reports is one Spark managed to translate and hand to
+      // `pushFilters`. The scan relation's own `pushedFilters` cannot be used here: it keeps only
+      // fully-pushed filters, and this source evaluates none on an unpartitioned table.
+      def filtersReachingSource(df: DataFrame): Seq[String] = {
+        scanRelationOf(df).scan match {
+          case scan: InMemoryBaseTable#InMemoryBatchScan =>
+            scan.pushedFilters.map(_.toString).toSeq
+          case other =>
+            fail(s"unexpected scan type ${other.getClass.getName}")
+        }
+      }
+
+      // capture the plan, then add a field inside the struct
+      val captured = nestedQuery()
+      assert(captured.queryExecution.analyzed.resolved)
+      sql(s"ALTER TABLE $t ADD COLUMN person.age INT FIRST")
+      sql(s"INSERT INTO $t VALUES (2, named_struct('age', 25, 'name', 'Bob', 'city', 'NY'))")
+
+      val fresh = nestedQuery()
+      checkAnswer(captured, Seq(Row("Alice")))
+      checkAnswer(fresh, Seq(Row("Alice")))
+
+      // a plan analyzed after the change prunes to the one field it reads and translates the
+      // predicate down to the source
+      assert(personFieldsRead(fresh) == Seq("name"))
+      assert(filtersReachingSource(fresh).nonEmpty)
+
+      // Rebinding rebuilds the struct as `If(IsNull(person), null, CreateNamedStruct(...))`, which
+      // the extraction, schema-pruning and pushdown rules cannot see through, so the captured plan
+      // reads the whole current struct and its predicate never reaches the source. Results stay
+      // correct, only wider than necessary. Tighten both assertions to match `fresh` once the
+      // projection uses an optimizer-friendly null-preserving form.
+      assert(personFieldsRead(captured) == Seq("age", "name", "city"))
+      assert(filtersReachingSource(captured).isEmpty)
+    }
+  }
+
   test("refresh recreates a captured schema nested through a map and an array") {
     val t = "testcat.ns1.ns2.tbl"
     withTable(t) {
@@ -3988,9 +4075,13 @@ class DataSourceV2DataFrameSuite
       assert(spark.sharedState.cacheManager.numCachedEntries == 1)
 
       // Rebinding an already rebound plan adds a second projection rather than replacing the first,
-      // so this entry no longer matches the single projection a query rebuilds from its own
-      // captured output and stops being reused. Derived queries must still return the captured
-      // schema and the latest data.
+      // so the retained entry no longer matches the single projection a derived query rebuilds from
+      // its own captured output, and stops being reused. Assert that directly: the entry surviving
+      // is not the same claim as the entry being usable. Flip this back to `assertCached` once
+      // refresh replaces the generated projection instead of nesting inside it.
+      assertNotCached(df.filter("id > 0"))
+
+      // Derived queries must still return the captured schema and the latest data.
       checkAnswer(df.filter("id > 0"), Seq(Row(1, 10), Row(2, 20), Row(3, 30), Row(4, 40)))
 
       // verify latest schema is propagated again