[Spark] Refactor deduplication in DeltaMergeBuilder to utilize DeduplicateRelations Rule. (delta-io#2771)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->

Currently, `DeltaMergeBuilder` performs a custom relation deduplication
logic reminiscent to the DeduplicateRelations analyzer rule. This
duplicates logic and might cause issues from unforeseen interaction with
the rule during the analysis of the MERGE plan. This PR refactors the
deduplication in DeltaMergeBuilder to utilise the `DeduplicateRelations`
rule. The new flow is as follows:

1. Un-resolve the ambiguous pre-resolved references.
2. Invoke DeduplicateRelations to do the deduplication work with
FakeLogicalPlan.
3. Use the deduplicated source and target plan in the final MERGE
command.


## How was this patch tested?
Existing tests. `MergeIntoScalaSuite` covers self merges with the Scala
API.

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->

## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->
No.

Author: andreaschat-db

spark/src/main/scala/io/delta/tables/DeltaMergeBuilder.scala

@@ -337,7 +337,9 @@ class DeltaMergeBuilder private(
 
   private def mergePlan: DeltaMergeInto = {
     var targetPlan = targetTable.toDF.queryExecution.analyzed
-    val sourcePlan = source.queryExecution.analyzed
+    var sourcePlan = source.queryExecution.analyzed
+    var condition = onCondition.expr
+    var clauses = whenClauses
 
     // If source and target have duplicate, pre-resolved references (can happen with self-merge),
     // then rewrite the references in target with new exprId to avoid ambiguity.
@@ -346,17 +348,25 @@ class DeltaMergeBuilder private(
     // optional SubqueryAlias.
     val duplicateResolvedRefs = targetPlan.outputSet.intersect(sourcePlan.outputSet)
     if (duplicateResolvedRefs.nonEmpty) {
-      val refReplacementMap = duplicateResolvedRefs.toSeq.flatMap {
-        case a: AttributeReference =>
-          Some(a.exprId -> a.withExprId(NamedExpression.newExprId))
-        case _ => None
-      }.toMap
-      targetPlan = targetPlan.transformAllExpressions {
-        case a: AttributeReference if refReplacementMap.contains(a.exprId) =>
-          refReplacementMap(a.exprId)
-      }
-      logInfo("Rewritten duplicate refs between target and source plans: "
-        + refReplacementMap.toSeq.mkString(", "))
+      val exprs = (condition +: clauses).map(_.transform {
+        // If any expression contain duplicate, pre-resolved references, we can't simply
+        // replace the references in the same way as the target because we don't know
+        // whether the user intended to refer to the source or the target columns. Instead,
+        // we unresolve them (only the duplicate refs) and let the analysis resolve the ambiguity
+        // and throw the usual error messages when needed.
+        case a: AttributeReference if duplicateResolvedRefs.contains(a) =>
+          UnresolvedAttribute(a.qualifier :+ a.name)
+      })
+      // Deduplicate the attribute IDs in the target and source plans, and all the MERGE
+      // expressions (condition and MERGE clauses), so that we can avoid duplicated attribute ID
+      // when building the MERGE command later.
+      val fakePlan = AnalysisHelper.FakeLogicalPlan(exprs, Seq(sourcePlan, targetPlan))
+      val newPlan = org.apache.spark.sql.catalyst.analysis.DeduplicateRelations(fakePlan)
+        .asInstanceOf[AnalysisHelper.FakeLogicalPlan]
+      sourcePlan = newPlan.children(0)
+      targetPlan = newPlan.children(1)
+      condition = newPlan.exprs.head
+      clauses = newPlan.exprs.takeRight(clauses.size).asInstanceOf[Seq[DeltaMergeIntoClause]]
     }
 
     // Note: The Scala API cannot generate MergeIntoTable just like the SQL parser because
@@ -367,24 +377,9 @@ class DeltaMergeBuilder private(
     // (possible in Scala API, but syntactically not possible in SQL). This issue is tracked
     // by https://issues.apache.org/jira/browse/SPARK-34962.
     val merge = DeltaMergeInto(
-      targetPlan,
-      sourcePlan,
-      onCondition.expr,
-      whenClauses,
-      withSchemaEvolution = schemaEvolutionEnabled)
-    val finalMerge = if (duplicateResolvedRefs.nonEmpty) {
-      // If any expression contain duplicate, pre-resolved references, we can't simply
-      // replace the references in the same way as the target because we don't know
-      // whether the user intended to refer to the source or the target columns. Instead,
-      // we unresolve them (only the duplicate refs) and let the analysis resolve the ambiguity
-      // and throw the usual error messages when needed.
-      merge.transformExpressions {
-        case a: AttributeReference if duplicateResolvedRefs.contains(a) =>
-          UnresolvedAttribute(a.qualifier :+ a.name)
-      }
-    } else merge
-    logDebug("Generated merged plan:\n" + finalMerge)
-    finalMerge
+      targetPlan, sourcePlan, condition, clauses, withSchemaEvolution = schemaEvolutionEnabled)
+    logDebug("Generated merged plan:\n" + merge)
+    merge
   }
 }