[SPARK-54525][SQL] Disable nested struct coercion in MERGE INTO under a config

### What changes were proposed in this pull request?
#52225 allow MERGE INTO to support case where assignment value is a struct with less fields than the assignment key, ie UPDATE SET big_struct = source.small_struct.

This makes this feature off by default, and turned on via a config.

### Why are the changes needed?

The change brought some interesting question, for example there is some ambiguity in user intent.  Does the UPDATE SET * mean set all nested fields or top level columns?  In the first case, missing fields are kept.  In the second case, missing fields are nullified.

I tried to make a choice in #53149 but after some feedback, it may be a bit controversial, choosing one interpretation over another.  A SQLConf may not be the right choice, and instead we may need to introduce some new syntax, which require more discussion.

### Does this PR introduce _any_ user-facing change?
No this feature is unreleased

### How was this patch tested?
Existing unit test

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #53229 from szehon-ho/disable_merge_update_source_coercion.

Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 23d9253)
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: szehon-ho

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala

@@ -21,8 +21,7 @@ import scala.collection.mutable
 
 import org.apache.spark.sql.catalyst.SQLConfHelper
 import org.apache.spark.sql.catalyst.analysis.TableOutputResolver.DefaultValueFillMode.{NONE, RECURSE}
-import org.apache.spark.sql.catalyst.expressions.{And, Attribute, CreateNamedStruct, Expression, GetStructField, If, IsNull, Literal}
-import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
+import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateNamedStruct, Expression, GetStructField, IsNull, Literal}
 import org.apache.spark.sql.catalyst.plans.logical.Assignment
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
@@ -182,31 +181,11 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
     } else if (exactAssignments.isEmpty && fieldAssignments.isEmpty) {
       TableOutputResolver.checkNullability(colExpr, col, conf, colPath)
     } else if (exactAssignments.nonEmpty) {
-     if (updateStar) {
-        val value = exactAssignments.head.value
-        col.dataType match {
-          case structType: StructType =>
-            // Expand assignments to leaf fields
-            val structAssignment =
-              applyNestedFieldAssignments(col, colExpr, value, addError, colPath,
-                coerceNestedTypes)
-
-            // Wrap with null check for missing source fields
-            fixNullExpansion(col, value, structType, structAssignment,
-              colPath, addError)
-          case _ =>
-            // For non-struct types, resolve directly
-            val coerceMode = if (coerceNestedTypes) RECURSE else NONE
-            TableOutputResolver.resolveUpdate("", value, col, conf, addError, colPath,
-              coerceMode)
-        }
-       } else {
-         val value = exactAssignments.head.value
-         val coerceMode = if (coerceNestedTypes) RECURSE else NONE
-         val resolvedValue = TableOutputResolver.resolveUpdate("", value, col, conf, addError,
-           colPath, coerceMode)
-         resolvedValue
-       }
+      val value = exactAssignments.head.value
+      val coerceMode = if (coerceNestedTypes) RECURSE else NONE
+      val resolvedValue = TableOutputResolver.resolveUpdate("", value, col, conf, addError,
+        colPath, coerceMode)
+      resolvedValue
     } else {
       applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes)
     }
@@ -240,63 +219,6 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
     }
   }
 
-  private def applyNestedFieldAssignments(
-      col: Attribute,
-      colExpr: Expression,
-      value: Expression,
-      addError: String => Unit,
-      colPath: Seq[String],
-      coerceNestedTyptes: Boolean): Expression = {
-
-    col.dataType match {
-      case structType: StructType =>
-        val fieldAttrs = DataTypeUtils.toAttributes(structType)
-
-        val updatedFieldExprs = fieldAttrs.zipWithIndex.map { case (fieldAttr, ordinal) =>
-          val fieldPath = colPath :+ fieldAttr.name
-          val targetFieldExpr = GetStructField(colExpr, ordinal, Some(fieldAttr.name))
-
-          // Try to find a corresponding field in the source value by name
-          val sourceFieldValue: Expression = value.dataType match {
-            case valueStructType: StructType =>
-              valueStructType.fields.find(f => conf.resolver(f.name, fieldAttr.name)) match {
-                case Some(matchingField) =>
-                  // Found matching field in source, extract it
-                  val fieldIndex = valueStructType.fieldIndex(matchingField.name)
-                  GetStructField(value, fieldIndex, Some(matchingField.name))
-                case None =>
-                  // Field doesn't exist in source, use target's current value with null check
-                  TableOutputResolver.checkNullability(targetFieldExpr, fieldAttr, conf, fieldPath)
-              }
-            case _ =>
-              // Value is not a struct, cannot extract field
-              addError(s"Cannot assign non-struct value to struct field '${fieldPath.quoted}'")
-              Literal(null, fieldAttr.dataType)
-          }
-
-          // Recurse or resolve based on field type
-          fieldAttr.dataType match {
-            case nestedStructType: StructType =>
-              // Field is a struct, recurse
-              applyNestedFieldAssignments(fieldAttr, targetFieldExpr, sourceFieldValue,
-                addError, fieldPath, coerceNestedTyptes)
-            case _ =>
-              // Field is not a struct, resolve with TableOutputResolver
-              val coerceMode = if (coerceNestedTyptes) RECURSE else NONE
-              TableOutputResolver.resolveUpdate("", sourceFieldValue, fieldAttr, conf, addError,
-                fieldPath, coerceMode)
-          }
-        }
-        toNamedStruct(structType, updatedFieldExprs)
-
-      case otherType =>
-        addError(
-          "Updating nested fields is only supported for StructType but " +
-          s"'${colPath.quoted}' is of type $otherType")
-        colExpr
-    }
-  }
-
   private def toNamedStruct(structType: StructType, fieldExprs: Seq[Expression]): Expression = {
     val namedStructExprs = structType.fields.zip(fieldExprs).flatMap { case (field, expr) =>
       Seq(Literal(field.name), expr)
@@ -350,55 +272,6 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
     IsNull(currentExpr)
   }
 
-  /**
-   * As UPDATE SET * can assign struct fields individually (preserving existing fields),
-   * this will lead to null expansion, ie, a struct is created where all fields are null.
-   * Wraps a struct assignment with null checks for the source and missing source fields.
-   * Return null if all are null.
-   *
-   * @param col the target column attribute
-   * @param value the source value expression
-   * @param structType the target struct type
-   * @param structAssignment the struct assignment result to wrap
-   * @param colPath the column path for error reporting
-   * @param addError error reporting function
-   * @return the wrapped expression with null checks
-   */
-  private def fixNullExpansion(
-      col: Attribute,
-      value: Expression,
-      structType: StructType,
-      structAssignment: Expression,
-      colPath: Seq[String],
-      addError: String => Unit): Expression = {
-    // As StoreAssignmentPolicy.LEGACY is not allowed in DSv2, always add null check for
-    // non-nullable column
-    if (!col.nullable) {
-      AssertNotNull(value)
-    } else {
-      // Check if source struct is null
-      val valueIsNull = IsNull(value)
-
-      // Check if missing source paths (paths in target but not in source) are not null
-      // These will be null for the case of UPDATE SET * and
-      val missingSourcePaths = getMissingSourcePaths(structType, value.dataType, colPath, addError)
-      val condition = if (missingSourcePaths.nonEmpty) {
-        // Check if all target attributes at missing source paths are null
-        val missingFieldNullChecks = missingSourcePaths.map { path =>
-          createNullCheckForFieldPath(col, path)
-        }
-        // Combine all null checks with AND
-        val allMissingFieldsNull = missingFieldNullChecks.reduce[Expression]((a, b) => And(a, b))
-        And(valueIsNull, allMissingFieldsNull)
-      } else {
-        valueIsNull
-      }
-
-      // Return: If (condition) THEN NULL ELSE structAssignment
-      If(condition, Literal(null, structAssignment.dataType), structAssignment)
-    }
-  }
-
   /**
    * Checks whether assignments are aligned and compatible with table columns.
    *

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -6615,10 +6615,11 @@ object SQLConf {
     buildConf("spark.sql.merge.nested.type.coercion.enabled")
       .internal()
       .doc("If enabled, allow MERGE INTO to coerce source nested types if they have less" +
-        "nested fields than the target table's nested types.")
+        "nested fields than the target table's nested types. This is experimental and" +
+        "the semantics may change.")
       .version("4.1.0")
       .booleanConf
-      .createWithDefault(true)
+      .createWithDefault(false)
 
   /**
    * Holds information about keys that have been deprecated.