support leftsemijoin for sparkSQL

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala

@@ -22,3 +22,4 @@ case object Inner extends JoinType
 case object LeftOuter extends JoinType
 case object RightOuter extends JoinType
 case object FullOuter extends JoinType
+case object LeftSemi extends JoinType

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.plans.logical
 
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.JoinType
+import org.apache.spark.sql.catalyst.plans.{LeftSemi, JoinType}
 import org.apache.spark.sql.catalyst.types._
 
 case class Project(projectList: Seq[NamedExpression], child: LogicalPlan) extends UnaryNode {
@@ -81,7 +81,12 @@ case class Join(
   condition: Option[Expression]) extends BinaryNode {
 
   def references = condition.map(_.references).getOrElse(Set.empty)
-  def output = left.output ++ right.output
+  def output = joinType match {
+    case LeftSemi =>
+      left.output
+    case _ =>
+      left.output ++ right.output
+  }
 }
 
 case class InsertIntoTable(

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

@@ -165,36 +165,66 @@ case class BroadcastNestedLoopJoin(
   def execute() = {
     val broadcastedRelation = sc.broadcast(broadcast.execute().map(_.copy()).collect().toIndexedSeq)
 
-    val streamedPlusMatches = streamed.execute().mapPartitions { streamedIter =>
-      val matchedRows = new ArrayBuffer[Row]
-      // TODO: Use Spark's BitSet.
-      val includedBroadcastTuples = new BitSet(broadcastedRelation.value.size)
-      val joinedRow = new JoinedRow
-
-      streamedIter.foreach { streamedRow =>
-        var i = 0
-        var matched = false
-
-        while (i < broadcastedRelation.value.size) {
-          // TODO: One bitset per partition instead of per row.
-          val broadcastedRow = broadcastedRelation.value(i)
-          if (boundCondition(joinedRow(streamedRow, broadcastedRow))) {
-            matchedRows += buildRow(streamedRow ++ broadcastedRow)
-            matched = true
-            includedBroadcastTuples += i
-          }
-          i += 1
+    val streamedPlusMatches = joinType match {
+      case LeftSemi =>
+        streamed.execute().mapPartitions {
+          streamedIter =>
+            val matchedRows = new ArrayBuffer[Row]
+            val joinedRow = new JoinedRow
+
+            streamedIter.foreach {
+              streamedRow =>
+                var i = 0
+                var matched = false
+
+                while (i < broadcastedRelation.value.size && !matched) {
+                  // TODO: One bitset per partition instead of per row.
+                  val broadcastedRow = broadcastedRelation.value(i)
+                    if (boundCondition(joinedRow(streamedRow, broadcastedRow))) {
+                      matchedRows += streamedRow
+                      matched = true
+                  }
+                  i += 1
+                }
+            }
+            Iterator((matchedRows, null))
         }
-
-        if (!matched && (joinType == LeftOuter || joinType == FullOuter)) {
-          matchedRows += buildRow(streamedRow ++ Array.fill(right.output.size)(null))
+      case _ =>
+        streamed.execute().mapPartitions {
+          streamedIter =>
+            val matchedRows = new ArrayBuffer[Row]
+            // TODO: Use Spark's BitSet.
+            val includedBroadcastTuples = new BitSet(broadcastedRelation.value.size)
+            val joinedRow = new JoinedRow
+
+            val rightNull = Array.fill(right.output.size)(null)
+
+            streamedIter.foreach {
+              streamedRow =>
+                var i = 0
+                var matched = false
+
+                while (i < broadcastedRelation.value.size) {
+                  // TODO: One bitset per partition instead of per row.
+                  val broadcastedRow = broadcastedRelation.value(i)
+                  if (boundCondition(joinedRow(streamedRow, broadcastedRow))) {
+                    matchedRows += joinedRow(streamedRow.copy, broadcastedRow)
+                    matched = true
+                    includedBroadcastTuples += i
+                  }
+                  i += 1
+                }
+
+                if (!matched && (joinType == LeftOuter || joinType == FullOuter)) {
+                  matchedRows += joinedRow(streamedRow.copy, rightNull)
+                }
+            }
+            Iterator((matchedRows, includedBroadcastTuples))
         }
-      }
-      Iterator((matchedRows, includedBroadcastTuples))
     }
 
-    val includedBroadcastTuples = streamedPlusMatches.map(_._2)
-    val allIncludedBroadcastTuples =
+    lazy val includedBroadcastTuples = streamedPlusMatches.map(_._2)
+    lazy val allIncludedBroadcastTuples =
       if (includedBroadcastTuples.count == 0) {
         new scala.collection.mutable.BitSet(broadcastedRelation.value.size)
       } else {

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala

@@ -676,6 +676,7 @@ object HiveQl {
         case "TOK_RIGHTOUTERJOIN" => RightOuter
         case "TOK_LEFTOUTERJOIN" => LeftOuter
         case "TOK_FULLOUTERJOIN" => FullOuter
+        case "TOK_LEFTSEMIJOIN" => LeftSemi
       }
       assert(other.size <= 1, "Unhandled join clauses.")
       Join(nodeToRelation(relation1),

sql/hive/src/test/resources/golden/leftsemijoin-10-89737a8857b5b61cc909e0c797f86aea

@@ -0,0 +1,4 @@
+Hank	2
+Hank	2
+Joe	2
+Joe	2

sql/hive/src/test/resources/golden/leftsemijoin-2-43d53504df013e6b35f81811138a167a

@@ -0,0 +1 @@
+0

sql/hive/src/test/resources/golden/leftsemijoin-8-73cad58a10a1483ccb15e94a857013

@@ -0,0 +1,4 @@
+Hank	2
+Hank	2
+Joe	2
+Joe	2

sql/hive/src/test/resources/golden/leftsemijoin-9-c5efa6b8771a51610d655be461670e1e

@@ -0,0 +1,2 @@
+2	Tie
+2	Tie

sql/hive/src/test/resources/golden/leftsemijoin_mr-10-9914f44ecb6ae7587b62e5349ff60d04

@@ -0,0 +1 @@
+0

sql/hive/src/test/resources/golden/leftsemijoin_mr-11-2027ecb1495d5550c5d56abf6b95b0a7

@@ -0,0 +1,2 @@
+1
+1

sql/hive/src/test/resources/golden/leftsemijoin_mr-4-333895fe6abca27c8edb5c91bfe10d2f

@@ -0,0 +1,2 @@
+1
+1

sql/hive/src/test/resources/golden/leftsemijoin_mr-5-896d0948c1df849df9764a6d8ad8fff9

@@ -0,0 +1,20 @@
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1

sql/hive/src/test/resources/golden/leftsemijoin_mr-6-b1e2ade89ae898650f0be4f796d8947b

@@ -0,0 +1 @@
+0

sql/hive/src/test/resources/golden/leftsemijoin_mr-7-8e9c2969b999557363e40f9ebb3f6d7c

@@ -0,0 +1 @@
+0

sql/hive/src/test/resources/golden/leftsemijoin_mr-8-c61b972d4409babe41d8963e841af45b

@@ -0,0 +1 @@
+0

sql/hive/src/test/resources/golden/leftsemijoin_mr-9-2027ecb1495d5550c5d56abf6b95b0a7

@@ -0,0 +1,2 @@
+1
+1

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala

@@ -440,6 +440,8 @@ class HiveCompatibilitySuite extends HiveQueryFileTest {
     "join_view",
     "lateral_view_cp",
     "lateral_view_ppd",
+    "leftsemijoin",
+    "leftsemijoin_mr",
     "lineage1",
     "literal_double",
     "literal_ints",