Skip to content

[SQL] SPARK-1800 Add broadcast hash join operator & associated hints. #1163

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
concretevitamin wants to merge 18 commits into from
Closed

[SQL] SPARK-1800 Add broadcast hash join operator & associated hints. #1163

Changes from 1 commit
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
a8420ca
Copy records in executeCollect to avoid issues with mutable rows.
marmbrus May 11, 2014
cf6b381
Split out generic logic for hash joins and create two concrete physic…
marmbrus May 11, 2014
76ca434
A simple strategy that broadcasts tables only when they are found in …
marmbrus May 11, 2014
a92ed0c
Formatting.
marmbrus May 12, 2014
3e5d77c
WIP: giant and messy WIP.
concretevitamin Jun 18, 2014
0ad122f
Merge branch 'master' into auto-broadcast-hash-join
concretevitamin Jun 18, 2014
7c7158b
Prototype of auto conversion to broadcast hash join.
concretevitamin Jun 20, 2014
91461c2
Merge branch 'master' into auto-broadcast-hash-join
concretevitamin Jun 20, 2014
0e64b08
Scalastyle fix.
concretevitamin Jun 20, 2014
a4267be
Add test for broadcast hash join and related necessary refactorings:
concretevitamin Jun 24, 2014
6fd8443
Cut down size estimation related stuff.
concretevitamin Jun 24, 2014
a8a093e
Minor cleanups.
concretevitamin Jun 24, 2014
814b3bf
Merge branch 'master' into auto-broadcast-hash-join
concretevitamin Jun 24, 2014
ad6c7cc
Minor cleanups.
concretevitamin Jun 24, 2014
208d5f6
Make LeftSemiJoinHash mix in HashJoin.
concretevitamin Jun 24, 2014
440d277
Fixes to imports; add back requiredChildDistribution (lost when merging)
concretevitamin Jun 24, 2014
af080d7
Fix in joinIterators()'s next().
concretevitamin Jun 25, 2014
d0f4991
Fix bug in broadcast hash join & add test to cover it.
concretevitamin Jun 25, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Formatting.
  • Loading branch information
@marmbrus
marmbrus committed May 12, 2014
commit a92ed0cfbc9fbfada933f702e3879e70cb95283c
28 changes: 21 additions & 7 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,22 +34,36 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
*/
object HashJoin extends Strategy with PredicateHelper {
var broadcastTables: Seq[String] =
sparkContext.conf.get("spark.sql.hints.broadcastTables", "").split(",")
sparkContext.conf.get("spark.sql.hints.broadcastTables", "").split(",").toBuffer

def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {

case HashFilteredJoin(Inner, leftKeys, rightKeys, condition,
left, PhysicalOperation(_, _, b: BaseRelation)) if broadcastTables.contains(b.tableName)=>
case HashFilteredJoin(
Inner,
leftKeys,
rightKeys,
condition,
left,
right @ PhysicalOperation(_, _, b: BaseRelation))
if broadcastTables.contains(b.tableName)=>

val hashJoin =
execution.BroadcastHashJoin(
leftKeys, rightKeys, BuildRight, planLater(left), planLater(b))(sparkContext)
leftKeys, rightKeys, BuildRight, planLater(left), planLater(right))(sparkContext)
condition.map(Filter(_, hashJoin)).getOrElse(hashJoin) :: Nil

case HashFilteredJoin(Inner, leftKeys, rightKeys, condition,
PhysicalOperation(_, _, b: BaseRelation), right) if broadcastTables.contains(b.tableName)=>
case HashFilteredJoin(
Inner,
leftKeys,
rightKeys,
condition,
left @ PhysicalOperation(_, _, b: BaseRelation),
right)
if broadcastTables.contains(b.tableName) =>

val hashJoin =
execution.BroadcastHashJoin(
leftKeys, rightKeys, BuildLeft, planLater(b), planLater(right))(sparkContext)
leftKeys, rightKeys, BuildLeft, planLater(left), planLater(right))(sparkContext)
condition.map(Filter(_, hashJoin)).getOrElse(hashJoin) :: Nil

case HashFilteredJoin(Inner, leftKeys, rightKeys, condition, left, right) =>
Expand Down