This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new b3ecff34ab6 [SPARK-34079][SQL][FOLLOW-UP] Revert some changes in
InjectRuntimeFilterSuite
b3ecff34ab6 is described below
commit b3ecff34ab6e3f7b0852db7c0b391cefd176e6ca
Author: Peter Toth <[email protected]>
AuthorDate: Wed Apr 27 16:16:15 2022 +0800
[SPARK-34079][SQL][FOLLOW-UP] Revert some changes in
InjectRuntimeFilterSuite
To remove unnecessary changes from `InjectRuntimeFilterSuite` after
https://github.com/apache/spark/pull/32298. These are not needed after
https://github.com/apache/spark/pull/34929 as the final optimized plan does'n
contain any `WithCTE` nodes.
No need for those changes.
No.
Added new test.
Closes #36361 from
peter-toth/SPARK-34079-multi-column-scalar-subquery-follow-up-2.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit d05e01d54024e3844f1e48e03bad3fd814b8f6b9)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/InjectRuntimeFilterSuite.scala | 73 +++++++++++++++++-----
1 file changed, 57 insertions(+), 16 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
index b541419c823..6c6bd1799e1 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
@@ -19,13 +19,17 @@ package org.apache.spark.sql
import org.apache.spark.sql.catalyst.expressions.{Alias,
BloomFilterMightContain, Literal}
import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
BloomFilterAggregate}
+import org.apache.spark.sql.catalyst.optimizer.MergeScalarSubqueries
import org.apache.spark.sql.catalyst.plans.LeftSemi
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join,
LogicalPlan}
+import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec}
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
AQEPropagateEmptyRelation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}
import org.apache.spark.sql.types.{IntegerType, StructType}
-class InjectRuntimeFilterSuite extends QueryTest with SQLTestUtils with
SharedSparkSession {
+class InjectRuntimeFilterSuite extends QueryTest with SQLTestUtils with
SharedSparkSession
+ with AdaptiveSparkPlanHelper {
protected override def beforeAll(): Unit = {
super.beforeAll()
@@ -201,9 +205,16 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
sql("analyze table bf4 compute statistics for columns a4, b4, c4, d4, e4,
f4")
sql("analyze table bf5part compute statistics for columns a5, b5, c5, d5,
e5, f5")
sql("analyze table bf5filtered compute statistics for columns a5, b5, c5,
d5, e5, f5")
+
+ // `MergeScalarSubqueries` can duplicate subqueries in the optimized plan
and would make testing
+ // complicated.
+ conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key,
MergeScalarSubqueries.ruleName)
}
protected override def afterAll(): Unit = try {
+ conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key,
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.defaultValueString)
+
sql("DROP TABLE IF EXISTS bf1")
sql("DROP TABLE IF EXISTS bf2")
sql("DROP TABLE IF EXISTS bf3")
@@ -264,24 +275,28 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
}
}
- // `MergeScalarSubqueries` can duplicate subqueries in the optimized plan,
but the subqueries will
- // be reused in the physical plan.
- def getNumBloomFilters(plan: LogicalPlan, scalarSubqueryCTEMultiplicator:
Int = 1): Integer = {
- val numBloomFilterAggs = plan.collectWithSubqueries {
- case Aggregate(_, aggregateExpressions, _) =>
- aggregateExpressions.collect {
- case Alias(AggregateExpression(bfAgg: BloomFilterAggregate, _, _, _,
_), _) =>
- assert(bfAgg.estimatedNumItemsExpression.isInstanceOf[Literal])
- assert(bfAgg.numBitsExpression.isInstanceOf[Literal])
- 1
+ def getNumBloomFilters(plan: LogicalPlan): Integer = {
+ val numBloomFilterAggs = plan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: org.apache.spark.sql.catalyst.expressions.ScalarSubquery
+ => subquery.plan.collect {
+ case Aggregate(_, aggregateExpressions, _) =>
+ aggregateExpressions.map {
+ case Alias(AggregateExpression(bfAgg : BloomFilterAggregate, _,
_, _, _),
+ _) =>
+ assert(bfAgg.estimatedNumItemsExpression.isInstanceOf[Literal])
+ assert(bfAgg.numBitsExpression.isInstanceOf[Literal])
+ 1
+ }.sum
}.sum
+ }.sum
}.sum
val numMightContains = plan.collect {
case Filter(condition, _) => condition.collect {
case BloomFilterMightContain(_, _) => 1
}.sum
}.sum
- assert(numBloomFilterAggs == numMightContains *
scalarSubqueryCTEMultiplicator)
+ assert(numBloomFilterAggs == numMightContains)
numMightContains
}
@@ -385,7 +400,7 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
planEnabled = sql(query).queryExecution.optimizedPlan
checkAnswer(sql(query), expectedAnswer)
}
- assert(getNumBloomFilters(planEnabled, 2) ==
getNumBloomFilters(planDisabled) + 2)
+ assert(getNumBloomFilters(planEnabled) ==
getNumBloomFilters(planDisabled) + 2)
}
}
@@ -413,10 +428,10 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
checkAnswer(sql(query), expectedAnswer)
}
if (numFilterThreshold < 3) {
- assert(getNumBloomFilters(planEnabled, numFilterThreshold) ==
- getNumBloomFilters(planDisabled) + numFilterThreshold)
+ assert(getNumBloomFilters(planEnabled) ==
getNumBloomFilters(planDisabled)
+ + numFilterThreshold)
} else {
- assert(getNumBloomFilters(planEnabled, 2) ==
getNumBloomFilters(planDisabled) + 2)
+ assert(getNumBloomFilters(planEnabled) ==
getNumBloomFilters(planDisabled) + 2)
}
}
}
@@ -535,4 +550,30 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
""".stripMargin)
}
}
+
+ test("Merge runtime bloom filters") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000",
+ SQLConf.RUNTIME_FILTER_SEMI_JOIN_REDUCTION_ENABLED.key -> "false",
+ SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key -> "true",
+ // Re-enable `MergeScalarSubqueries`
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> "",
+ SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key ->
AQEPropagateEmptyRelation.ruleName) {
+
+ val query = "select * from bf1 join bf2 on bf1.c1 = bf2.c2 and " +
+ "bf1.b1 = bf2.b2 where bf2.a2 = 62"
+ val df = sql(query)
+ df.collect()
+ val plan = df.queryExecution.executedPlan
+
+ val subqueryIds = collectWithSubqueries(plan) { case s: SubqueryExec =>
s.id }
+ val reusedSubqueryIds = collectWithSubqueries(plan) {
+ case rs: ReusedSubqueryExec => rs.child.id
+ }
+
+ assert(subqueryIds.size == 1, "Missing or unexpected SubqueryExec in the
plan")
+ assert(reusedSubqueryIds.size == 1,
+ "Missing or unexpected reused ReusedSubqueryExec in the plan")
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
