[SPARK-59342][SQL] Allow configuring schema alignment behavior for DSv2 writes

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Schema-alignment configuration for writes to a {@link Table}. This allows connectors to
+ * configure casting behavior and handling of schema mismatches during writes.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface SchemaAlignmentConfig {
+
+  /** The strict data source v2 configuration, returned by {@link Table} by default. */
+  SchemaAlignmentConfig DEFAULT = new SchemaAlignmentConfig() {};
+
+  /**
+   * Whether {@code spark.sql.storeAssignmentPolicy=LEGACY} is allowed for writes and row-level
+   * operations targeting this table. Data source v2 rejects LEGACY by default; a table can decide
+   * to opt-out from this restriction.
+   */
+  default boolean allowLegacyStoreAssignmentPolicy() {
+    return false;
+  }
+
+  /**
+   * Whether the {@code ANSI} store-assignment cast check is deferred from analysis to runtime under
+   * {@code spark.sql.storeAssignmentPolicy=ANSI}. When {@code true}, the analyzer skips the
+   * store-assignment compatibility check and inserts an ANSI cast, so malformed values or
+   * overflows surface at execution time.
+   */
+  default boolean deferCastValidationToRuntime() {
+    return false;
+  }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java

@@ -112,4 +112,14 @@ default Map<String, String> properties() {
    * the version that corresponds to the current state of this table instance.
    */
   default String version() { return null; }
+
+  /**
+   * Returns the schema-alignment configuration for writes to this table. This allows connectors to
+   * configure casting behavior and handling of schema mismatches during writes.
+   * It is recommended to use the DEFAULT configuration to provide a unified behavior across data
+   * sources, but some connectors may require deviating from the default behavior.
+   */
+  default SchemaAlignmentConfig schemaAlignmentConfig() {
+    return SchemaAlignmentConfig.DEFAULT;
+  }
 }

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

@@ -3945,7 +3945,10 @@ class Analyzer(
       case v2Write: V2WriteCommand
           if v2Write.table.resolved && v2Write.query.resolved && !v2Write.outputResolved &&
             v2Write.pendingSchemaChanges.isEmpty =>
-        validateStoreAssignmentPolicy()
+        val schemaAlignment = v2Write.table.collectFirst {
+          case r: DataSourceV2Relation => r.table.schemaAlignmentConfig()
+        }.getOrElse(SchemaAlignmentConfig.DEFAULT)
+        validateStoreAssignmentPolicy(schemaAlignment)
         TableOutputResolver.suitableForByNameCheck(v2Write.isByName,
           expected = v2Write.table.output, queryOutput = v2Write.query.output)
         // With schema evolution + coercion flag, missing top-level columns AND missing nested
@@ -3965,7 +3968,8 @@ class Analyzer(
         val (projection, autoFilledGenCols) =
           TableOutputResolver.resolveOutputColumnsWithGeneratedInfo(
             v2Write.table.name, expected, v2Write.query, v2Write.isByName, conf,
-            defaultValueFillMode)
+            defaultValueFillMode,
+            deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime())
         if (projection != v2Write.query) {
           val cleanedTable = v2Write.table match {
             case r: DataSourceV2Relation =>
@@ -3981,9 +3985,10 @@ class Analyzer(
     }
   }
 
-  private def validateStoreAssignmentPolicy(): Unit = {
-    // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2.
-    if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) {
+  private def validateStoreAssignmentPolicy(schemaAlignment: SchemaAlignmentConfig): Unit = {
+    // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2 by default.
+    if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY &&
+        !schemaAlignment.allowLegacyStoreAssignmentPolicy()) {
       throw QueryCompilationErrors.legacyStoreAssignmentPolicyError()
     }
   }

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

@@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getDefaultValueExprOrNullLit
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.connector.catalog.SchemaAlignmentConfig
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataType, StructType}
@@ -63,7 +64,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       attrs: Seq[Attribute],
       assignments: Seq[Assignment],
       fromStar: Boolean,
-      coerceNestedTypes: Boolean): Seq[Assignment] = {
+      coerceNestedTypes: Boolean,
+      schemaAlignment: SchemaAlignmentConfig = SchemaAlignmentConfig.DEFAULT): Seq[Assignment] = {
 
     val errors = new mutable.ArrayBuffer[String]()
 
@@ -75,7 +77,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
         addError = err => errors += err,
         colPath = Seq(attr.name),
         coerceNestedTypes,
-        fromStar)
+        fromStar,
+        schemaAlignment = schemaAlignment)
     }
 
     if (errors.nonEmpty) {
@@ -103,7 +106,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
   def alignInsertAssignments(
       attrs: Seq[Attribute],
       assignments: Seq[Assignment],
-      coerceNestedTypes: Boolean = false): Seq[Assignment] = {
+      coerceNestedTypes: Boolean = false,
+      schemaAlignment: SchemaAlignmentConfig = SchemaAlignmentConfig.DEFAULT): Seq[Assignment] = {
 
     val errors = new mutable.ArrayBuffer[String]()
 
@@ -137,7 +141,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
         val value = matchingAssignments.head.value
         val coerceMode = if (coerceNestedTypes) RECURSE else NONE
         TableOutputResolver.resolveUpdate(
-          "", value, actualAttr, conf, err => errors += err, colPath, coerceMode)
+          "", value, actualAttr, conf, err => errors += err, colPath, coerceMode,
+          deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime())
       }
       Assignment(attr, resolvedValue)
     }
@@ -160,7 +165,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       addError: String => Unit,
       colPath: Seq[String],
       coerceNestedTypes: Boolean = false,
-      updateStar: Boolean = false): Expression = {
+      updateStar: Boolean = false,
+      schemaAlignment: SchemaAlignmentConfig): Expression = {
 
     val (exactAssignments, otherAssignments) = assignments.partition { assignment =>
       assignment.key.semanticEquals(colExpr)
@@ -188,21 +194,24 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
           case _: StructType =>
             // Expand assignments to leaf fields (fixNullExpansion is applied inside)
             applyNestedFieldAssignments(col, colExpr, value, addError, colPath,
-              coerceNestedTypes)
+              coerceNestedTypes, schemaAlignment = schemaAlignment)
           case _ =>
             // For non-struct types, resolve directly
             val coerceMode = if (coerceNestedTypes) RECURSE else NONE
             TableOutputResolver.resolveUpdate("", value, col, conf, addError, colPath,
-              coerceMode)
+              coerceMode,
+              deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime())
         }
       } else {
         val value = exactAssignments.head.value
         val coerceMode = if (coerceNestedTypes) RECURSE else NONE
         TableOutputResolver.resolveUpdate("", value, col, conf, addError,
-          colPath, coerceMode)
+          colPath, coerceMode,
+          deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime())
       }
     } else {
-      applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes)
+      applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes,
+        schemaAlignment = schemaAlignment)
     }
   }
 
@@ -212,7 +221,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       assignments: Seq[Assignment],
       addError: String => Unit,
       colPath: Seq[String],
-      coerceNestedTypes: Boolean): Expression = {
+      coerceNestedTypes: Boolean,
+      schemaAlignment: SchemaAlignmentConfig): Expression = {
 
     col.dataType match {
       case structType: StructType =>
@@ -222,7 +232,7 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
         }
         val updatedFieldExprs = fieldAttrs.zip(fieldExprs).map { case (fieldAttr, fieldExpr) =>
           applyAssignments(fieldAttr, fieldExpr, assignments, addError, colPath :+ fieldAttr.name,
-            coerceNestedTypes)
+            coerceNestedTypes, schemaAlignment = schemaAlignment)
         }
         toNamedStruct(structType, updatedFieldExprs)
 
@@ -240,7 +250,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       value: Expression,
       addError: String => Unit,
       colPath: Seq[String],
-      coerceNestedTypes: Boolean): Expression = {
+      coerceNestedTypes: Boolean,
+      schemaAlignment: SchemaAlignmentConfig): Expression = {
 
     col.dataType match {
       case structType: StructType =>
@@ -273,12 +284,15 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
             case _: StructType =>
               // Field is a struct, recurse
               applyNestedFieldAssignments(fieldAttr, targetFieldExpr,
-                sourceFieldValue, addError, fieldPath, coerceNestedTypes)
+                sourceFieldValue, addError, fieldPath, coerceNestedTypes,
+                schemaAlignment = schemaAlignment)
             case _ =>
               // Field is not a struct, resolve with TableOutputResolver
               val coerceMode = if (coerceNestedTypes) RECURSE else NONE
-              TableOutputResolver.resolveUpdate("", sourceFieldValue, fieldAttr, conf, addError,
-                fieldPath, coerceMode)
+              TableOutputResolver.resolveUpdate(
+                "", sourceFieldValue, fieldAttr, conf, addError,
+                fieldPath, coerceMode,
+                deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime())
           }
         }
         val namedStruct = toNamedStruct(structType, updatedFieldExprs)

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

@@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Assignment, DeleteAction, In
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.connector.catalog.SchemaAlignmentConfig
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
 import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
@@ -40,32 +41,37 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
   override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsWithPruning(
     _.containsPattern(COMMAND), ruleId) {
     case u: UpdateTable if !u.skipSchemaResolution && u.resolved && u.rewritable && !u.aligned =>
-      validateStoreAssignmentPolicy()
+      val schemaAlignment = schemaAlignmentConfig(u.table)
+      validateStoreAssignmentPolicy(schemaAlignment)
       val newTable = cleanAttrMetadata(u.table)
       val newAssignments = AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments,
-        fromStar = false, coerceNestedTypes = false)
+        fromStar = false, coerceNestedTypes = false, schemaAlignment = schemaAlignment)
       u.copy(table = newTable, assignments = newAssignments)
 
     case u: UpdateTable if !u.skipSchemaResolution && u.resolved && !u.aligned =>
       resolveAssignments(u)
 
     case m: MergeIntoTable if m.rewritable && shouldAlignAssignments(m) && containsFinalSchema(m) =>
-      validateStoreAssignmentPolicy()
+      val schemaAlignment = schemaAlignmentConfig(m.targetTable)
+      validateStoreAssignmentPolicy(schemaAlignment)
       val coerceNestedTypes = conf.coerceMergeNestedTypes && m.withSchemaEvolution
       m.copy(
         targetTable = cleanAttrMetadata(m.targetTable),
         matchedActions = alignActions(
           m.targetTable.output,
           m.matchedActions,
-          coerceNestedTypes),
+          coerceNestedTypes,
+          schemaAlignment),
         notMatchedActions = alignActions(
           m.targetTable.output,
           m.notMatchedActions,
-          coerceNestedTypes),
+          coerceNestedTypes,
+          schemaAlignment),
         notMatchedBySourceActions = alignActions(
           m.targetTable.output,
           m.notMatchedBySourceActions,
-          coerceNestedTypes))
+          coerceNestedTypes,
+          schemaAlignment))
 
     case m: MergeIntoTable if shouldAlignAssignments(m) && containsFinalSchema(m) =>
       resolveAssignments(m)
@@ -79,9 +85,16 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
     !m.schemaEvolutionEnabled || (m.schemaEvolutionReady && m.pendingSchemaChanges.isEmpty)
   }
 
-  private def validateStoreAssignmentPolicy(): Unit = {
-    // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2
-    if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) {
+  private def schemaAlignmentConfig(target: LogicalPlan): SchemaAlignmentConfig =
+    target.collectFirst {
+      case relation: DataSourceV2Relation => relation.table.schemaAlignmentConfig()
+    }.getOrElse(SchemaAlignmentConfig.DEFAULT)
+
+  private def validateStoreAssignmentPolicy(schemaAlignment: SchemaAlignmentConfig): Unit = {
+    // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2, unless the
+    // target table opts into it via its SchemaAlignmentConfig.
+    if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY &&
+        !schemaAlignment.allowLegacyStoreAssignmentPolicy()) {
       throw QueryCompilationErrors.legacyStoreAssignmentPolicyError()
     }
   }
@@ -127,16 +140,17 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
   private def alignActions(
       attrs: Seq[Attribute],
       actions: Seq[MergeAction],
-      coerceNestedTypes: Boolean): Seq[MergeAction] = {
+      coerceNestedTypes: Boolean,
+      schemaAlignment: SchemaAlignmentConfig): Seq[MergeAction] = {
     actions.map {
       case u @ UpdateAction(_, assignments, fromStar) =>
         u.copy(assignments = AssignmentUtils.alignUpdateAssignments(attrs, assignments,
-          fromStar, coerceNestedTypes))
+          fromStar, coerceNestedTypes, schemaAlignment = schemaAlignment))
       case d: DeleteAction =>
         d
       case i @ InsertAction(_, assignments) =>
         i.copy(assignments = AssignmentUtils.alignInsertAssignments(attrs, assignments,
-          coerceNestedTypes))
+          coerceNestedTypes, schemaAlignment = schemaAlignment))
       case other =>
         throw new AnalysisException(
           errorClass = "_LEGACY_ERROR_TEMP_3052",