Spark SQL add LeftSemiBloomFilterBroadcastJoin

Hi ,All .
I want to submit  a join operator called
LeftSemiBloomFilterBroadcastJoin (LeftSemiJoinBFB)
Sometimes the Semijoin's broadcast table can't fit memory.So  we can make it as Bloomfilter to  reduce the space  and then broadcast it do the mapside  join .
Some code  reference HashJoin and BroadcastNestedLoopJoin implementation.
The bloomfilter  code   use Shark's BloomFilter class implementation.


This main code is in the patch file 

here is the strategies code

  object LeftSemiJoinBFB extends Strategy with PredicateHelper {
    def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
      case LeftSemiJoinBFBFilteredJoin(Inner, leftKeys, rightKeys, condition, left, right) =>
        val hashJoin =
          execution.LeftSemiJoinBFBJoin(leftKeys, rightKeys, BuildRight, planLater(left), planLater(right),sparkContext)
        condition.map(Filter(_, hashJoin)).getOrElse(hashJoin) :: Nil
      case _ => Nil
    }
  }

here is the code patterns


object LeftSemiJoinBFBFilteredJoin extends Logging with PredicateHelper {
  /** (joinType, rightKeys, leftKeys, condition, leftChild, rightChild) */
  type ReturnType =
  (JoinType, Seq[Expression], Seq[Expression], Option[Expression], LogicalPlan, LogicalPlan)

  def unapply(plan: LogicalPlan): Option[ReturnType] = plan match {
    // All predicates can be evaluated for inner join (i.e., those that are in the ON
    // clause and WHERE clause.)
    case FilteredOperation(predicates, join @ Join(left, right, Inner, condition)) =>
      logger.debug(s"Considering Semi inner join on: ${predicates ++ condition}")
      splitPredicates(predicates ++ condition, join)
    case join @ Join(left, right, joinType, condition) =>
      logger.debug(s"Considering Semi join on: $condition")
      splitPredicates(condition.toSeq, join)
    case _ => None
  }


4 space indent