Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1448#discussion_r15267261
--- 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 --
Actually I think a lot of this is fixed in the codegen PR (i.e. we use
projection instead of creating new arrays everytime). Perhaps we should
revisit this PR after that has been merged.
---
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.
---
