Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1448#discussion_r15267580
--- 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] =
--- End diff --
Can we just use a while loop instead of the collection actions
(`zipWithIndex`, `map`, `filter`, `filter`, and `map` again)? While loop just
require a single scan instead of 5.
---
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.
---
