apache/spark — pull request #1163 by concretevitamin (2014-06-20 22:48:11+00:00)
Title: [WIP] [SQL] SPARK-1800 Add broadcast hash join operator, and simple size-based auto conversion optimization.

This PR is based off Michael's [PR 734](https://github.com/apache/spark/pull/734) and adds a prototype optimization that detects small tables involved in joins so that they are broadcasted.

The following is a sample local hive/console session showing that the optimization works on a simple case. In it, `points.parquet` and `points2.parquet` are identical, and are generated from `data/kmeans_data.txt` (I read the file into Spark, defined a simple case class, saved to Parquet using Spark SQL).

```
scala> val points = parquetFile("points.parquet")
scala> val points2 = parquetFile("points2.parquet")
scala> points.registerAsTable("points")
scala> points2.registerAsTable("points2")
scala> hql("SELECT p.x, p.y FROM points p JOIN points2 p2 ON p.z = p2.z")
res3: org.apache.spark.sql.SchemaRDD =
SchemaRDD[2] at RDD at SchemaRDD.scala:100
== Query Plan ==
Project [x#0:0,y#1:1]
BroadcastHashJoin [z#2], [z#5], BuildRight
ParquetTableScan [x#0,y#1,z#2], (ParquetRelation points.parquet), []
ParquetTableScan [z#5], (ParquetRelation points2.parquet), []

scala> hql("SET spark.sql.auto.convert.join.size=10")
scala> hql("SELECT p.x, p.y FROM points p JOIN points2 p2 ON p.z = p2.z")
res6: org.apache.spark.sql.SchemaRDD =
SchemaRDD[6] at RDD at SchemaRDD.scala:100
== Query Plan ==
Project [x#0:0,y#1:1]
ShuffledHashJoin [z#2], [z#5], BuildRight
Exchange (HashPartitioning [z#2:2], 200)
ParquetTableScan [x#0,y#1,z#2], (ParquetRelation points.parquet), []
Exchange (HashPartitioning [z#5:0], 200)
ParquetTableScan [z#5], (ParquetRelation points2.parquet), []
```

Open issues:
- [ ] Figure out how to unit test the strategy.
- [ ] Check if this optimization is applicable to more cases than the one shown above.