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

sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/ConfigurableSchemaAlignment.java

@@ -0,0 +1,31 @@
+/*
+ * 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.internal.connector;
+
+import org.apache.spark.sql.connector.catalog.Table;
+
+/**
+ * Allows connectors to configure schema alignment and casting behavior for DSv2 batch/row-level
+ * writes to a {@link Table}. It is not consulted for streaming writes, where alignment is
+ * delegated to the connector.
+ */
+public interface ConfigurableSchemaAlignment extends Table {
+
+  /** The schema alignment configuration for writes to this table. */
+  SchemaAlignmentConfig schemaAlignmentConfig();
+}

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

@@ -0,0 +1,50 @@
+/*
+ * 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.internal.connector;
+
+import org.apache.spark.sql.connector.catalog.Table;
+
+/**
+ * Schema alignment configuration for DSv2 batch/row-level writes to a {@link Table}, exposed via
+ * {@link ConfigurableSchemaAlignment}.
+ */
+public interface SchemaAlignmentConfig {
+
+  /** The default data source v2 configuration. */
+  SchemaAlignmentConfig DEFAULT = new SchemaAlignmentConfig() {};
+
+  /**
+   * When the {@code ANSI} store-assignment cast check runs for writes to a table, under
+   * {@code spark.sql.storeAssignmentPolicy=ANSI}. Has no effect under the {@code STRICT} policy.
+   */
+  enum AnsiStoreAssignmentCastCheck {
+    /** (default) Reject incompatible casts at analysis. See {@code Cast.canANSIStoreAssign}. */
+    AT_ANALYSIS,
+
+    /**
+     * Insert an ANSI cast check, so malformed values or overflows fail at execution time instead
+     * of being rejected during analysis.
+     */
+    AT_RUNTIME
+  }
+
+  /** When the {@code ANSI} store-assignment cast check runs. */
+  default AnsiStoreAssignmentCastCheck ansiStoreAssignmentCastCheck() {
+    return AnsiStoreAssignmentCastCheck.AT_ANALYSIS;
+  }
+}

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

@@ -60,10 +60,10 @@ import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
 import org.apache.spark.sql.connector.catalog.procedures.{BoundProcedure, ProcedureParameter, UnboundProcedure}
 import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform}
 import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, ExtractV2Table}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.SQLConf.{PartitionOverwriteMode, StoreAssignmentPolicy}
-import org.apache.spark.sql.internal.connector.V1Function
+import org.apache.spark.sql.internal.connector.{ConfigurableSchemaAlignment, SchemaAlignmentConfig, V1Function}
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 import org.apache.spark.util.ArrayImplicits._
@@ -3973,10 +3973,15 @@ class Analyzer(
           case r: DataSourceV2Relation => GeneratedColumn.attachGenerationExpressions(r)
           case _ => v2Write.table.output
         }
+        val schemaAlignmentConfig = v2Write.table match {
+          case ExtractV2Table(table: ConfigurableSchemaAlignment) =>
+            table.schemaAlignmentConfig()
+          case _ => SchemaAlignmentConfig.DEFAULT
+        }
         val (projection, autoFilledGenCols) =
           TableOutputResolver.resolveOutputColumnsWithGeneratedInfo(
             v2Write.table.name, expected, v2Write.query, v2Write.isByName, conf,
-            defaultValueFillMode)
+            defaultValueFillMode, schemaAlignmentConfig.ansiStoreAssignmentCastCheck())
         if (projection != v2Write.query) {
           val cleanedTable = v2Write.table match {
             case r: DataSourceV2Relation =>

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

@@ -30,6 +30,8 @@ import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getDefaultValueE
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck.AT_ANALYSIS
 import org.apache.spark.sql.types.{DataType, StructType}
 import org.apache.spark.util.ArrayImplicits._
 
@@ -57,13 +59,17 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
    *                 (preserving existing fields).
    * @param coerceNestedTypes whether to coerce nested types to match the target type
    *                         for complex types
+   * @param ansiStoreAssignmentCastCheck when the ANSI store-assignment cast check runs:
+   *                                     AT_ANALYSIS (default) rejects an unsafe cast at analysis
+   *                                     time; AT_RUNTIME rejects invalid values at execution time.
    * @return aligned update assignments that match table attributes
    */
   def alignUpdateAssignments(
       attrs: Seq[Attribute],
       assignments: Seq[Assignment],
       fromStar: Boolean,
-      coerceNestedTypes: Boolean): Seq[Assignment] = {
+      coerceNestedTypes: Boolean,
+      ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck = AT_ANALYSIS): Seq[Assignment] = {
 
     val errors = new mutable.ArrayBuffer[String]()
 
@@ -75,7 +81,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
         addError = err => errors += err,
         colPath = Seq(attr.name),
         coerceNestedTypes,
-        fromStar)
+        fromStar,
+        ansiStoreAssignmentCastCheck)
     }
 
     if (errors.nonEmpty) {
@@ -98,12 +105,16 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
    * @param assignments insert assignments to align
    * @param coerceNestedTypes whether to coerce nested types to match the target type
    *                          for complex types
+   * @param ansiStoreAssignmentCastCheck when the ANSI store-assignment cast check runs:
+   *                                     AT_ANALYSIS (default) rejects an unsafe cast at analysis
+   *                                     time; AT_RUNTIME rejects invalid values at execution time.
    * @return aligned insert assignments that match table attributes
    */
   def alignInsertAssignments(
       attrs: Seq[Attribute],
       assignments: Seq[Assignment],
-      coerceNestedTypes: Boolean = false): Seq[Assignment] = {
+      coerceNestedTypes: Boolean = false,
+      ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck = AT_ANALYSIS): Seq[Assignment] = {
 
     val errors = new mutable.ArrayBuffer[String]()
 
@@ -137,7 +148,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,
+          ansiStoreAssignmentCastCheck)
       }
       Assignment(attr, resolvedValue)
     }
@@ -160,7 +172,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       addError: String => Unit,
       colPath: Seq[String],
       coerceNestedTypes: Boolean = false,
-      updateStar: Boolean = false): Expression = {
+      updateStar: Boolean = false,
+      ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression = {
 
     val (exactAssignments, otherAssignments) = assignments.partition { assignment =>
       assignment.key.semanticEquals(colExpr)
@@ -188,21 +201,22 @@ 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, ansiStoreAssignmentCastCheck)
           case _ =>
             // For non-struct types, resolve directly
             val coerceMode = if (coerceNestedTypes) RECURSE else NONE
             TableOutputResolver.resolveUpdate("", value, col, conf, addError, colPath,
-              coerceMode)
+              coerceMode, ansiStoreAssignmentCastCheck)
         }
       } else {
         val value = exactAssignments.head.value
         val coerceMode = if (coerceNestedTypes) RECURSE else NONE
         TableOutputResolver.resolveUpdate("", value, col, conf, addError,
-          colPath, coerceMode)
+          colPath, coerceMode, ansiStoreAssignmentCastCheck)
       }
     } else {
-      applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes)
+      applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes,
+        ansiStoreAssignmentCastCheck)
     }
   }
 
@@ -212,7 +226,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       assignments: Seq[Assignment],
       addError: String => Unit,
       colPath: Seq[String],
-      coerceNestedTypes: Boolean): Expression = {
+      coerceNestedTypes: Boolean,
+      ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression = {
 
     col.dataType match {
       case structType: StructType =>
@@ -222,7 +237,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, ansiStoreAssignmentCastCheck = ansiStoreAssignmentCastCheck)
         }
         toNamedStruct(structType, updatedFieldExprs)
 
@@ -240,7 +255,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
       value: Expression,
       addError: String => Unit,
       colPath: Seq[String],
-      coerceNestedTypes: Boolean): Expression = {
+      coerceNestedTypes: Boolean,
+      ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression = {
 
     col.dataType match {
       case structType: StructType =>
@@ -273,12 +289,13 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
             case _: StructType =>
               // Field is a struct, recurse
               applyNestedFieldAssignments(fieldAttr, targetFieldExpr,
-                sourceFieldValue, addError, fieldPath, coerceNestedTypes)
+                sourceFieldValue, addError, fieldPath, coerceNestedTypes,
+                ansiStoreAssignmentCastCheck)
             case _ =>
               // Field is not a struct, resolve with TableOutputResolver
               val coerceMode = if (coerceNestedTypes) RECURSE else NONE
               TableOutputResolver.resolveUpdate("", sourceFieldValue, fieldAttr, conf, addError,
-                fieldPath, coerceMode)
+                fieldPath, coerceMode, ansiStoreAssignmentCastCheck)
           }
         }
         val namedStruct = toNamedStruct(structType, updatedFieldExprs)

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

@@ -25,8 +25,10 @@ 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.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, ExtractV2Table}
 import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.internal.connector.ConfigurableSchemaAlignment
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
 
 /**
  * A rule that resolves assignments in row-level commands.
@@ -43,7 +45,8 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
       validateStoreAssignmentPolicy()
       val newTable = cleanAttrMetadata(u.table)
       val newAssignments = AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments,
-        fromStar = false, coerceNestedTypes = false)
+        fromStar = false, coerceNestedTypes = false,
+        ansiStoreAssignmentCastCheck = ansiStoreAssignmentCastCheck(u.table))
       u.copy(table = newTable, assignments = newAssignments)
 
     case u: UpdateTable if !u.skipSchemaResolution && u.resolved && !u.aligned =>
@@ -52,20 +55,24 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
     case m: MergeIntoTable if m.rewritable && shouldAlignAssignments(m) && containsFinalSchema(m) =>
       validateStoreAssignmentPolicy()
       val coerceNestedTypes = conf.coerceMergeNestedTypes && m.withSchemaEvolution
+      val castCheck = ansiStoreAssignmentCastCheck(m.targetTable)
       m.copy(
         targetTable = cleanAttrMetadata(m.targetTable),
         matchedActions = alignActions(
           m.targetTable.output,
           m.matchedActions,
-          coerceNestedTypes),
+          coerceNestedTypes,
+          castCheck),
         notMatchedActions = alignActions(
           m.targetTable.output,
           m.notMatchedActions,
-          coerceNestedTypes),
+          coerceNestedTypes,
+          castCheck),
         notMatchedBySourceActions = alignActions(
           m.targetTable.output,
           m.notMatchedBySourceActions,
-          coerceNestedTypes))
+          coerceNestedTypes,
+          castCheck))
 
     case m: MergeIntoTable if shouldAlignAssignments(m) && containsFinalSchema(m) =>
       resolveAssignments(m)
@@ -79,6 +86,13 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
     !m.schemaEvolutionEnabled || (m.schemaEvolutionReady && m.pendingSchemaChanges.isEmpty)
   }
 
+  private def ansiStoreAssignmentCastCheck(target: LogicalPlan): AnsiStoreAssignmentCastCheck = {
+    target.collectFirst {
+      case ExtractV2Table(table: ConfigurableSchemaAlignment) =>
+        table.schemaAlignmentConfig().ansiStoreAssignmentCastCheck()
+    }.getOrElse(AnsiStoreAssignmentCastCheck.AT_ANALYSIS)
+  }
+
   private def validateStoreAssignmentPolicy(): Unit = {
     // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2
     if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) {
@@ -127,16 +141,17 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
   private def alignActions(
       attrs: Seq[Attribute],
       actions: Seq[MergeAction],
-      coerceNestedTypes: Boolean): Seq[MergeAction] = {
+      coerceNestedTypes: Boolean,
+      ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Seq[MergeAction] = {
     actions.map {
       case u @ UpdateAction(_, assignments, fromStar) =>
         u.copy(assignments = AssignmentUtils.alignUpdateAssignments(attrs, assignments,
-          fromStar, coerceNestedTypes))
+          fromStar, coerceNestedTypes, ansiStoreAssignmentCastCheck))
       case d: DeleteAction =>
         d
       case i @ InsertAction(_, assignments) =>
         i.copy(assignments = AssignmentUtils.alignInsertAssignments(attrs, assignments,
-          coerceNestedTypes))
+          coerceNestedTypes, ansiStoreAssignmentCastCheck))
       case other =>
         throw new AnalysisException(
           errorClass = "_LEGACY_ERROR_TEMP_3052",