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

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

@@ -1836,13 +1836,15 @@ 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 => left.output.find(attr => resolver(attr.name, keyName)).get)
+    val rightKeys =
+      joinNames.map(keyName => right.output.find(attr => resolver(attr.name, keyName)).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)
+    }
+  }
 }