[CARMEL-6496] Implement bloom filter join hint #1201

Author: wangyum

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala

@@ -148,7 +148,8 @@ object JoinStrategyHint {
     BROADCAST,
     SHUFFLE_MERGE,
     SHUFFLE_HASH,
-    SHUFFLE_REPLICATE_NL)
+    SHUFFLE_REPLICATE_NL,
+    BLOOM_FILTER_JOIN)
 }
 
 /**
@@ -192,6 +193,15 @@ case object SHUFFLE_REPLICATE_NL extends JoinStrategyHint {
     "SHUFFLE_REPLICATE_NL")
 }
 
+/**
+ * The hint for bloom filter join.
+ */
+case object BLOOM_FILTER_JOIN extends JoinStrategyHint {
+  override def displayName: String = "bloom_filter_join"
+  override def hintAliases: Set[String] = Set(
+    "BLOOM_FILTER_JOIN")
+}
+
 /**
  * An internal hint to discourage broadcast hash join, used by adaptive query execution.
  */

sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/DynamicBloomFilterPruning.scala

@@ -58,6 +58,14 @@ object DynamicBloomFilterPruning extends Rule[LogicalPlan]
       }.headOption
   }
 
+  private def hintToBuildBloomFilterLeft(hint: JoinHint): Boolean = {
+    hint.leftHint.exists(_.strategy.contains(BLOOM_FILTER_JOIN))
+  }
+
+  private def hintToBuildBloomFilterRight(hint: JoinHint): Boolean = {
+    hint.rightHint.exists(_.strategy.contains(BLOOM_FILTER_JOIN))
+  }
+
   private def rowCounts(plan: LogicalPlan): BigInt = {
     plan.stats.rowCount
       .getOrElse(plan.stats.sizeInBytes / EstimationUtils.getSizePerRow(plan.output))
@@ -164,24 +172,24 @@ object DynamicBloomFilterPruning extends Rule[LogicalPlan]
             val leftDistCnt = distinctCounts(l, left)
             val rightDistCnt = distinctCounts(r, right)
 
-            if (isRightSideSmall && canPruneLeft(joinType) && rightRowCnt > 0 &&
-              !rightHasScalarSubquery &&
-              rightRowCnt <= conf.dynamicBloomFilterJoinPruningMaxBloomFilterEntries &&
-              !hasDynamicBloomFilterPruningSubquery(right) && !right.isStreaming &&
-              getFilterableTableScan(l, left).exists(p =>
-                !hasSelectivePredicate(p) &&
-                pruningHasBenefit(rightRowCnt, rightDistCnt, leftDistCnt, p))) {
+            if (canPruneLeft(joinType) && (hintToBuildBloomFilterRight(hint) ||
+              (isRightSideSmall && rightRowCnt > 0 && !rightHasScalarSubquery &&
+                rightRowCnt <= conf.dynamicBloomFilterJoinPruningMaxBloomFilterEntries &&
+                !hasDynamicBloomFilterPruningSubquery(right) && !right.isStreaming &&
+                getFilterableTableScan(l, left).exists(p =>
+                  !hasSelectivePredicate(p) &&
+                    pruningHasBenefit(rightRowCnt, rightDistCnt, leftDistCnt, p))))) {
               newLeft = insertPredicate(l, newLeft, r, right, rightKeys, rightDistCnt, rightRowCnt)
               newHint = newHint.copy(leftHint = Some(HintInfo(strategy = Some(NO_BROADCAST_HASH))))
             }
 
-            if (isLeftSideSmall && canPruneRight(joinType) && leftRowCount > 0 &&
-              !leftHasScalarSubquery &&
-              leftRowCount <= conf.dynamicBloomFilterJoinPruningMaxBloomFilterEntries &&
-              !hasDynamicBloomFilterPruningSubquery(left) && !left.isStreaming &&
-              getFilterableTableScan(r, right).exists(p =>
-                !hasSelectivePredicate(p) &&
-                pruningHasBenefit(leftRowCount, leftDistCnt, rightDistCnt, p))) {
+            if (canPruneRight(joinType) && (hintToBuildBloomFilterLeft(hint) ||
+              (isLeftSideSmall && leftRowCount > 0 && !leftHasScalarSubquery &&
+                leftRowCount <= conf.dynamicBloomFilterJoinPruningMaxBloomFilterEntries &&
+                !hasDynamicBloomFilterPruningSubquery(left) && !left.isStreaming &&
+                getFilterableTableScan(r, right).exists(p =>
+                  !hasSelectivePredicate(p) &&
+                    pruningHasBenefit(leftRowCount, leftDistCnt, rightDistCnt, p))))) {
               newRight = insertPredicate(r, newRight, l, left, leftKeys, leftDistCnt, leftRowCount)
               newHint = newHint.copy(rightHint = Some(HintInfo(strategy = Some(NO_BROADCAST_HASH))))
             }

sql/core/src/test/scala/org/apache/spark/sql/DynamicBloomFilterJoinPruningSuite.scala

@@ -166,6 +166,15 @@ abstract class DynamicBloomFilterPruningSuiteBase
         checkBloomFilterPruningPredicate(df, true)
         checkAnswer(df, Row(2, 2) :: Nil)
       }
+
+      // test hint
+      withSQLConf(SQLConf.DYNAMIC_BLOOM_FILTER_JOIN_PRUNING_ENABLED.key -> "true") {
+        checkBloomFilterPruningPredicate(
+          sql("SELECT * FROM t1 f JOIN t2 s ON f.a = s.a AND f.b = 2"), false)
+        checkBloomFilterPruningPredicate(
+          sql("SELECT /*+ BLOOM_FILTER_JOIN(s) */ * FROM t1 f JOIN t2 s ON f.a = s.a AND f.b = 2"),
+          true)
+      }
     }
   }