Fix bug in broadcast hash join & add test to cover it.

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

@@ -58,7 +58,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
         condition: Option[Expression],
         side: BuildSide) = {
       val broadcastHashJoin = execution.BroadcastHashJoin(
-        leftKeys, rightKeys, BuildRight, planLater(left), planLater(right))(sqlContext)
+        leftKeys, rightKeys, side, planLater(left), planLater(right))(sqlContext)
       condition.map(Filter(_, broadcastHashJoin)).getOrElse(broadcastHashJoin) :: Nil
     }
 
@@ -73,7 +73,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
               left,
               right @ PhysicalOperation(_, _, b: BaseRelation))
         if broadcastTables.contains(b.tableName) =>
-            broadcastHashJoin(leftKeys, rightKeys, left, right, condition, BuildRight)
+          broadcastHashJoin(leftKeys, rightKeys, left, right, condition, BuildRight)
 
       case HashFilteredJoin(
               Inner,
@@ -83,7 +83,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
               left @ PhysicalOperation(_, _, b: BaseRelation),
               right)
         if broadcastTables.contains(b.tableName) =>
-            broadcastHashJoin(leftKeys, rightKeys, left, right, condition, BuildLeft)
+          broadcastHashJoin(leftKeys, rightKeys, left, right, condition, BuildLeft)
 
       case HashFilteredJoin(Inner, leftKeys, rightKeys, condition, left, right) =>
         val hashJoin =

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

@@ -37,24 +37,31 @@ class JoinSuite extends QueryTest {
   }
 
   test("plans broadcast hash join, given hints") {
-    TestSQLContext.set("spark.sql.join.broadcastTables", "testData2")
-    val rdd = sql("""SELECT * FROM testData t JOIN testData2 t2 ON t.key = t2.a""")
-    // Using `sparkPlan` because for relevant patterns in HashJoin to be
-    // matched, other strategies need to be applied.
-    val physical = rdd.queryExecution.sparkPlan
-    val bhj = physical.collect { case j: BroadcastHashJoin if j.buildSide == BuildRight => j }
-
-    assert(bhj.size === 1)
-    checkAnswer(
-      rdd,
-      Seq(
-        (1, "1", 1, 1),
-        (1, "1", 1, 2),
-        (2, "2", 2, 1),
-        (2, "2", 2, 2),
-        (3, "3", 3, 1),
-        (3, "3", 3, 2)
-      ))
+
+    def mkTest(buildSide: BuildSide, leftTable: String, rightTable: String) = {
+      TestSQLContext.set("spark.sql.join.broadcastTables",
+        s"${if (buildSide == BuildRight) rightTable else leftTable}")
+      val rdd = sql(s"""SELECT * FROM $leftTable JOIN $rightTable ON key = a""")
+      // Using `sparkPlan` because for relevant patterns in HashJoin to be
+      // matched, other strategies need to be applied.
+      val physical = rdd.queryExecution.sparkPlan
+      val bhj = physical.collect { case j: BroadcastHashJoin if j.buildSide == buildSide => j }
+
+      assert(bhj.size === 1, "planner does not pick up hint to generate broadcast hash join")
+      checkAnswer(
+        rdd,
+        Seq(
+          (1, "1", 1, 1),
+          (1, "1", 1, 2),
+          (2, "2", 2, 1),
+          (2, "2", 2, 2),
+          (3, "3", 3, 1),
+          (3, "3", 3, 2)
+        ))
+    }
+
+    mkTest(BuildRight, "testData", "testData2")
+    mkTest(BuildLeft, "testData", "testData2")
   }
 
   test("multiple-key equi-join is hash-join") {