Github user concretevitamin commented on a diff in the pull request:
https://github.com/apache/spark/pull/1448#discussion_r15599752
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/joins.scala ---
@@ -332,71 +342,88 @@ case class BroadcastNestedLoopJoin(
}
}
- /** The Streamed Relation */
- def left = streamed
- /** The Broadcast relation */
- def right = broadcast
-
@transient lazy val boundCondition =
InterpretedPredicate(
condition
.map(c => BindReferences.bindReference(c, left.output ++
right.output))
.getOrElse(Literal(true)))
+ // TODO: low-hanging fruits for performance? (Reduce branching / pattern
matches?)
def execute() = {
val broadcastedRelation =
sqlContext.sparkContext.broadcast(broadcast.execute().map(_.copy()).collect().toIndexedSeq)
- val streamedPlusMatches = streamed.execute().mapPartitions {
streamedIter =>
+ /** All rows that either match both-way, or rows from streamed joined
with nulls. */
+ val matchesOrStreamedRowsWithNulls = streamed.execute().mapPartitions
{ streamedIter =>
val matchedRows = new ArrayBuffer[Row]
// TODO: Use Spark's BitSet.
- val includedBroadcastTuples = new
BitSet(broadcastedRelation.value.size)
+ val includedBroadcastTuples =
+ new scala.collection.mutable.BitSet(broadcastedRelation.value.size)
val joinedRow = new JoinedRow
streamedIter.foreach { streamedRow =>
var i = 0
- var matched = false
+ var streamRowMatched = 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
+ val jr = buildSide match {
+ case BuildRight => joinedRow(streamedRow, broadcastedRow)
+ case BuildLeft => joinedRow(broadcastedRow, streamedRow)
+ }
+ if (boundCondition(jr)) {
+ // Putting this branching inside this conditional: assume ++
has a
+ // much higher cost than another branch & pattern matching.
+ val br = buildSide match {
+ case BuildRight => streamedRow ++ broadcastedRow
+ case BuildLeft => broadcastedRow ++ streamedRow
+ }
+ matchedRows += buildRow(br)
+ streamRowMatched = true
includedBroadcastTuples += i
}
i += 1
}
- if (!matched && (joinType == LeftOuter || joinType == FullOuter)) {
- matchedRows += buildRow(streamedRow ++
Array.fill(right.output.size)(null))
+ (streamRowMatched, joinType, buildSide) match {
+ case (false, LeftOuter | FullOuter, BuildRight) =>
+ matchedRows += buildRow(streamedRow ++
Array.fill(right.output.size)(null))
+ case (false, RightOuter | FullOuter, BuildLeft) =>
+ matchedRows += buildRow(Array.fill(left.output.size)(null) ++
streamedRow)
+ case _ =>
}
}
Iterator((matchedRows, includedBroadcastTuples))
}
- val includedBroadcastTuples = streamedPlusMatches.map(_._2)
+ val includedBroadcastTuples = matchesOrStreamedRowsWithNulls.map(_._2)
val allIncludedBroadcastTuples =
if (includedBroadcastTuples.count == 0) {
new scala.collection.mutable.BitSet(broadcastedRelation.value.size)
} else {
- streamedPlusMatches.map(_._2).reduce(_ ++ _)
+ includedBroadcastTuples.reduce(_ ++ _)
}
- val rightOuterMatches: Seq[Row] =
- if (joinType == RightOuter || joinType == FullOuter) {
- broadcastedRelation.value.zipWithIndex.filter {
- case (row, i) => !allIncludedBroadcastTuples.contains(i)
- }.map {
- // TODO: Use projection.
- case (row, _) => buildRow(Vector.fill(left.output.size)(null) ++
row)
+ /** Rows from broadcasted joined with nulls. */
+ val broadcastRowsWithNulls: Seq[Row] =
+ broadcastedRelation.value.zipWithIndex.filter { case (row, i) =>
+ // keep all rows in the broadcast relation that are not matched
before
+ !allIncludedBroadcastTuples.contains(i)
+ }.map { case (row, _) =>
+ // TODO: Use projection.
+ val br = (joinType, buildSide) match {
+ case (RightOuter | FullOuter, BuildRight) =>
Vector.fill(left.output.size)(null) ++ row
--- End diff --
Hey @marmbrus -- now that codegen is in (congrats!), can you point to the
places that I should look at?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
