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 2018/06/19 17:52:55 UTC
spark git commit: [SPARK-24556][SQL] Always rewrite output
partitioning in ReusedExchangeExec and InMemoryTableScanExec
Repository: spark
Updated Branches:
refs/heads/master a78a90464 -> 9dbe53eb6
[SPARK-24556][SQL] Always rewrite output partitioning in ReusedExchangeExec and InMemoryTableScanExec
## What changes were proposed in this pull request?
Currently, ReusedExchange and InMemoryTableScanExec only rewrite output partitioning if child's partitioning is HashPartitioning and do nothing for other partitioning, e.g., RangePartitioning. We should always rewrite it, otherwise, unnecessary shuffle could be introduced like https://issues.apache.org/jira/browse/SPARK-24556.
## How was this patch tested?
Add new tests.
Author: yucai <yy...@ebay.com>
Closes #21564 from yucai/SPARK-24556.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/9dbe53eb
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/9dbe53eb
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/9dbe53eb
Branch: refs/heads/master
Commit: 9dbe53eb6bb5916d28000f2c0d646cf23094ac11
Parents: a78a904
Author: yucai <yy...@ebay.com>
Authored: Tue Jun 19 10:52:51 2018 -0700
Committer: Wenchen Fan <we...@databricks.com>
Committed: Tue Jun 19 10:52:51 2018 -0700
----------------------------------------------------------------------
.../columnar/InMemoryTableScanExec.scala | 6 +-
.../spark/sql/execution/exchange/Exchange.scala | 4 +-
.../spark/sql/execution/PlannerSuite.scala | 64 +++++++++++++++++++-
3 files changed, 67 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/9dbe53eb/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
index 0b4dd76..997cf92 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning}
+import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.execution.{ColumnarBatchScan, LeafExecNode, SparkPlan, WholeStageCodegenExec}
import org.apache.spark.sql.execution.vectorized._
import org.apache.spark.sql.types._
@@ -169,8 +169,8 @@ case class InMemoryTableScanExec(
// But the cached version could alias output, so we need to replace output.
override def outputPartitioning: Partitioning = {
relation.cachedPlan.outputPartitioning match {
- case h: HashPartitioning => updateAttribute(h).asInstanceOf[HashPartitioning]
- case _ => relation.cachedPlan.outputPartitioning
+ case e: Expression => updateAttribute(e).asInstanceOf[Partitioning]
+ case other => other
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/9dbe53eb/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala
index 09f79a2..1a5b759 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala
@@ -24,7 +24,7 @@ import org.apache.spark.broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, Expression, SortOrder}
-import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning}
+import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, UnaryExecNode}
import org.apache.spark.sql.internal.SQLConf
@@ -70,7 +70,7 @@ case class ReusedExchangeExec(override val output: Seq[Attribute], child: Exchan
}
override def outputPartitioning: Partitioning = child.outputPartitioning match {
- case h: HashPartitioning => h.copy(expressions = h.expressions.map(updateAttr))
+ case e: Expression => updateAttr(e).asInstanceOf[Partitioning]
case other => other
}
http://git-wip-us.apache.org/repos/asf/spark/blob/9dbe53eb/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
index 37d4687..d254345 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
@@ -18,13 +18,13 @@
package org.apache.spark.sql.execution
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{execution, Row}
+import org.apache.spark.sql.{execution, DataFrame, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Range, Repartition, Sort, Union}
import org.apache.spark.sql.catalyst.plans.physical._
-import org.apache.spark.sql.execution.columnar.InMemoryRelation
+import org.apache.spark.sql.execution.columnar.{InMemoryRelation, InMemoryTableScanExec}
import org.apache.spark.sql.execution.exchange.{EnsureRequirements, ReusedExchangeExec, ReuseExchange, ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.functions._
@@ -703,6 +703,66 @@ class PlannerSuite extends SharedSQLContext {
Range(1, 2, 1, 1)))
df.queryExecution.executedPlan.execute()
}
+
+ test("SPARK-24556: always rewrite output partitioning in ReusedExchangeExec " +
+ "and InMemoryTableScanExec") {
+ def checkOutputPartitioningRewrite(
+ plans: Seq[SparkPlan],
+ expectedPartitioningClass: Class[_]): Unit = {
+ assert(plans.size == 1)
+ val plan = plans.head
+ val partitioning = plan.outputPartitioning
+ assert(partitioning.getClass == expectedPartitioningClass)
+ val partitionedAttrs = partitioning.asInstanceOf[Expression].references
+ assert(partitionedAttrs.subsetOf(plan.outputSet))
+ }
+
+ def checkReusedExchangeOutputPartitioningRewrite(
+ df: DataFrame,
+ expectedPartitioningClass: Class[_]): Unit = {
+ val reusedExchange = df.queryExecution.executedPlan.collect {
+ case r: ReusedExchangeExec => r
+ }
+ checkOutputPartitioningRewrite(reusedExchange, expectedPartitioningClass)
+ }
+
+ def checkInMemoryTableScanOutputPartitioningRewrite(
+ df: DataFrame,
+ expectedPartitioningClass: Class[_]): Unit = {
+ val inMemoryScan = df.queryExecution.executedPlan.collect {
+ case m: InMemoryTableScanExec => m
+ }
+ checkOutputPartitioningRewrite(inMemoryScan, expectedPartitioningClass)
+ }
+
+ // ReusedExchange is HashPartitioning
+ val df1 = Seq(1 -> "a").toDF("i", "j").repartition($"i")
+ val df2 = Seq(1 -> "a").toDF("i", "j").repartition($"i")
+ checkReusedExchangeOutputPartitioningRewrite(df1.union(df2), classOf[HashPartitioning])
+
+ // ReusedExchange is RangePartitioning
+ val df3 = Seq(1 -> "a").toDF("i", "j").orderBy($"i")
+ val df4 = Seq(1 -> "a").toDF("i", "j").orderBy($"i")
+ checkReusedExchangeOutputPartitioningRewrite(df3.union(df4), classOf[RangePartitioning])
+
+ // InMemoryTableScan is HashPartitioning
+ Seq(1 -> "a").toDF("i", "j").repartition($"i").persist()
+ checkInMemoryTableScanOutputPartitioningRewrite(
+ Seq(1 -> "a").toDF("i", "j").repartition($"i"), classOf[HashPartitioning])
+
+ // InMemoryTableScan is RangePartitioning
+ spark.range(1, 100, 1, 10).toDF().persist()
+ checkInMemoryTableScanOutputPartitioningRewrite(
+ spark.range(1, 100, 1, 10).toDF(), classOf[RangePartitioning])
+
+ // InMemoryTableScan is PartitioningCollection
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ Seq(1 -> "a").toDF("i", "j").join(Seq(1 -> "a").toDF("m", "n"), $"i" === $"m").persist()
+ checkInMemoryTableScanOutputPartitioningRewrite(
+ Seq(1 -> "a").toDF("i", "j").join(Seq(1 -> "a").toDF("m", "n"), $"i" === $"m"),
+ classOf[PartitioningCollection])
+ }
+ }
}
// Used for unit-testing EnsureRequirements
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org