Mailing-List: contact reviews-help@spark.apache.org; run by ezmlm
Precedence: bulk
Reply-To: reviews@spark.apache.org
From: concretevitamin <git@git.apache.org>
To: reviews@spark.apache.org
Reply-To: reviews@spark.apache.org
References: <git-pr-1448-spark@git.apache.org>
In-Reply-To: <git-pr-1448-spark@git.apache.org>
Subject: [GitHub] spark pull request: [SPARK-2531] [SQL] Make
 BroadcastNestedLoopJoi...
Content-Type: text/plain
Message-Id: <20140730174322.D815D99F8E6@tyr.zones.apache.org>
Date: Wed, 30 Jul 2014 17:43:22 +0000 (UTC)

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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---