[SPARK-16301] [SQL] The analyzer rule for resolving using joins should respect the case sensitivity setting.

## What changes were proposed in this pull request?
The analyzer rule for resolving using joins should respect the case sensitivity setting.

## How was this patch tested?
New tests in ResolveNaturalJoinSuite

Author: Yin Huai <[email protected]>

Closes #13977 from yhuai/SPARK-16301.

(cherry picked from commit 8b5a8b2)
Signed-off-by: Davies Liu <[email protected]>

Author: yhuai

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

@@ -1836,13 +1836,25 @@ class Analyzer(
   }
 
   private def commonNaturalJoinProcessing(
-     left: LogicalPlan,
-     right: LogicalPlan,
-     joinType: JoinType,
-     joinNames: Seq[String],
-     condition: Option[Expression]) = {
-    val leftKeys = joinNames.map(keyName => left.output.find(_.name == keyName).get)
-    val rightKeys = joinNames.map(keyName => right.output.find(_.name == keyName).get)
+      left: LogicalPlan,
+      right: LogicalPlan,
+      joinType: JoinType,
+      joinNames: Seq[String],
+      condition: Option[Expression]) = {
+    val leftKeys = joinNames.map { keyName =>
+      val joinColumn = left.output.find(attr => resolver(attr.name, keyName))
+      assert(
+        joinColumn.isDefined,
+        s"$keyName should exist in ${left.output.map(_.name).mkString(",")}")
+      joinColumn.get
+    }
+    val rightKeys = joinNames.map { keyName =>
+      val joinColumn = right.output.find(attr => resolver(attr.name, keyName))
+      assert(
+        joinColumn.isDefined,
+        s"$keyName should exist in ${right.output.map(_.name).mkString(",")}")
+      joinColumn.get
+    }
     val joinPairs = leftKeys.zip(rightKeys)
 
     val newCondition = (condition ++ joinPairs.map(EqualTo.tupled)).reduceOption(And)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveNaturalJoinSuite.scala

@@ -113,4 +113,34 @@ class ResolveNaturalJoinSuite extends AnalysisTest {
     assert(error.message.contains(
       "using columns ['d] can not be resolved given input columns: [b, a, c]"))
   }
+
+  test("using join with a case sensitive analyzer") {
+    val expected = r1.join(r2, Inner, Some(EqualTo(a, a))).select(a, b, c)
+
+    {
+      val usingPlan = r1.join(r2, UsingJoin(Inner, Seq(UnresolvedAttribute("a"))), None)
+      checkAnalysis(usingPlan, expected, caseSensitive = true)
+    }
+
+    {
+      val usingPlan = r1.join(r2, UsingJoin(Inner, Seq(UnresolvedAttribute("A"))), None)
+      assertAnalysisError(
+        usingPlan,
+        Seq("using columns ['A] can not be resolved given input columns: [b, a, c, a]"))
+    }
+  }
+
+  test("using join with a case insensitive analyzer") {
+    val expected = r1.join(r2, Inner, Some(EqualTo(a, a))).select(a, b, c)
+
+    {
+      val usingPlan = r1.join(r2, UsingJoin(Inner, Seq(UnresolvedAttribute("a"))), None)
+      checkAnalysis(usingPlan, expected, caseSensitive = false)
+    }
+
+    {
+      val usingPlan = r1.join(r2, UsingJoin(Inner, Seq(UnresolvedAttribute("A"))), None)
+      checkAnalysis(usingPlan, expected, caseSensitive = false)
+    }
+  }
 }