[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.