[Spark] Allow overriding StructField metadata in mergeDataTypes (delta-io#4890)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
Currently, `SchemaMergingUtils.mergeDataTypes` will always use the
existing metadata when merging a new `StructField` with the same name as
an existing field. This change provides a new flag `overrideMetadata`
that allows the update schema to override the metadata in the current
schema.
## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
New unit test in `SchemaUtilsSuite`.scala
## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->
No

Author: aldenlau-db

spark/src/main/scala/org/apache/spark/sql/delta/schema/SchemaMergingUtils.scala

@@ -163,7 +163,8 @@ object SchemaMergingUtils {
       keepExistingType,
       typeWideningMode,
       caseSensitive,
-      allowOverride = false
+      allowOverride = false,
+      overrideMetadata = false
     ).asInstanceOf[StructType]
   }
 
@@ -184,6 +185,8 @@ object SchemaMergingUtils {
    * @param caseSensitive Whether we should keep field mapping case-sensitively.
    *                      This should default to false for Delta, which is case insensitive.
    * @param allowOverride Whether to let incoming type override the existing type if unmatched.
+   * @param overrideMetadata Whether to let metadata of new fields override the existing
+   *                         metadata of matching fields
    */
   def mergeDataTypes(
       current: DataType,
@@ -192,7 +195,8 @@ object SchemaMergingUtils {
       keepExistingType: Boolean,
       typeWideningMode: TypeWideningMode,
       caseSensitive: Boolean,
-      allowOverride: Boolean): DataType = {
+      allowOverride: Boolean,
+      overrideMetadata: Boolean): DataType = {
     def merge(current: DataType, update: DataType): DataType = {
       (current, update) match {
         case (StructType(currentFields), StructType(updateFields)) =>
@@ -202,11 +206,14 @@ object SchemaMergingUtils {
             updateFieldMap.get(currentField.name) match {
               case Some(updateField) =>
                 try {
+                  val updatedCurrentFieldMetadata =
+                    if (overrideMetadata) updateField.metadata
+                    else currentField.metadata
                   StructField(
                     currentField.name,
                     merge(currentField.dataType, updateField.dataType),
                     currentField.nullable,
-                    currentField.metadata)
+                    updatedCurrentFieldMetadata)
                 } catch {
                   case NonFatal(e) =>
                     throw new DeltaAnalysisException(

spark/src/test/scala/org/apache/spark/sql/delta/schema/SchemaUtilsSuite.scala

@@ -2486,7 +2486,8 @@ class SchemaUtilsSuite extends QueryTest
         keepExistingType = false,
         typeWideningMode = TypeWideningMode.NoTypeWidening,
         caseSensitive = false,
-        allowOverride = false)
+        allowOverride = false,
+        overrideMetadata = false)
     assert(mergedType1 === ArrayType(new StructType().add("a", IntegerType).add("b", IntegerType)))
 
     // Map root type
@@ -2506,7 +2507,8 @@ class SchemaUtilsSuite extends QueryTest
         keepExistingType = false,
         typeWideningMode = TypeWideningMode.NoTypeWidening,
         caseSensitive = false,
-        allowOverride = false)
+        allowOverride = false,
+        overrideMetadata = false)
     assert(mergedType2 ===
       MapType(
         new StructType().add("a", IntegerType).add("b", IntegerType),
@@ -2526,7 +2528,8 @@ class SchemaUtilsSuite extends QueryTest
         keepExistingType = false,
         typeWideningMode = TypeWideningMode.NoTypeWidening,
         caseSensitive = false,
-        allowOverride = true)
+        allowOverride = true,
+        overrideMetadata = false)
     assert(mergedSchema1 === ArrayType(LongType))
 
     // override nested type
@@ -2540,7 +2543,8 @@ class SchemaUtilsSuite extends QueryTest
         keepExistingType = false,
         typeWideningMode = TypeWideningMode.NoTypeWidening,
         caseSensitive = false,
-        allowOverride = true)
+        allowOverride = true,
+        overrideMetadata = false)
     assert(mergedSchema2 ===
       ArrayType(new StructType().add("a", MapType(StringType, StringType)).add("b", StringType)))
   }
@@ -2824,6 +2828,60 @@ class SchemaUtilsSuite extends QueryTest
     }
   }
 
+  test("schema merging override field metadata") {
+    val base1 = new StructType()
+      .add("a", IntegerType)
+    val update1 = new StructType()
+      .add("a", IntegerType, nullable = true, new MetadataBuilder().putString("x", "1").build())
+    val mergedSchema1 =
+      mergeDataTypes(
+        current = base1,
+        update = update1,
+        allowImplicitConversions = false,
+        keepExistingType = false,
+        typeWideningMode = TypeWideningMode.NoTypeWidening,
+        caseSensitive = false,
+        allowOverride = false,
+        overrideMetadata = true
+      )
+    assert(mergedSchema1 ===
+      new StructType()
+        .add("a", IntegerType, nullable = true,
+          new MetadataBuilder().putString("x", "1").build()))
+
+    // override nested metadata
+    val base2 = ArrayType(new StructType()
+      .add("a", new StructType()
+        .add("b", IntegerType)
+        .add("c", IntegerType)))
+    val update2 = ArrayType(new StructType()
+      .add("a", new StructType()
+        .add("b", IntegerType)
+        .add("c", IntegerType, nullable = true,
+          new MetadataBuilder().putString("c_metadata", "2").build()),
+        nullable = true,
+        new MetadataBuilder().putString("a_metadata", "3").build()))
+    val mergedSchema2 =
+      mergeDataTypes(
+        current = base2,
+        update = update2,
+        allowImplicitConversions = false,
+        keepExistingType = false,
+        typeWideningMode = TypeWideningMode.NoTypeWidening,
+        caseSensitive = false,
+        allowOverride = false,
+        overrideMetadata = true
+      )
+    assert(mergedSchema2 ===
+      ArrayType(new StructType()
+        .add("a", new StructType()
+          .add("b", IntegerType)
+          .add("c", IntegerType, nullable = true,
+            new MetadataBuilder().putString("c_metadata", "2").build()),
+          nullable = true,
+          new MetadataBuilder().putString("a_metadata", "3").build())))
+  }
+
   ////////////////////////////
   // transformColumns
   ////////////////////////////