[SPARK-56333][SQL] Use multiset intersection for NATURAL JOIN column matching

### What changes were proposed in this pull request?
Replace `distinct.filter(resolver)` with a multiset intersection (`canonicalizedIntersect`) when computing common columns for NATURAL JOIN. This preserves duplicate column multiplicity (each name appears `min(left count, right count)` times) instead of deduplicating.

### Why are the changes needed?
`distinct` drops duplicate column names, so `NATURAL JOIN` between relations with repeated column names (e.g., c1, c1, c2) silently loses join conditions. The multiset approach matches `Seq.intersect` semantics while still respecting `spark.sql.caseSensitive`.

Now `NATURAL JOIN` behaves in the same way as before #54858 except it now respects caseSensitive conf.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
New golden file tests.

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes #55163 from stefankandic/unionMultiplicity.

Authored-by: Stefan Kandic <stefan.kandic@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: stefankandic

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

@@ -3632,9 +3632,8 @@ class Analyzer(
       case j @ Join(left, right, NaturalJoin(joinType), condition, hint)
           if j.resolvedExceptNatural =>
         // find common column names from both sides
-        val joinNames = left.output.map(_.name).distinct.filter { leftName =>
-          right.output.map(_.name).exists(resolver(leftName, _))
-        }
+        val joinNames = NaturalAndUsingJoinResolution.canonicalizedIntersect(
+            left.output.map(_.name), right.output.map(_.name))
         val project = commonNaturalJoinProcessing(
           left, right, joinType, joinNames, condition, hint)
         j.getTagValue(LogicalPlan.PLAN_ID_TAG)

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

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
+import scala.collection.mutable
+
 import org.apache.spark.sql.catalyst.SQLConfHelper
 import org.apache.spark.sql.catalyst.expressions.{
   Alias,
@@ -82,6 +84,30 @@ object NaturalAndUsingJoinResolution extends DataTypeErrorsBase with SQLConfHelp
     (output, hiddenOutput, newCondition)
   }
 
+  /**
+   * Computes a multiset intersection of two name sequences, respecting case sensitivity.
+   * Preserves multiplicity: each name appears min(left count, right count) times.
+   */
+  def canonicalizedIntersect(
+      leftNames: Seq[String],
+      rightNames: Seq[String]): Seq[String] = {
+    val rightNameCounts = mutable.HashMap[String, Int]()
+    for (name <- rightNames) {
+      val key = conf.canonicalize(name)
+      rightNameCounts(key) = rightNameCounts.getOrElse(key, 0) + 1
+    }
+    leftNames.filter { leftName =>
+      val key = conf.canonicalize(leftName)
+      val count = rightNameCounts.getOrElse(key, 0)
+      if (count > 0) {
+        rightNameCounts(key) = count - 1
+        true
+      } else {
+        false
+      }
+    }
+  }
+
   /**
    * Returns resolved keys for joining based on the output of [[Join]]'s children or throws and
    * error if a key name doesn't exist.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/JoinResolver.scala

@@ -370,7 +370,10 @@ class JoinResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
   private def getJoinNamesForNaturalJoin(
       leftNameScope: NameScope,
       rightNameScope: NameScope): Seq[String] = {
-    leftNameScope.output.map(_.name).intersect(rightNameScope.output.map(_.name))
+    NaturalAndUsingJoinResolution.canonicalizedIntersect(
+      leftNameScope.output.map(_.name),
+      rightNameScope.output.map(_.name)
+    )
   }
 
   /**

sql/core/src/test/resources/sql-tests/analyzer-results/natural-join.sql.out

@@ -692,3 +692,31 @@ Project [ID#x]
       +- SubqueryAlias t2
          +- Project [1 AS id#x]
             +- OneRowRelation
+
+
+-- !query
+SELECT * FROM (SELECT 1 AS c1, 2 AS c1, 3 AS c2) NATURAL JOIN (SELECT 1 AS c1, 2 AS c1, 3 AS c2)
+-- !query analysis
+Project [c1#x, c1#x, c2#x, c1#x, c1#x]
++- Project [c1#x, c1#x, c2#x, c1#x, c1#x]
+   +- Join Inner, (((c1#x = c1#x) AND (c1#x = c1#x)) AND (c2#x = c2#x))
+      :- SubqueryAlias __auto_generated_subquery_name
+      :  +- Project [1 AS c1#x, 2 AS c1#x, 3 AS c2#x]
+      :     +- OneRowRelation
+      +- SubqueryAlias __auto_generated_subquery_name
+         +- Project [1 AS c1#x, 2 AS c1#x, 3 AS c2#x]
+            +- OneRowRelation
+
+
+-- !query
+SELECT * FROM (SELECT 1 AS c1, 2 AS c1, 3 AS c2) NATURAL JOIN (SELECT 1 AS c1, 3 AS c2)
+-- !query analysis
+Project [c1#x, c2#x, c1#x]
++- Project [c1#x, c2#x, c1#x]
+   +- Join Inner, ((c1#x = c1#x) AND (c2#x = c2#x))
+      :- SubqueryAlias __auto_generated_subquery_name
+      :  +- Project [1 AS c1#x, 2 AS c1#x, 3 AS c2#x]
+      :     +- OneRowRelation
+      +- SubqueryAlias __auto_generated_subquery_name
+         +- Project [1 AS c1#x, 3 AS c2#x]
+            +- OneRowRelation

sql/core/src/test/resources/sql-tests/inputs/natural-join.sql

@@ -77,3 +77,7 @@ SELECT * FROM nt1 natural join nt2 join nt3 on nt2.k = nt3.k;
 SELECT nt1.*, nt2.*, nt3.*, nt4.* FROM nt1 natural join nt2 natural join nt3 natural join nt4;
 
 SELECT * FROM (SELECT 1 AS ID) t1 NATURAL JOIN (SELECT 1 AS id) t2;
+
+SELECT * FROM (SELECT 1 AS c1, 2 AS c1, 3 AS c2) NATURAL JOIN (SELECT 1 AS c1, 2 AS c1, 3 AS c2);
+
+SELECT * FROM (SELECT 1 AS c1, 2 AS c1, 3 AS c2) NATURAL JOIN (SELECT 1 AS c1, 3 AS c2);

sql/core/src/test/resources/sql-tests/results/natural-join.sql.out

@@ -348,3 +348,19 @@ SELECT * FROM (SELECT 1 AS ID) t1 NATURAL JOIN (SELECT 1 AS id) t2
 struct<ID:int>
 -- !query output
 1
+
+
+-- !query
+SELECT * FROM (SELECT 1 AS c1, 2 AS c1, 3 AS c2) NATURAL JOIN (SELECT 1 AS c1, 2 AS c1, 3 AS c2)
+-- !query schema
+struct<c1:int,c1:int,c2:int,c1:int,c1:int>
+-- !query output
+1	1	3	2	2
+
+
+-- !query
+SELECT * FROM (SELECT 1 AS c1, 2 AS c1, 3 AS c2) NATURAL JOIN (SELECT 1 AS c1, 3 AS c2)
+-- !query schema
+struct<c1:int,c2:int,c1:int>
+-- !query output
+1	3	2