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 2020/08/03 13:00:45 UTC
[spark] branch branch-3.0 updated: [SPARK-32083][SQL][3.0] AQE
should not coalesce partitions for SinglePartition
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 64b3b56 [SPARK-32083][SQL][3.0] AQE should not coalesce partitions for SinglePartition
64b3b56 is described below
commit 64b3b56698dcb8446dd2801ce943938177d3fbd5
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Mon Aug 3 12:56:23 2020 +0000
[SPARK-32083][SQL][3.0] AQE should not coalesce partitions for SinglePartition
This is a partial backport of https://github.com/apache/spark/pull/29307
Most of the changes are not needed because https://github.com/apache/spark/pull/28226 is in master only.
This PR only backports the safeguard in `ShuffleExchangeExec.canChangeNumPartitions`
Closes #29321 from cloud-fan/aqe.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../scala/org/apache/spark/sql/execution/SparkStrategies.scala | 4 ++--
.../spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala | 6 ++++--
.../apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala | 7 ++++++-
3 files changed, 12 insertions(+), 5 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 28d1ccb..f836deb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -752,7 +752,7 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case logical.Repartition(numPartitions, shuffle, child) =>
if (shuffle) {
ShuffleExchangeExec(RoundRobinPartitioning(numPartitions),
- planLater(child), canChangeNumPartitions = false) :: Nil
+ planLater(child), noUserSpecifiedNumPartition = false) :: Nil
} else {
execution.CoalesceExec(numPartitions, planLater(child)) :: Nil
}
@@ -786,7 +786,7 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
execution.RangeExec(r) :: Nil
case r: logical.RepartitionByExpression =>
exchange.ShuffleExchangeExec(
- r.partitioning, planLater(r.child), canChangeNumPartitions = false) :: Nil
+ r.partitioning, planLater(r.child), noUserSpecifiedNumPartition = false) :: Nil
case ExternalRDD(outputObjAttr, rdd) => ExternalRDDScanExec(outputObjAttr, rdd) :: Nil
case r: LogicalRDD =>
RDDScanExec(r.output, r.rdd, "ExistingRDD", r.outputPartitioning, r.outputOrdering) :: Nil
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala
index 6684376..31d1f34 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala
@@ -141,8 +141,10 @@ object OptimizeLocalShuffleReader {
}
def canUseLocalShuffleReader(plan: SparkPlan): Boolean = plan match {
- case s: ShuffleQueryStageExec => s.shuffle.canChangeNumPartitions
- case CustomShuffleReaderExec(s: ShuffleQueryStageExec, _, _) => s.shuffle.canChangeNumPartitions
+ case s: ShuffleQueryStageExec =>
+ s.shuffle.canChangeNumPartitions && s.mapStats.isDefined
+ case CustomShuffleReaderExec(s: ShuffleQueryStageExec, _, _) =>
+ s.shuffle.canChangeNumPartitions && s.mapStats.isDefined
case _ => false
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala
index b7da78c..24c7369 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala
@@ -83,7 +83,12 @@ trait ShuffleExchangeLike extends Exchange {
case class ShuffleExchangeExec(
override val outputPartitioning: Partitioning,
child: SparkPlan,
- canChangeNumPartitions: Boolean = true) extends ShuffleExchangeLike {
+ noUserSpecifiedNumPartition: Boolean = true) extends ShuffleExchangeLike {
+
+ // If users specify the num partitions via APIs like `repartition`, we shouldn't change it.
+ // For `SinglePartition`, it requires exactly one partition and we can't change it either.
+ def canChangeNumPartitions: Boolean =
+ noUserSpecifiedNumPartition && outputPartitioning != SinglePartition
private lazy val writeMetrics =
SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org