Skip to content

Create a way to get the broadcast future from BroadcastHashJoin #152

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
wants to merge 3 commits into from
Closed

Create a way to get the broadcast future from BroadcastHashJoin #152

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5a14d70
Create a way to get the broadcast future from BroadcastHashJoin
mbautin Feb 22, 2016
6df7606
Fix @volatile syntax
mbautin Feb 22, 2016
1f1386f
Update BroadcastHashJoin.scala
mbautin Feb 22, 2016
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
24 changes: 22 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.joins
import scala.concurrent._
import scala.concurrent.duration._

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Expression
Expand Down Expand Up @@ -62,10 +63,29 @@ case class BroadcastHashJoin(
override def requiredChildDistribution: Seq[Distribution] =
UnspecifiedDistribution :: UnspecifiedDistribution :: Nil

private[this] val broadcastFutureInitLock = new Object()

// Use @volatile so we can read a snapshot of this without locking.
@volatile
private[this] var broadcastFutureValue: Option[Future[Broadcast[HashedRelation]]] = None

// Use lazy so that we won't do broadcast when calling explain but still cache the broadcast value
// for the same query.
Copy link

@markhamstra markhamstra Feb 22, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment needs to be updated. Have you checked that the behavior described in the comment is maintained?

Copy link

@markhamstra markhamstra Feb 22, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment needs to be updated. Have you checked that the behavior described in the comment is maintained?

@transient
private lazy val broadcastFuture = {
private def broadcastFuture = {
broadcastFutureInitLock.synchronized {
if (broadcastFutureValue.isEmpty) {
broadcastFutureValue = Some(createBroadcastFuture)
}
broadcastFutureValue.get
}
}

/**
* Exposes the broadcast future so we can do external accounting of memory usage.
*/
def broadcastFutureOpt: Option[Future[Broadcast[HashedRelation]]] = broadcastFutureValue

private def createBroadcastFuture: Future[Broadcast[HashedRelation]] = {
val numBuildRows = buildSide match {
case BuildLeft => longMetric("numLeftRows")
case BuildRight => longMetric("numRightRows")
Expand Down