[HADP-55621] Avoid apply bloom filter pruning on the build bloom filter side (apache#653)

Author: wangyum

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

@@ -360,7 +360,7 @@ object RuntimeFilterPruning
   private def hasDynamicPruningSubquery(plan: LogicalPlan): Boolean = {
     plan match {
       case Filter(_: DynamicPruningSubquery, _) => true
-      case _ => false
+      case _ => plan.children.exists(hasDynamicPruningSubquery)
     }
   }
 

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

@@ -22,6 +22,7 @@ import java.sql.{Date, Timestamp}
 import org.scalatest.GivenWhenThen
 
 import org.apache.spark.sql.catalyst.expressions.{BloomFilterMightContain, CodegenObjectFactoryMode, DynamicPruningExpression, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.Filter
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.adaptive._
 import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
@@ -356,4 +357,33 @@ class DynamicBloomFilterPruningSuiteAEOn extends DynamicBloomFilterPruningSuiteB
       }
     }
   }
+
+  test("HADP-55621: Avoid apply bloom filter pruning on the build bloom filter side") {
+    withTable("t1", "t2", "t3") {
+      sql(
+        """
+          |CREATE TABLE t1 using parquet AS
+          |SELECT id, split(concat(id, ','), ',') AS s FROM range(1000)
+          |""".stripMargin)
+      sql(
+        """
+          |CREATE TABLE t2 using parquet AS
+          |SELECT id FROM range(10000000)
+          |""".stripMargin)
+      withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "12mb") {
+        val df = sql(
+          """
+            |SELECT * FROM (
+            |  SELECT id, tmp.listings display_id FROM t1 LATERAL VIEW explode(s) tmp AS listings
+            |)
+            |WHERE id IN (SELECT id FROM t2) AND display_id IN (SELECT id FROM t2)
+            |""".stripMargin)
+        val containsBloomFilter = df.queryExecution.optimizedPlan.exists {
+          case f: Filter => f.condition.exists(_.isInstanceOf[BloomFilterMightContain])
+          case _ => false
+        }
+        assert(containsBloomFilter)
+      }
+    }
+  }
 }