[SPARK-56125][SQL] Simplify schema calculation for Merge Into Schema Evolution

[szehon-ho]

What changes were proposed in this pull request?

Simplifies the schema evolution logic for MERGE INTO in two ways:

1. Replace pre-filtering source schema with post-filtering schema changes: Instead of building a pruned source schema (sourceSchemaForSchemaEvolution) and then diffing it against the target, we now compute all schema diffs against the full source schema and filter the resulting TableChange objects to only those whose field paths fall under identity assignment paths. This eliminates the recursive filterSchema function, the isEqual helper (which was a duplicate of isSameColumnAssignment), and the isPrefix helper.

2. Replace originalTarget/originalSource threading with onIncompatible callback in ResolveSchemaEvolution.computeSchemaChanges: The original target and source schemas were passed through every recursive call (7+ call sites across 3 methods) but only used once in the error case. A by-name error parameter replaces both, reducing parameter count and improving readability.

Also removes dead code: the containsStarAction check in sourceSchemaForSchemaEvolution was unreachable because actionsSchemaEvolutionReady returns false for star actions (via case _ => false), preventing pendingSchemaChanges from calling sourceSchemaForSchemaEvolution until ResolveReferences has expanded all star actions.

Why are the changes needed?

Code simplification and dead code removal. The existing logic was harder to follow due to the recursive schema pruning approach and redundant parameter threading.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests (the logic is semantically equivalent).

[aokolnychyi]

This was the most complicated method that did many recursive calls, so I agree we should avoid it.

The idea that I had was compatible with changes that Johan did and could look like this:

  private def computeSchemaChanges(merge: MergeIntoTable): Seq[TableChange] = {
    val actions = merge.matchedActions ++ merge.notMatchedActions

    val assignments = actions.flatMap {
      case a: UpdateAction => a.assignments
      case a: InsertAction => a.assignments
      case _ => Seq.empty
    }

    val changes = new mutable.HashSet[TableChange]()

    assignments.foreach {
      case a if isFieldAdditionCandidate(a, merge) =>
        val fieldPath = extractFieldPath(a.key)
        changes += TableChange.addColumn(fieldPath.toArray, a.value.dataType)
      case a if a.resolved && a.key.dataType != a.value.dataType =>
        changes ++= ResolveSchemaEvolution.computeSchemaChanges(
          a.key.dataType,
          a.value.dataType,
          merge.targetTable.schema,
          merge.sourceTable.schema,
          fieldPath = extractFieldPath(a.key),
          isByName = true)
      case _ =>
        // OK
    }

    changes.toSeq
  }

  private def extractFieldPath(expr: Expression): Seq[String] = {
    expr match {
      case UnresolvedAttribute(nameParts) => nameParts
      case a: AttributeReference => Seq(a.name)
      case Alias(child, _) => extractFieldPath(child)
      case GetStructField(child, ordinal, nameOpt) =>
        extractFieldPath(child) :+ nameOpt.getOrElse(s"col$ordinal")
      case _ => Seq.empty
    }
  }

  private def areSchemaEvolutionReady(
      assignments: Seq[Assignment],
      merge: MergeIntoTable): Boolean = {
    assignments.forall(assign => assign.resolved || isFieldAdditionCandidate(assign, merge))
  }

  // TODO: clean up and add doc
  private def isFieldAdditionCandidate(
      assignment: Assignment,
      merge: MergeIntoTable): Boolean = {
    val key = assignment.key
    val keyPath = extractFieldPath(key)
    val value = assignment.value
    val valuePath = extractFieldPath(value)
    !key.resolved &&
      value.resolved &&
      keyPath == valuePath &&
      assignment.value.references.subsetOf(merge.sourceTable.outputSet) &&
      merge.targetTable.resolve(keyPath, SQLConf.get.resolver).isEmpty
  }

[aokolnychyi]

With that said, yours may be better. Let me explore it in more detail.

[szehon-ho]

chat offline with @aokolnychyi , we want to check one scenario still behaves the same before and after this refactor. so added the test in a separate pr so it runs before : #55173 and add it here too to verify it runs after.

[aokolnychyi]

I wonder why can't we do it the other way around, though.

• Find all isSameColumn assignments (compares key and value column paths that must match)
• Iterate over assignments
  • If the key is unresolved and target table doesn't have the column yet, add it.
  • If the key and value are resolved but types differ, recursively call ResolveSchemaEvolution?

[szehon-ho]

I put an alternate pr here: #55302 based on @aokolnychyi 's above comment to compare. Its longer but maybe cleaner (it explicitly shows the two paths- missing or existing column)

[johanl-db]

My vote goes to the approach in this PR, the alternative in https://github.com/apache/spark/pull/55302/changes is a lot less intuitive in my opinion.

Collect all schema changes + filter after the fact is a simpler model.

On the other hand, going over all assignments and collecting all schema changes that an assignment triggers (both at its level and fields nested under it) and deduplicating across all assignments is more complex, e.p. due to the split recursion: computeSchemaChanges recurses into nested fields to find schema changes, and assignments are themselves already nested (in a sense) with e.g. both a and a.x both overlapping