You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2022/03/17 09:06:13 UTC
[spark] branch branch-3.2 updated: [SPARK-37995][SQL] PlanAdaptiveDynamicPruningFilters should use prepareExecutedPlan rather than createSparkPlan to re-plan subquery
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new a27f13e [SPARK-37995][SQL] PlanAdaptiveDynamicPruningFilters should use prepareExecutedPlan rather than createSparkPlan to re-plan subquery
a27f13e is described below
commit a27f13ea8c14d72558c77bfb75d725d649cd2081
Author: ulysses-you <ul...@gmail.com>
AuthorDate: Thu Mar 17 16:58:22 2022 +0800
[SPARK-37995][SQL] PlanAdaptiveDynamicPruningFilters should use prepareExecutedPlan rather than createSparkPlan to re-plan subquery
### What changes were proposed in this pull request?
Use exists adaptive execution context to re-compile subquery.
### Why are the changes needed?
If a subquery which is inferred by dpp contains a nested subquery, AQE can not compile it correctly.
the added test will fail before this pr:
```java
java.lang.ClassCastException: org.apache.spark.sql.catalyst.plans.logical.Aggregate cannot be cast to org.apache.spark.sql.execution.SparkPlan
at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableLike.map(TraversableLike.scala:286)
at scala.collection.TraversableLike.map$(TraversableLike.scala:279)
at scala.collection.AbstractTraversable.map(Traversable.scala:108)
at org.apache.spark.sql.execution.SparkPlanInfo$.fromSparkPlan(SparkPlanInfo.scala:75)
at org.apache.spark.sql.execution.SparkPlanInfo$.$anonfun$fromSparkPlan$3(SparkPlanInfo.scala:75)
```
### Does this PR introduce _any_ user-facing change?
yes, bug fix
### How was this patch tested?
add test
Closes #35849 from ulysses-you/SPARK-37995.
Authored-by: ulysses-you <ul...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 3afc4fb08b01597a5677ce706731639c687fd2dd)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../apache/spark/sql/execution/QueryExecution.scala | 13 +++++++++++++
.../adaptive/PlanAdaptiveDynamicPruningFilters.scala | 11 ++++-------
.../spark/sql/DynamicPartitionPruningSuite.scala | 20 +++++++++++++++++++-
3 files changed, 36 insertions(+), 8 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index 0d3b752..4c5b0cc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
@@ -482,6 +482,19 @@ object QueryExecution {
prepareExecutedPlan(spark, sparkPlan)
}
+ /**
+ * Prepare the [[SparkPlan]] for execution using exists adaptive execution context.
+ * This method is only called by [[PlanAdaptiveDynamicPruningFilters]].
+ */
+ def prepareExecutedPlan(
+ session: SparkSession,
+ plan: LogicalPlan,
+ context: AdaptiveExecutionContext): SparkPlan = {
+ val sparkPlan = createSparkPlan(session, session.sessionState.planner, plan.clone())
+ val preparationRules = preparations(session, Option(InsertAdaptiveSparkPlan(context)), true)
+ prepareForExecution(preparationRules, sparkPlan.clone())
+ }
+
private val currentCteMap = new ThreadLocal[mutable.HashMap[Long, CTERelationDef]]()
def cteMap: mutable.HashMap[Long, CTERelationDef] = currentCteMap.get()
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala
index f9c1bbe..aa8f236 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala
@@ -71,13 +71,10 @@ case class PlanAdaptiveDynamicPruningFilters(
val aggregate = Aggregate(Seq(alias), Seq(alias), buildPlan)
val session = adaptivePlan.context.session
- val planner = session.sessionState.planner
- // Here we can't call the QueryExecution.prepareExecutedPlan() method to
- // get the sparkPlan as Non-AQE use case, which will cause the physical
- // plan optimization rules be inserted twice, once in AQE framework and
- // another in prepareExecutedPlan() method.
- val sparkPlan = QueryExecution.createSparkPlan(session, planner, aggregate)
- val newAdaptivePlan = adaptivePlan.copy(inputPlan = sparkPlan)
+ val sparkPlan = QueryExecution.prepareExecutedPlan(
+ session, aggregate, adaptivePlan.context)
+ assert(sparkPlan.isInstanceOf[AdaptiveSparkPlanExec])
+ val newAdaptivePlan = sparkPlan.asInstanceOf[AdaptiveSparkPlanExec]
val values = SubqueryExec(name, newAdaptivePlan)
DynamicPruningExpression(InSubqueryExec(value, values, exprId))
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
index fe84bbb..1033d43 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
@@ -1567,7 +1567,25 @@ class DynamicPartitionPruningV1SuiteAEOff extends DynamicPartitionPruningV1Suite
with DisableAdaptiveExecutionSuite
class DynamicPartitionPruningV1SuiteAEOn extends DynamicPartitionPruningV1Suite
- with EnableAdaptiveExecutionSuite
+ with EnableAdaptiveExecutionSuite {
+
+ test("SPARK-37995: PlanAdaptiveDynamicPruningFilters should use prepareExecutedPlan " +
+ "rather than createSparkPlan to re-plan subquery") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ val df = sql(
+ """
+ |SELECT f.date_id, f.store_id FROM fact_sk f
+ |JOIN dim_store s ON f.store_id = s.store_id AND s.country = 'NL'
+ |WHERE s.state_province != (SELECT max(state_province) FROM dim_stats)
+ """.stripMargin)
+
+ checkPartitionPruningPredicate(df, true, false)
+ checkAnswer(df, Row(1000, 1) :: Row(1010, 2) :: Row(1020, 2) :: Nil)
+ }
+ }
+}
abstract class DynamicPartitionPruningV2Suite extends DynamicPartitionPruningSuiteBase {
override protected def runAnalyzeColumnCommands: Boolean = false
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org