Skip to content

[SPARK-53482][SQL] MERGE INTO support for when source has less nested field than target #52347

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
szehon-ho wants to merge 7 commits into from
Closed

[SPARK-53482][SQL] MERGE INTO support for when source has less nested field than target #52347

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
165ae51
[SPARK-53482][SQL] MERGE INTO support for when source has less nested…
szehon-ho Sep 14, 2025
500b011
Add test
szehon-ho Sep 15, 2025
4ed2f04
rebase and fix tests
szehon-ho Oct 27, 2025
62cce3e
Fix more tests
szehon-ho Oct 27, 2025
9a943f3
Fix more tests
szehon-ho Oct 28, 2025
bebdac5
Review comments and add more tests
szehon-ho Oct 29, 2025
01c536e
Lint
szehon-ho Oct 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 22 additions & 9 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.analysis
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.{Attribute, CreateNamedStruct, Expression, GetStructField, Literal}
import org.apache.spark.sql.catalyst.plans.logical.Assignment
import org.apache.spark.sql.catalyst.types.DataTypeUtils
Expand Down Expand Up @@ -49,11 +50,14 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
*
* @param attrs table attributes
* @param assignments assignments to align
* @param coerceNestedTypes whether to coerce nested types to match the target type
* for complex types
* @return aligned update assignments that match table attributes
*/
def alignUpdateAssignments(
attrs: Seq[Attribute],
assignments: Seq[Assignment]): Seq[Assignment] = {
assignments: Seq[Assignment],
coerceNestedTypes: Boolean): Seq[Assignment] = {

val errors = new mutable.ArrayBuffer[String]()

Expand All @@ -63,7 +67,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
colExpr = attr,
assignments,
addError = err => errors += err,
colPath = Seq(attr.name))
colPath = Seq(attr.name),
coerceNestedTypes)
}

if (errors.nonEmpty) {
Expand All @@ -84,11 +89,14 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
*
* @param attrs table attributes
* @param assignments insert assignments to align
* @param coerceNestedTypes whether to coerce nested types to match the target type
* for complex types
* @return aligned insert assignments that match table attributes
*/
def alignInsertAssignments(
attrs: Seq[Attribute],
assignments: Seq[Assignment]): Seq[Assignment] = {
assignments: Seq[Assignment],
coerceNestedTypes: Boolean = false): Seq[Assignment] = {

val errors = new mutable.ArrayBuffer[String]()

Expand Down Expand Up @@ -120,8 +128,9 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
val colPath = Seq(attr.name)
val actualAttr = restoreActualType(attr)
val value = matchingAssignments.head.value
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate(
"", value, actualAttr, conf, err => errors += err, colPath)
"", value, actualAttr, conf, err => errors += err, colPath, coerceMode)
}
Assignment(attr, resolvedValue)
}
Expand All @@ -142,7 +151,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
colExpr: Expression,
assignments: Seq[Assignment],
addError: String => Unit,
colPath: Seq[String]): Expression = {
colPath: Seq[String],
coerceNestedTypes: Boolean = false): Expression = {

val (exactAssignments, otherAssignments) = assignments.partition { assignment =>
assignment.key.semanticEquals(colExpr)
Expand All @@ -165,9 +175,10 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
TableOutputResolver.checkNullability(colExpr, col, conf, colPath)
} else if (exactAssignments.nonEmpty) {
val value = exactAssignments.head.value
TableOutputResolver.resolveUpdate("", value, col, conf, addError, colPath)
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate("", value, col, conf, addError, colPath, coerceMode)
} else {
applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath)
applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes)
}
}

Expand All @@ -176,7 +187,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
colExpr: Expression,
assignments: Seq[Assignment],
addError: String => Unit,
colPath: Seq[String]): Expression = {
colPath: Seq[String],
coerceNestedTyptes: Boolean): Expression = {

col.dataType match {
case structType: StructType =>
Expand All @@ -185,7 +197,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
GetStructField(colExpr, ordinal, Some(field.name))
}
val updatedFieldExprs = fieldAttrs.zip(fieldExprs).map { case (fieldAttr, fieldExpr) =>
applyAssignments(fieldAttr, fieldExpr, assignments, addError, colPath :+ fieldAttr.name)
applyAssignments(fieldAttr, fieldExpr, assignments, addError, colPath :+ fieldAttr.name,
coerceNestedTyptes)
}
toNamedStruct(structType, updatedFieldExprs)

Expand Down
22 changes: 15 additions & 7 deletions ...main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ 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.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy

/**
Expand All @@ -42,7 +43,8 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
case u: UpdateTable if !u.skipSchemaResolution && u.resolved && u.rewritable && !u.aligned =>
validateStoreAssignmentPolicy()
val newTable = cleanAttrMetadata(u.table)
val newAssignments = AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments)
val newAssignments = AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments,
coerceNestedTypes = false)
Comment thread
cloud-fan marked this conversation as resolved.
u.copy(table = newTable, assignments = newAssignments)

case u: UpdateTable if !u.skipSchemaResolution && u.resolved && !u.aligned =>
Expand All @@ -51,11 +53,14 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {
case m: MergeIntoTable if !m.skipSchemaResolution && m.resolved && m.rewritable && !m.aligned &&
!m.needSchemaEvolution =>
validateStoreAssignmentPolicy()
val coerceNestedTypes = SQLConf.get.coerceMergeNestedTypes
m.copy(
targetTable = cleanAttrMetadata(m.targetTable),
matchedActions = alignActions(m.targetTable.output, m.matchedActions),
notMatchedActions = alignActions(m.targetTable.output, m.notMatchedActions),
notMatchedBySourceActions = alignActions(m.targetTable.output, m.notMatchedBySourceActions))
matchedActions = alignActions(m.targetTable.output, m.matchedActions, coerceNestedTypes),
notMatchedActions = alignActions(m.targetTable.output, m.notMatchedActions,
coerceNestedTypes),
notMatchedBySourceActions = alignActions(m.targetTable.output, m.notMatchedBySourceActions,
coerceNestedTypes))

case m: MergeIntoTable if !m.skipSchemaResolution && m.resolved && !m.aligned
&& !m.needSchemaEvolution =>
Expand Down Expand Up @@ -109,14 +114,17 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] {

private def alignActions(
attrs: Seq[Attribute],
actions: Seq[MergeAction]): Seq[MergeAction] = {
actions: Seq[MergeAction],
coerceNestedTypes: Boolean): Seq[MergeAction] = {
actions.map {
case u @ UpdateAction(_, assignments) =>
u.copy(assignments = AssignmentUtils.alignUpdateAssignments(attrs, assignments))
u.copy(assignments = AssignmentUtils.alignUpdateAssignments(attrs, assignments,
coerceNestedTypes))
case d: DeleteAction =>
d
case i @ InsertAction(_, assignments) =>
i.copy(assignments = AssignmentUtils.alignInsertAssignments(attrs, assignments))
i.copy(assignments = AssignmentUtils.alignInsertAssignments(attrs, assignments,
coerceNestedTypes))
case other =>
throw new AnalysisException(
errorClass = "_LEGACY_ERROR_TEMP_3052",
Expand Down
69 changes: 48 additions & 21 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import scala.collection.mutable
import org.apache.spark.internal.Logging
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.analysis.TableOutputResolver.DefaultValueFillMode.{FILL, NONE, RECURSE}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
Expand All @@ -38,6 +39,17 @@ import org.apache.spark.sql.types.{ArrayType, DataType, DecimalType, IntegralTyp

object TableOutputResolver extends SQLConfHelper with Logging {

/**
* Modes for filling in default or null values for missing columns.
* If FILL, fill missing top-level columns with their default values.
* If RECURSE, fill missing top-level columns and also recurse into nested struct
* fields to fill null.
* If NONE, do not fill any missing columns.
*/
object DefaultValueFillMode extends Enumeration {

@szehon-ho szehon-ho Oct 28, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunately, we need to distinguish between filling top level defaults and nested defaults within structs.

  1. Normal V2Writes does not expect nested source nested type coercion. For example, write into a target dataframe from a source dataframe that has a struct column with less fields does not work today.

This goes through Analyzer.ResolveOutputRelation, which calls resolveOutputColumns() => reorderColumnsByName() => resolveStruct/Array/MapType()

  1. RowLevelOperations , in particular Merge Into, want recursive support to coerce source struct columns with less fields than target column.

This goes through resolveUpdate => resolveStruct/Array/MapType()

So hence, we need a three-way enum here to distinguish the three cases (none, first-level, recurse).

val FILL, RECURSE, NONE = Value
}

def resolveVariableOutputColumns(
expected: Seq[VariableReference],
query: LogicalPlan,
Expand Down Expand Up @@ -90,15 +102,17 @@ object TableOutputResolver extends SQLConfHelper with Logging {
val errors = new mutable.ArrayBuffer[String]()
val resolved: Seq[NamedExpression] = if (byName) {
// If a top-level column does not have a corresponding value in the input query, fill with
// the column's default value. We need to pass `fillDefaultValue` as true here, if the
// the column's default value. We need to pass `fillDefaultValue` as FILL here, if the
// `supportColDefaultValue` parameter is also true.
val defaultValueFillMode = if (supportColDefaultValue) FILL else NONE
reorderColumnsByName(
tableName,
query.output,
expected,
conf,
errors += _,
fillDefaultValue = supportColDefaultValue)
Nil,
defaultValueFillMode)
} else {
if (expected.size > query.output.size) {
throw QueryCompilationErrors.cannotWriteNotEnoughColumnsToTableError(
Expand All @@ -125,8 +139,10 @@ object TableOutputResolver extends SQLConfHelper with Logging {
col: Attribute,
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String]): Expression = {
colPath: Seq[String],
defaultValueFillMode: DefaultValueFillMode.Value): Expression = {

val fillChildDefaultValue = defaultValueFillMode == RECURSE
(value.dataType, col.dataType) match {
// no need to reorder inner fields or cast if types are already compatible
case (valueType, colType) if DataType.equalsIgnoreCompatibleNullability(valueType, colType) =>
Expand All @@ -141,17 +157,17 @@ object TableOutputResolver extends SQLConfHelper with Logging {
case (valueType: StructType, colType: StructType) =>
val resolvedValue = resolveStructType(
tableName, value, valueType, col, colType,
byName = true, conf, addError, colPath)
byName = true, conf, addError, colPath, fillChildDefaultValue)
resolvedValue.getOrElse(value)
case (valueType: ArrayType, colType: ArrayType) =>
val resolvedValue = resolveArrayType(
tableName, value, valueType, col, colType,
byName = true, conf, addError, colPath)
byName = true, conf, addError, colPath, fillChildDefaultValue)
resolvedValue.getOrElse(value)
case (valueType: MapType, colType: MapType) =>
val resolvedValue = resolveMapType(
tableName, value, valueType, col, colType,
byName = true, conf, addError, colPath)
byName = true, conf, addError, colPath, fillChildDefaultValue)
resolvedValue.getOrElse(value)
case _ =>
checkUpdate(tableName, value, col, conf, addError, colPath)
Expand Down Expand Up @@ -288,13 +304,13 @@ object TableOutputResolver extends SQLConfHelper with Logging {
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String] = Nil,
fillDefaultValue: Boolean = false): Seq[NamedExpression] = {
defaultValueFillMode: DefaultValueFillMode.Value): Seq[NamedExpression] = {
val matchedCols = mutable.HashSet.empty[String]
val reordered = expectedCols.flatMap { expectedCol =>
val matched = inputCols.filter(col => conf.resolver(col.name, expectedCol.name))
val newColPath = colPath :+ expectedCol.name
if (matched.isEmpty) {
val defaultExpr = if (fillDefaultValue) {
val defaultExpr = if (Set(FILL, RECURSE).contains(defaultValueFillMode)) {
getDefaultValueExprOrNullLit(expectedCol, conf.useNullsForMissingDefaultColumnValues)
} else {
None
Expand All @@ -315,19 +331,20 @@ object TableOutputResolver extends SQLConfHelper with Logging {
val actualExpectedCol = expectedCol.withDataType {
CharVarcharUtils.getRawType(expectedCol.metadata).getOrElse(expectedCol.dataType)
}
val childFillDefaultValue = defaultValueFillMode == RECURSE
(matchedCol.dataType, actualExpectedCol.dataType) match {
case (matchedType: StructType, expectedType: StructType) =>
resolveStructType(
tableName, matchedCol, matchedType, actualExpectedCol, expectedType,
byName = true, conf, addError, newColPath)
byName = true, conf, addError, newColPath, childFillDefaultValue)
case (matchedType: ArrayType, expectedType: ArrayType) =>
resolveArrayType(
tableName, matchedCol, matchedType, actualExpectedCol, expectedType,
byName = true, conf, addError, newColPath)
byName = true, conf, addError, newColPath, childFillDefaultValue)
case (matchedType: MapType, expectedType: MapType) =>
resolveMapType(
tableName, matchedCol, matchedType, actualExpectedCol, expectedType,
byName = true, conf, addError, newColPath)
byName = true, conf, addError, newColPath, childFillDefaultValue)
case _ =>
checkField(
tableName, actualExpectedCol, matchedCol, byName = true, conf, addError, newColPath)
Expand Down Expand Up @@ -396,15 +413,15 @@ object TableOutputResolver extends SQLConfHelper with Logging {
case (inputType: StructType, expectedType: StructType) =>
resolveStructType(
tableName, inputCol, inputType, expectedCol, expectedType,
byName = false, conf, addError, newColPath)
byName = false, conf, addError, newColPath, fillDefaultValue = false)
case (inputType: ArrayType, expectedType: ArrayType) =>
resolveArrayType(
tableName, inputCol, inputType, expectedCol, expectedType,
byName = false, conf, addError, newColPath)
byName = false, conf, addError, newColPath, fillDefaultValue = false)
case (inputType: MapType, expectedType: MapType) =>
resolveMapType(
tableName, inputCol, inputType, expectedCol, expectedType,
byName = false, conf, addError, newColPath)
byName = false, conf, addError, newColPath, fillDefaultValue = false)
case _ =>
checkField(tableName, expectedCol, inputCol, byName = false, conf, addError, newColPath)
}
Expand Down Expand Up @@ -439,13 +456,16 @@ object TableOutputResolver extends SQLConfHelper with Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String]): Option[NamedExpression] = {
colPath: Seq[String],
fillDefaultValue: Boolean): Option[NamedExpression] = {
val nullCheckedInput = checkNullability(input, expected, conf, colPath)
val fields = inputType.zipWithIndex.map { case (f, i) =>
Alias(GetStructField(nullCheckedInput, i, Some(f.name)), f.name)()
}
val defaultValueMode = if (fillDefaultValue) RECURSE else NONE
val resolved = if (byName) {
reorderColumnsByName(tableName, fields, toAttributes(expectedType), conf, addError, colPath)
reorderColumnsByName(tableName, fields, toAttributes(expectedType), conf, addError, colPath,
defaultValueMode)
} else {
resolveColumnsByPosition(
tableName, fields, toAttributes(expectedType), conf, addError, colPath)
Expand All @@ -472,13 +492,16 @@ object TableOutputResolver extends SQLConfHelper with Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String]): Option[NamedExpression] = {
colPath: Seq[String],
fillDefaultValue: Boolean): Option[NamedExpression] = {
val nullCheckedInput = checkNullability(input, expected, conf, colPath)
val param = NamedLambdaVariable("element", inputType.elementType, inputType.containsNull)
val fakeAttr =
AttributeReference("element", expectedType.elementType, expectedType.containsNull)()
val res = if (byName) {
reorderColumnsByName(tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath)
val defaultValueMode = if (fillDefaultValue) RECURSE else NONE
reorderColumnsByName(tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath,
defaultValueMode)
} else {
resolveColumnsByPosition(tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath)
}
Expand Down Expand Up @@ -506,13 +529,16 @@ object TableOutputResolver extends SQLConfHelper with Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String]): Option[NamedExpression] = {
colPath: Seq[String],
fillDefaultValue: Boolean): Option[NamedExpression] = {
val nullCheckedInput = checkNullability(input, expected, conf, colPath)

val keyParam = NamedLambdaVariable("key", inputType.keyType, nullable = false)
val fakeKeyAttr = AttributeReference("key", expectedType.keyType, nullable = false)()
val defaultValueFillMode = if (fillDefaultValue) RECURSE else NONE
val resKey = if (byName) {
reorderColumnsByName(tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath)
reorderColumnsByName(tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath,
defaultValueFillMode)
} else {
resolveColumnsByPosition(tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath)
}
Expand All @@ -522,7 +548,8 @@ object TableOutputResolver extends SQLConfHelper with Logging {
val fakeValueAttr =
AttributeReference("value", expectedType.valueType, expectedType.valueContainsNull)()
val resValue = if (byName) {
reorderColumnsByName(tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError, colPath)
reorderColumnsByName(tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError, colPath,
defaultValueFillMode)
} else {
resolveColumnsByPosition(
tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError, colPath)
Expand Down
12 changes: 12 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6427,6 +6427,15 @@ object SQLConf {
.booleanConf
.createWithDefault(true)

val MERGE_INTO_SOURCE_NESTED_TYPE_COERCION_ENABLED =
buildConf("spark.sql.merge.source.nested.type.coercion.enabled")

@aokolnychyi aokolnychyi Nov 27, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't seem to align with the naming pattern we use elsewhere.

@dongjoon-hyun dongjoon-hyun Dec 11, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 for @aokolnychyi 's comment. Let me make a PR for this.

.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.")
.version("4.1.0")
.booleanConf
.createWithDefault(true)

/**
* Holds information about keys that have been deprecated.
*
Expand Down Expand Up @@ -7555,6 +7564,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def legacyXMLParserEnabled: Boolean =
getConf(SQLConf.LEGACY_XML_PARSER_ENABLED)

def coerceMergeNestedTypes: Boolean =
getConf(SQLConf.MERGE_INTO_SOURCE_NESTED_TYPE_COERCION_ENABLED)

/** ********************** SQLConf functionality methods ************ */

/** Set Spark SQL configuration properties. */
Expand Down
Loading