Support struct evolution inside maps

## Description

This PR resolves issue #1641 to allow automatic schema evolution in structs that are inside maps.

Assuming the target and source tables have the following schemas:
target: `id string, map map<int, struct<a: int, b: int>>`
source: `id string, map map<int, struct<a: int, b: int, c: int>>`
```
SET spark.databricks.delta.schema.autoMerge.enabled = true;

MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
```
returns an analysis error today:
```
AnalysisException: cannot resolve 's.map' due to data type mismatch: cannot cast map<string,struct<a:int,b:int>> to map<string,struct<a:int,b:int,c:string>>;
```

With this change, the merge command succeeds and the target table schema evolves to include field `c` inside the map value. The same also works for map keys.

- Tests are added to `MergeIntoSuiteBase` and `MergeIntoSQLSuite` to cover struct evolution inside of maps values and keys.

## Does this PR introduce _any_ user-facing changes?
Yes, struct evolution inside of maps now succeeds instead of failing with an analysis error, see previous example.

Closes #1868

GitOrigin-RevId: 07ce2531e03c4e2fa69e8a34f33ba8d2dc3a0228

spark/src/main/scala/org/apache/spark/sql/delta/UpdateExpressionsSupport.scala

@@ -114,6 +114,31 @@ trait UpdateExpressionsSupport extends CastSupport with SQLConfHelper with Analy
                   ArrayType(toEt, containsNull = true)
                 )
             }
+          case (from: MapType, to: MapType) if !Cast.canCast(from, to) =>
+            // Manually convert map keys and values if the types are not compatible to allow schema
+            // evolution. This is slower than direct cast so we only do it when required.
+            def createMapConverter(convert: (Expression, Expression) => Expression): Expression = {
+              val keyVar = NamedLambdaVariable("keyVar", from.keyType, nullable = false)
+              val valueVar =
+                NamedLambdaVariable("valueVar", from.valueType, from.valueContainsNull)
+              LambdaFunction(convert(keyVar, valueVar), Seq(keyVar, valueVar))
+            }
+
+            var transformedKeysAndValues = fromExpression
+            if (from.keyType != to.keyType) {
+              transformedKeysAndValues =
+                TransformKeys(transformedKeysAndValues, createMapConverter {
+                  (key, _) => castIfNeeded(key, to.keyType, allowStructEvolution)
+                })
+            }
+
+            if (from.valueType != to.valueType) {
+              transformedKeysAndValues =
+                TransformValues(transformedKeysAndValues, createMapConverter {
+                  (_, value) => castIfNeeded(value, to.valueType, allowStructEvolution)
+                })
+            }
+            cast(transformedKeysAndValues, to)
           case (from: StructType, to: StructType)
             if !DataType.equalsIgnoreCaseAndNullability(from, to) && resolveStructsByName =>
             // All from fields must be present in the final schema, or we'll silently lose data.

spark/src/test/scala/org/apache/spark/sql/delta/MergeIntoSuiteBase.scala

@@ -3454,6 +3454,325 @@ abstract class MergeIntoSuiteBase
       """{ "key": "A", "value": [ { "a": { "y": 20, "x": [ { "c": 10, "d": 30, "e": null } ] }, "b": 2 } ] }
           { "key": "B", "value": [ { "a": { "y": 60, "x": [ { "c": 20, "d": 50, "e": null } ] }, "b": 3 } ] }""",
     expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  // Struct evolution inside of map values.
+  testNestedStructsEvolution("new source column in map struct value")(
+    target =
+      """{ "key": "A", "map": { "key": { "a": 1 } } }
+         { "key": "C", "map": { "key": { "a": 3 } } }""",
+    source =
+      """{ "key": "A", "map": { "key": { "a": 2, "b": 2 } } }
+         { "key": "B", "map": { "key": { "a": 1, "b": 2 } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{ "key": "A", "map": { "key": { "a": 2, "b": 2 } } }
+         { "key": "B", "map": { "key": { "a": 1, "b": 2 } } }
+         { "key": "C", "map": { "key": { "a": 3, "b": null } } }""",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  testNestedStructsEvolution("new source column in nested map struct value")(
+    target =
+      """{"key": "A", "map": { "key": { "innerKey": { "a": 1 } } } }
+         {"key": "C", "map": { "key": { "innerKey": { "a": 3 } } } }""",
+    source =
+      """{"key": "A", "map": { "key": { "innerKey": { "a": 2, "b": 3 } } } }
+         {"key": "B", "map": { "key": { "innerKey": { "a": 2, "b": 3 } } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          MapType(StringType, new StructType().add("a", IntegerType)))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          MapType(StringType, new StructType().add("a", IntegerType).add("b", IntegerType)))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          MapType(StringType, new StructType().add("a", IntegerType).add("b", IntegerType)))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{"key": "A", "map": { "key": { "innerKey": { "a": 2, "b": 3 } } } }
+         {"key": "B", "map": { "key": { "innerKey": { "a": 2, "b": 3 } } } }
+         {"key": "C", "map": { "key": { "innerKey": { "a": 3, "b": null } } } }""",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  testNestedStructsEvolution("source map struct value contains less columns than target")(
+    target =
+      """{ "key": "A", "map": { "key": { "a": 1, "b": 1 } } }
+         { "key": "C", "map": { "key": { "a": 3, "b": 1 } } }""",
+    source =
+      """{ "key": "A", "map": { "key": { "a": 2 } } }
+         { "key": "B", "map": { "key": { "a": 1 } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{ "key": "A", "map": { "key": { "a": 2, "b": null } } }
+         { "key": "B", "map": { "key": { "a": 1, "b": null } } }
+         { "key": "C", "map": { "key": { "a": 3, "b": 1 } } }""",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  testNestedStructsEvolution("source nested map struct value contains less columns than target")(
+    target =
+      """{"key": "A", "map": { "key": { "innerKey": { "a": 1, "b": 1 } } } }
+         {"key": "C", "map": { "key": { "innerKey": { "a": 3, "b": 1 } } } }""",
+    source =
+      """{"key": "A", "map": { "key": { "innerKey": { "a": 2 } } } }
+         {"key": "B", "map": { "key": { "innerKey": { "a": 2 } } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          MapType(StringType, new StructType().add("a", IntegerType).add("b", IntegerType)))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          MapType(StringType, new StructType().add("a", IntegerType)))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          MapType(StringType, new StructType().add("a", IntegerType).add("b", IntegerType)))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{"key": "A", "map": { "key": { "innerKey": { "a": 2, "b": null } } } }
+         {"key": "B", "map": { "key": { "innerKey": { "a": 2, "b": null } } } }
+         {"key": "C", "map": { "key": { "innerKey": { "a": 3, "b": 1 } } } }""",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  testNestedStructsEvolution("source nested map struct value contains different type than target")(
+    target =
+      """{"key": "A", "map": { "key": { "a": 1, "b" : 1 } } }
+         {"key": "C", "map": { "key": { "a": 3, "b" : 1 } } }""",
+    source =
+      """{"key": "A", "map": { "key": { "a": 1, "b" : "2" } } }
+         {"key": "B", "map": { "key": { "a": 2, "b" : "2" } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", StringType))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{"key": "A", "map": { "key": { "a": 1, "b" : 2 } } }
+         {"key": "B", "map": { "key": { "a": 2, "b" : 2 } } }
+         {"key": "C", "map": { "key": { "a": 3, "b" : 1 } } }""",
+    resultWithoutEvolution =
+      """{"key": "A", "map": { "key": { "a": 1, "b" : 2 } } }
+         {"key": "B", "map": { "key": { "a": 2, "b" : 2 } } }
+         {"key": "C", "map": { "key": { "a": 3, "b" : 1 } } }""")
+
+
+  testNestedStructsEvolution("source nested map struct value in different order")(
+    target =
+      """{"key": "A", "map": { "key": { "a" : 1, "b" : 1 } } }
+         {"key": "C", "map": { "key": { "a" : 3, "b" : 1 } } }""",
+    source =
+      """{"key": "A", "map": { "key": { "b" : 2, "a" : 1, "c" : 3 } } }
+         {"key": "B", "map": { "key": { "b" : 2, "a" : 2, "c" : 4 } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType).add("c", IntegerType))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType).add("b", IntegerType).add("c", IntegerType))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{"key": "A", "map": { "key": { "a": 1, "b" : 2, "c" : 3 } } }
+         {"key": "B", "map": { "key": { "a": 2, "b" : 2, "c" : 4 } } }
+         {"key": "C", "map": { "key": { "a": 3, "b" : 1, "c" : null } } }""",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  testNestedStructsEvolution("source map struct value to map array value")(
+    target =
+      """{ "key": "A", "map": { "key": [ 1, 2 ] } }
+         { "key": "C", "map": { "key": [ 3, 4 ] } }""",
+    source =
+      """{ "key": "A", "map": { "key": { "a": 2 } } }
+         { "key": "B", "map": { "key": { "a": 1 } } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          ArrayType(IntegerType))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          new StructType().add("a", IntegerType))),
+    clauses = update("*") :: insert("*") :: Nil,
+    expectErrorContains = "Failed to merge incompatible data types",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  testNestedStructsEvolution("source struct nested in map array values contains more columns in different order")(
+    target =
+      """{ "key": "A", "map": { "key": [ { "a": 1, "b": 2 } ] } }
+         { "key": "C", "map": { "key": [ { "a": 3, "b": 4 } ] } }""",
+    source =
+      """{ "key": "A", "map": { "key": [ { "b": 6, "c": 7, "a": 5 } ] } }
+         { "key": "B", "map": { "key": [ { "b": 9, "c": 10, "a": 8 } ] } }""",
+    targetSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          ArrayType(
+            new StructType().add("a", IntegerType).add("b", IntegerType)))),
+    sourceSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          ArrayType(
+          new StructType().add("a", IntegerType).add("b", IntegerType).add("c", IntegerType)))),
+    resultSchema = new StructType()
+      .add("key", StringType)
+      .add("map", MapType(
+          StringType,
+          ArrayType(
+          new StructType().add("a", IntegerType).add("b", IntegerType).add("c", IntegerType)))),
+    clauses = update("*") :: insert("*") :: Nil,
+    result =
+      """{ "key": "A", "map": { "key": [ { "a": 5, "b": 6, "c": 7 } ] } }
+         { "key": "B", "map": { "key": [ { "a": 8, "b": 9, "c": 10 } ] } }
+         { "key": "C", "map": { "key": [ { "a": 3, "b": 4, "c": null } ] } }""",
+    expectErrorWithoutEvolutionContains = "Cannot cast")
+
+  // Struct evolution inside of map keys.
+  testEvolution("new source column in map struct key")(
+    targetData = Seq((1, 2, 3, 4), (3, 5, 6, 7)).toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x"),
+    sourceData = Seq((1, 10, 30, 50, 1), (2, 20, 40, 60, 2)).toDF("key", "a", "b", "c", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b, 'c', c), value) as x"),
+    clauses = update("*") :: insert("*") :: Nil,
+    expected = Seq((1, 10, 30, 50, 1), (2, 20, 40, 60, 2), (3, 5, 6, null, 7))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "c", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b, 'c', c), value) as x"),
+    expectErrorWithoutEvolutionContains = "Cannot cast"
+  )
+
+  testEvolution("source nested map struct key contains less columns than target")(
+    targetData = Seq((1, 2, 3, 4, 5), (3, 6, 7, 8, 9)).toDF("key", "a", "b", "c", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b, 'c', c), value) as x"),
+    sourceData = Seq((1, 10, 50, 1), (2, 20, 60, 2)).toDF("key", "a", "c", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'c', c), value) as x"),
+    clauses = update("*") :: insert("*") :: Nil,
+    expected = Seq((1, 10, null, 50, 1), (2, 20, null, 60, 2), (3, 6, 7, 8, 9))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "c", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b, 'c', c), value) as x"),
+    expectErrorWithoutEvolutionContains = "Cannot cast"
+  )
+
+  testEvolution("source nested map struct key contains different type than target")(
+    targetData = Seq((1, 2, 3, 4), (3, 5, 6, 7)).toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x"),
+    sourceData = Seq((1, 10, "30", 1), (2, 20, "40", 2)).toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x"),
+    clauses = update("*") :: insert("*") :: Nil,
+    expected = Seq((1, 10, 30, 1), (2, 20, 40, 2), (3, 5, 6, 7))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x"),
+    expectedWithoutEvolution = Seq((1, 10, 30, 1), (2, 20, 40, 2), (3, 5, 6, 7))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x")
+  )
+
+  testEvolution("source nested map struct key in different order")(
+    targetData = Seq((1, 2, 3, 4), (3, 5, 6, 7)).toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x"),
+    sourceData = Seq((1, 10, 30, 1), (2, 20, 40, 2)).toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('b', b, 'a', a), value) as x"),
+    clauses = update("*") :: insert("*") :: Nil,
+    expected = Seq((1, 30, 10, 1), (2, 40, 20, 2), (3, 5, 6, 7))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x"),
+    expectedWithoutEvolution = Seq((1, 30, 10, 1), (2, 40, 20, 2), (3, 5, 6, 7))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), value) as x")
+  )
+
+  testEvolution(
+    "source struct nested in map array keys contains more columns")(
+    targetData = Seq((1, 2, 3, 4), (3, 5, 6, 7)).toDF("key", "a", "b", "value")
+      .selectExpr("key", "map(array(named_struct('a', a, 'b', b)), value) as x"),
+    sourceData = Seq((1, 10, 30, 50, 1), (2, 20, 40, 60, 2)).toDF("key", "a", "b", "c", "value")
+      .selectExpr("key", "map(array(named_struct('a', a, 'b', b, 'c', c)), value) as x"),
+    clauses = update("*") :: insert("*") :: Nil,
+    expected = Seq((1, 10, 30, 50, 1), (2, 20, 40, 60, 2), (3, 5, 6, null, 7))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "c", "value")
+      .selectExpr("key", "map(array(named_struct('a', a, 'b', b, 'c', c)), value) as x"),
+    expectErrorWithoutEvolutionContains = "cannot cast"
+  )
+
+  testEvolution("struct evolution in both map keys and values")(
+    targetData = Seq((1, 2, 3, 4, 5), (3, 6, 7, 8, 9)).toDF("key", "a", "b", "d", "e")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b), named_struct('d', d, 'e', e)) as x"),
+    sourceData = Seq((1, 10, 30, 50, 70, 90, 110), (2, 20, 40, 60, 80, 100, 120))
+      .toDF("key", "a", "b", "c", "d", "e", "f")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b, 'c', c), named_struct('d', d, 'e', e, 'f', f)) as x"),
+    clauses = update("*") :: insert("*") :: Nil,
+    expected = Seq((1, 10, 30, 50, 70, 90, 110), (2, 20, 40, 60, 80, 100, 120), (3, 6, 7, null, 8, 9, null))
+      .asInstanceOf[List[(Integer, Integer, Integer, Integer, Integer, Integer, Integer)]]
+      .toDF("key", "a", "b", "c", "d", "e", "f")
+      .selectExpr("key", "map(named_struct('a', a, 'b', b, 'c', c), named_struct('d', d, 'e', e, 'f', f)) as x"),
+    expectErrorWithoutEvolutionContains = "cannot cast"
+  )
   // scalastyle:on line.size.limit
 
   testEvolution("new column with update * and insert non-*")(