[SPARK-4570][SQL]add BroadcastLeftSemiJoinHash

[wangxiaojing]

JIRA issue: SPARK-4570
We are planning to create a BroadcastLeftSemiJoinHash to implement the broadcast join for left semijoin
In left semijoin :
If the size of data from right side is smaller than the user-settable threshold AUTO_BROADCASTJOIN_THRESHOLD,
the planner would mark it as the broadcast relation and mark the other relation as the stream side. The broadcast table will be broadcasted to all of the executors involved in the join, as a org.apache.spark.broadcast.Broadcast object. It will use joins.BroadcastLeftSemiJoinHash.,else it will use joins.LeftSemiJoinHash.

The benchmark suggests these made the optimized version 4x faster when left semijoin

Original:
left semi join : 9288 ms 
Optimized:
left semi join : 1963 ms 

The micro benchmark load data1/kv3.txt into a normal Hive table.
Benchmark code:

 def benchmark(f: => Unit) = {
    val begin = System.currentTimeMillis()
    f
    val end = System.currentTimeMillis()
    end - begin
  }
  val sc = new SparkContext(
    new SparkConf()
      .setMaster("local")
      .setAppName(getClass.getSimpleName.stripSuffix("$")))
  val hiveContext = new HiveContext(sc)
  import hiveContext._
  sql("drop table if exists left_table")
  sql("drop table if exists right_table")
  sql( """create table left_table (key int, value string)
       """.stripMargin)
  sql( s"""load data local inpath "/data1/kv3.txt" into table left_table""")
  sql( """create table right_table (key int, value string)
       """.stripMargin)
  sql(
    """
      |from left_table
      |insert overwrite table right_table
      |select left_table.key, left_table.value
    """.stripMargin)

  val leftSimeJoin = sql(
    """select a.key from left_table a
      |left semi join right_table b on a.key = b.key""".stripMargin)
  val leftSemiJoinDuration = benchmark(leftSimeJoin.count())
  println(s"left semi join : $leftSemiJoinDuration ms ")

[AmplabJenkins]

Can one of the admins verify this patch?

[wangxiaojing]

@liancheng

[liancheng]

ok to test

[SparkQA]

Test build #23972 has started for PR 3442 at commit 3a63ecb.

• This patch merges cleanly.

[SparkQA]

Test build #23972 has finished for PR 3442 at commit 3a63ecb.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BroadcastLeftSemiJoinHash(

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23972/
Test FAILed.

[wangxiaojing]

@liancheng

[liancheng]

retest this please

[SparkQA]

Test build #23983 has started for PR 3442 at commit 3a63ecb.

• This patch merges cleanly.

[SparkQA]

Test build #23983 has finished for PR 3442 at commit 3a63ecb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BroadcastLeftSemiJoinHash(

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23983/
Test PASSed.

[marmbrus]

Remove blank line

[marmbrus]

Thanks for working on this!

[liancheng]

-$tmp: typo? And we can just use setConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD, tmp.toString) here.

[SparkQA]

Test build #24025 has started for PR 3442 at commit d410f67.

• This patch merges cleanly.

[SparkQA]

Test build #24025 has finished for PR 3442 at commit d410f67.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BroadcastLeftSemiJoinHash(

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24025/
Test PASSed.

[SparkQA]

Test build #24122 has started for PR 3442 at commit 4fdcfe7.

• This patch merges cleanly.

[SparkQA]

Test build #24122 has finished for PR 3442 at commit 4fdcfe7.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BroadcastLeftSemiJoinHash(

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24122/
Test PASSed.

[SparkQA]

Test build #24533 has started for PR 3442 at commit 3a58191.

• This patch merges cleanly.

[SparkQA]

Test build #24533 has finished for PR 3442 at commit 3a58191.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BroadcastLeftSemiJoinHash(

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24533/
Test FAILed.

[SparkQA]

Test build #24753 has started for PR 3442 at commit a4a43c9.

• This patch merges cleanly.

[SparkQA]

Test build #24753 has finished for PR 3442 at commit a4a43c9.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24753/
Test PASSed.

[marmbrus]

Thanks, merged to master.