protected override def doExecute(): RDD[InternalRow] = {
  val numOutputRows = longMetric("numOutputRows")

  val broadcastRelation = buildPlan.executeBroadcast[HashedRelation]()
  streamedPlan.execute().mapPartitions { streamedIter =>
    val hashed = broadcastRelation.value.asReadOnlyCopy()
    TaskContext.get().taskMetrics().incPeakExecutionMemory(hashed.estimatedSize)
    join(streamedIter, hashed, numOutputRows)
  }
}

private def buildHashedRelation(iter: Iterator[InternalRow]): HashedRelation = {
  val buildDataSize = longMetric("buildDataSize")
  val buildTime = longMetric("buildTime")
  val start = System.nanoTime()
  val context = TaskContext.get()
  val relation = HashedRelation(iter, buildKeys, taskMemoryManager = context.taskMemoryManager())
  buildTime += (System.nanoTime() - start) / 1000000
  buildDataSize += relation.estimatedSize
  // This relation is usually used until the end of task.
  context.addTaskCompletionListener[Unit](_ => relation.close())
  relation
}

protected override def doExecute(): RDD[InternalRow] = {
  val numOutputRows = longMetric("numOutputRows")
  streamedPlan.execute().zipPartitions(buildPlan.execute()) { (streamIter, buildIter) =>
    val hashed = buildHashedRelation(buildIter)
    join(streamIter, hashed, numOutputRows)
  }
}

final def findNextInnerJoinRows(): Boolean = {
  while (advancedStreamed() && streamedRowKey.anyNull) {
    // Advance the streamed side of the join until we find the next row whose join key contains
    // no nulls or we hit the end of the streamed iterator.
  }
  val found = if (streamedRow == null) {
    // We have consumed the entire streamed iterator, so there can be no more matches.
    matchJoinKey = null
    bufferedMatches.clear()
    false
  } else if (matchJoinKey != null && keyOrdering.compare(streamedRowKey, matchJoinKey) == 0) {
    // The new streamed row has the same join key as the previous row, so return the same matches.
    true
  } else if (bufferedRow == null) {
    // The streamed row's join key does not match the current batch of buffered rows and there are
    // no more rows to read from the buffered iterator, so there can be no more matches.
    matchJoinKey = null
    bufferedMatches.clear()
    false
  } else {
    // Advance both the streamed and buffered iterators to find the next pair of matching rows.
    var comp = keyOrdering.compare(streamedRowKey, bufferedRowKey)
    do {
      if (streamedRowKey.anyNull) {
        advancedStreamed()
      } else {
        assert(!bufferedRowKey.anyNull)
        comp = keyOrdering.compare(streamedRowKey, bufferedRowKey)
        if (comp > 0) advancedBufferedToRowWithNullFreeJoinKey()
        else if (comp < 0) advancedStreamed()
      }
    } while (streamedRow != null && bufferedRow != null && comp != 0)
    if (streamedRow == null || bufferedRow == null) {
      // We have either hit the end of one of the iterators, so there can be no more matches.
      matchJoinKey = null
      bufferedMatches.clear()
      false
    } else {
      // The streamed row's join key matches the current buffered row's join, so walk through the
      // buffered iterator to buffer the rest of the matching rows.
      assert(comp == 0)
      bufferMatchingRows()
      true
    }
  }
  if (!found) eagerCleanupResources()
  found
}

final def findNextOuterJoinRows(): Boolean = {
  val found = if (!advancedStreamed()) {
    // We have consumed the entire streamed iterator, so there can be no more matches.
    matchJoinKey = null
    bufferedMatches.clear()
    false
  } else {
    if (matchJoinKey != null && keyOrdering.compare(streamedRowKey, matchJoinKey) == 0) {
      // Matches the current group, so do nothing.
    } else {
      // The streamed row does not match the current group.
      matchJoinKey = null
      bufferedMatches.clear()
      if (bufferedRow != null && !streamedRowKey.anyNull) {
        var comp = 1
        do {
          comp = keyOrdering.compare(streamedRowKey, bufferedRowKey)
        } while (comp > 0 && advancedBufferedToRowWithNullFreeJoinKey())
        if (comp == 0) {
          // We have found matches, so buffer them (this updates matchJoinKey)
          bufferMatchingRows()
        } else {
          // We have overshot the position where the row would be found, hence no matches.
        }
      }
    }
    // If there is a streamed input then we always return true
    true
  }
  if (!found) eagerCleanupResources()
  found
}

private def scanNextInBuffered(): Boolean = {
  while (leftIndex < leftMatches.size) {
    while (rightIndex < rightMatches.size) {
      joinedRow(leftMatches(leftIndex), rightMatches(rightIndex))
      if (boundCondition(joinedRow)) {
        leftMatched.set(leftIndex)
        rightMatched.set(rightIndex)
        rightIndex += 1
        return true
      }
      rightIndex += 1
    }
    rightIndex = 0
    if (!leftMatched.get(leftIndex)) {
      // the left row has never matched any right row, join it with null row
      joinedRow(leftMatches(leftIndex), rightNullRow)
      leftIndex += 1
      return true
    }
    leftIndex += 1
  }

  while (rightIndex < rightMatches.size) {
    if (!rightMatched.get(rightIndex)) {
      // the right row has never matched any left row, join it with null row
      joinedRow(leftNullRow, rightMatches(rightIndex))
      rightIndex += 1
      return true
    }
    rightIndex += 1
  }

  // There are no more valid matches in the left and right buffers
  false
}

private def findMatchingRows(matchingKey: InternalRow): Unit = {
  leftMatches.clear()
  rightMatches.clear()
  leftIndex = 0
  rightIndex = 0

  while (leftRowKey != null && keyOrdering.compare(leftRowKey, matchingKey) == 0) {
    leftMatches += leftRow.copy()
    advancedLeft()
  }
  while (rightRowKey != null && keyOrdering.compare(rightRowKey, matchingKey) == 0) {
    rightMatches += rightRow.copy()
    advancedRight()
  }

  if (leftMatches.size <= leftMatched.capacity) {
    leftMatched.clearUntil(leftMatches.size)
  } else {
    leftMatched = new BitSet(leftMatches.size)
  }
  if (rightMatches.size <= rightMatched.capacity) {
    rightMatched.clearUntil(rightMatches.size)
  } else {
    rightMatched = new BitSet(rightMatches.size)
  }
}

protected override def doExecute(): RDD[InternalRow] = {
  val numOutputRows = longMetric("numOutputRows")
  val spillThreshold = getSpillThreshold
  val inMemoryThreshold = getInMemoryThreshold
  left.execute().zipPartitions(right.execute()) { (leftIter, rightIter) =>
    val boundCondition: (InternalRow) => Boolean = {
      condition.map { cond =>
        newPredicate(cond, left.output ++ right.output).eval _
      }.getOrElse {
        (r: InternalRow) => true
      }
    }

    // An ordering that can be used to compare keys from both sides.
    val keyOrdering = newNaturalAscendingOrdering(leftKeys.map(_.dataType))
    val resultProj: InternalRow => InternalRow = UnsafeProjection.create(output, output)

    joinType match {
      case _: InnerLike =>
      new RowIterator {
        private[this] var currentLeftRow: InternalRow = _
        private[this] var currentRightMatches: ExternalAppendOnlyUnsafeRowArray = _
        private[this] var rightMatchesIterator: Iterator[UnsafeRow] = null
        private[this] val smjScanner = new SortMergeJoinScanner(
          createLeftKeyGenerator(),
          createRightKeyGenerator(),
          keyOrdering,
          RowIterator.fromScala(leftIter),
          RowIterator.fromScala(rightIter),
          inMemoryThreshold,
          spillThreshold,
          cleanupResources
        )
        private[this] val joinRow = new JoinedRow

        if (smjScanner.findNextInnerJoinRows()) {
          currentRightMatches = smjScanner.getBufferedMatches
          currentLeftRow = smjScanner.getStreamedRow
          rightMatchesIterator = currentRightMatches.generateIterator()
        }

        override def advanceNext(): Boolean = {
          while (rightMatchesIterator != null) {
            if (!rightMatchesIterator.hasNext) {
              if (smjScanner.findNextInnerJoinRows()) {
                currentRightMatches = smjScanner.getBufferedMatches
                currentLeftRow = smjScanner.getStreamedRow
                rightMatchesIterator = currentRightMatches.generateIterator()
              } else {
                currentRightMatches = null
                currentLeftRow = null
                rightMatchesIterator = null
                return false
              }
            }
            joinRow(currentLeftRow, rightMatchesIterator.next())
            if (boundCondition(joinRow)) {
              numOutputRows += 1
              return true
            }
          }
          false
        }

        override def getRow: InternalRow = resultProj(joinRow)
      }.toScala

      case LeftOuter =>
      val smjScanner = new SortMergeJoinScanner(
        streamedKeyGenerator = createLeftKeyGenerator(),
        bufferedKeyGenerator = createRightKeyGenerator(),
        keyOrdering,
        streamedIter = RowIterator.fromScala(leftIter),
        bufferedIter = RowIterator.fromScala(rightIter),
        inMemoryThreshold,
        spillThreshold,
        cleanupResources
      )
      val rightNullRow = new GenericInternalRow(right.output.length)
      new LeftOuterIterator(
        smjScanner, rightNullRow, boundCondition, resultProj, numOutputRows).toScala

      case RightOuter =>
      val smjScanner = new SortMergeJoinScanner(
        streamedKeyGenerator = createRightKeyGenerator(),
        bufferedKeyGenerator = createLeftKeyGenerator(),
        keyOrdering,
        streamedIter = RowIterator.fromScala(rightIter),
        bufferedIter = RowIterator.fromScala(leftIter),
        inMemoryThreshold,
        spillThreshold,
        cleanupResources
      )
      val leftNullRow = new GenericInternalRow(left.output.length)
      new RightOuterIterator(
        smjScanner, leftNullRow, boundCondition, resultProj, numOutputRows).toScala

      case FullOuter =>
      val leftNullRow = new GenericInternalRow(left.output.length)
      val rightNullRow = new GenericInternalRow(right.output.length)
      val smjScanner = new SortMergeFullOuterJoinScanner(
        leftKeyGenerator = createLeftKeyGenerator(),
        rightKeyGenerator = createRightKeyGenerator(),
        keyOrdering,
        leftIter = RowIterator.fromScala(leftIter),
        rightIter = RowIterator.fromScala(rightIter),
        boundCondition,
        leftNullRow,
        rightNullRow)

      new FullOuterIterator(
        smjScanner,
        resultProj,
        numOutputRows).toScala

      case LeftSemi =>
      new RowIterator {
        private[this] var currentLeftRow: InternalRow = _
        private[this] val smjScanner = new SortMergeJoinScanner(
          createLeftKeyGenerator(),
          createRightKeyGenerator(),
          keyOrdering,
          RowIterator.fromScala(leftIter),
          RowIterator.fromScala(rightIter),
          inMemoryThreshold,
          spillThreshold,
          cleanupResources
        )
        private[this] val joinRow = new JoinedRow

        override def advanceNext(): Boolean = {
          while (smjScanner.findNextInnerJoinRows()) {
            val currentRightMatches = smjScanner.getBufferedMatches
            currentLeftRow = smjScanner.getStreamedRow
            if (currentRightMatches != null && currentRightMatches.length > 0) {
              val rightMatchesIterator = currentRightMatches.generateIterator()
              while (rightMatchesIterator.hasNext) {
                joinRow(currentLeftRow, rightMatchesIterator.next())
                if (boundCondition(joinRow)) {
                  numOutputRows += 1
                  return true
                }
              }
            }
          }
          false
        }

        override def getRow: InternalRow = currentLeftRow
      }.toScala

      case LeftAnti =>
      new RowIterator {
        private[this] var currentLeftRow: InternalRow = _
        private[this] val smjScanner = new SortMergeJoinScanner(
          createLeftKeyGenerator(),
          createRightKeyGenerator(),
          keyOrdering,
          RowIterator.fromScala(leftIter),
          RowIterator.fromScala(rightIter),
          inMemoryThreshold,
          spillThreshold,
          cleanupResources
        )
        private[this] val joinRow = new JoinedRow

        override def advanceNext(): Boolean = {
          while (smjScanner.findNextOuterJoinRows()) {
            currentLeftRow = smjScanner.getStreamedRow
            val currentRightMatches = smjScanner.getBufferedMatches
            if (currentRightMatches == null || currentRightMatches.length == 0) {
              numOutputRows += 1
              return true
            }
            var found = false
            val rightMatchesIterator = currentRightMatches.generateIterator()
            while (!found && rightMatchesIterator.hasNext) {
              joinRow(currentLeftRow, rightMatchesIterator.next())
              if (boundCondition(joinRow)) {
                found = true
              }
            }
            if (!found) {
              numOutputRows += 1
              return true
            }
          }
          false
        }

        override def getRow: InternalRow = currentLeftRow
      }.toScala

      case j: ExistenceJoin =>
      new RowIterator {
        private[this] var currentLeftRow: InternalRow = _
        private[this] val result: InternalRow = new GenericInternalRow(Array[Any](null))
        private[this] val smjScanner = new SortMergeJoinScanner(
          createLeftKeyGenerator(),
          createRightKeyGenerator(),
          keyOrdering,
          RowIterator.fromScala(leftIter),
          RowIterator.fromScala(rightIter),
          inMemoryThreshold,
          spillThreshold,
          cleanupResources
        )
        private[this] val joinRow = new JoinedRow

        override def advanceNext(): Boolean = {
          while (smjScanner.findNextOuterJoinRows()) {
            currentLeftRow = smjScanner.getStreamedRow
            val currentRightMatches = smjScanner.getBufferedMatches
            var found = false
            if (currentRightMatches != null && currentRightMatches.length > 0) {
              val rightMatchesIterator = currentRightMatches.generateIterator()
              while (!found && rightMatchesIterator.hasNext) {
                joinRow(currentLeftRow, rightMatchesIterator.next())
                if (boundCondition(joinRow)) {
                  found = true
                }
              }
            }
            result.setBoolean(0, found)
            numOutputRows += 1
            return true
          }
          false
        }

        override def getRow: InternalRow = resultProj(joinRow(currentLeftRow, result))
      }.toScala

      case x =>
      throw new IllegalArgumentException(
        s"SortMergeJoin should not take $x as the JoinType")
    }

  }
}

def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {

  // --- BroadcastHashJoin --------------------------------------------------------------------

  // broadcast hints were specified
  case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)
  if canBroadcastByHints(joinType, left, right) =>
  val buildSide = broadcastSideByHints(joinType, left, right)
  Seq(joins.BroadcastHashJoinExec(
    leftKeys, rightKeys, joinType, buildSide, condition, planLater(left), planLater(right)))

  // broadcast hints were not specified, so need to infer it from size and configuration.
  case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)
  if canBroadcastBySizes(joinType, left, right) =>
  val buildSide = broadcastSideBySizes(joinType, left, right)
  Seq(joins.BroadcastHashJoinExec(
    leftKeys, rightKeys, joinType, buildSide, condition, planLater(left), planLater(right)))

  // --- ShuffledHashJoin ---------------------------------------------------------------------

  case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)
  if !conf.preferSortMergeJoin && canBuildRight(joinType) && canBuildLocalHashMap(right)
  && muchSmaller(right, left) ||
  !RowOrdering.isOrderable(leftKeys) =>
  Seq(joins.ShuffledHashJoinExec(
    leftKeys, rightKeys, joinType, BuildRight, condition, planLater(left), planLater(right)))

  case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)
  if !conf.preferSortMergeJoin && canBuildLeft(joinType) && canBuildLocalHashMap(left)
  && muchSmaller(left, right) ||
  !RowOrdering.isOrderable(leftKeys) =>
  Seq(joins.ShuffledHashJoinExec(
    leftKeys, rightKeys, joinType, BuildLeft, condition, planLater(left), planLater(right)))

  // --- SortMergeJoin ------------------------------------------------------------

  case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right)
  if RowOrdering.isOrderable(leftKeys) =>
  joins.SortMergeJoinExec(
    leftKeys, rightKeys, joinType, condition, planLater(left), planLater(right)) :: Nil

  // --- Without joining keys ------------------------------------------------------------

  // Pick BroadcastNestedLoopJoin if one side could be broadcast
  case j @ logical.Join(left, right, joinType, condition)
  if canBroadcastByHints(joinType, left, right) =>
  val buildSide = broadcastSideByHints(joinType, left, right)
  joins.BroadcastNestedLoopJoinExec(
    planLater(left), planLater(right), buildSide, joinType, condition) :: Nil

  case j @ logical.Join(left, right, joinType, condition)
  if canBroadcastBySizes(joinType, left, right) =>
  val buildSide = broadcastSideBySizes(joinType, left, right)
  joins.BroadcastNestedLoopJoinExec(
    planLater(left), planLater(right), buildSide, joinType, condition) :: Nil

  // Pick CartesianProduct for InnerJoin
  case logical.Join(left, right, _: InnerLike, condition) =>
  joins.CartesianProductExec(planLater(left), planLater(right), condition) :: Nil

  case logical.Join(left, right, joinType, condition) =>
  val buildSide = broadcastSide(
    left.stats.hints.broadcast, right.stats.hints.broadcast, left, right)
  // This join could be very slow or OOM
  joins.BroadcastNestedLoopJoinExec(
    planLater(left), planLater(right), buildSide, joinType, condition) :: Nil

  // --- Cases where this strategy does not apply ---------------------------------------------

  case _ => Nil
}