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/09/10 11:42:02 UTC
spark git commit: [SPARK-25278][SQL] Avoid duplicated Exec nodes when
the same logical plan appears in the query
Repository: spark
Updated Branches:
refs/heads/master 6f6517837 -> 12e3e9f17
[SPARK-25278][SQL] Avoid duplicated Exec nodes when the same logical plan appears in the query
## What changes were proposed in this pull request?
In the Planner, we collect the placeholder which need to be substituted in the query execution plan and once we plan them, we substitute the placeholder with the effective plan.
In this second phase, we rely on the `==` comparison, ie. the `equals` method. This means that if two placeholder plans - which are different instances - have the same attributes (so that they are equal, according to the equal method) they are both substituted with their corresponding new physical plans. So, in such a situation, the first time we substitute both them with the first of the 2 new generated plan and the second time we substitute nothing.
This is usually of no harm for the execution of the query itself, as the 2 plans are identical. But since they are the same instance, now, the local variables are shared (which is unexpected). This causes issues for the metrics collected, as the same node is executed 2 times, so the metrics are accumulated 2 times, wrongly.
The PR proposes to use the `eq` method in checking which placeholder needs to be substituted,; thus in the previous situation, actually both the two different physical nodes which are created (one for each time the logical plan appears in the query plan) are used and the metrics are collected properly for each of them.
## How was this patch tested?
added UT
Closes #22284 from mgaido91/SPARK-25278.
Authored-by: Marco Gaido <ma...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/12e3e9f1
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/12e3e9f1
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/12e3e9f1
Branch: refs/heads/master
Commit: 12e3e9f17dca11a2cddf0fb99d72b4b97517fb56
Parents: 6f65178
Author: Marco Gaido <ma...@gmail.com>
Authored: Mon Sep 10 19:41:51 2018 +0800
Committer: Wenchen Fan <we...@databricks.com>
Committed: Mon Sep 10 19:41:51 2018 +0800
----------------------------------------------------------------------
.../spark/sql/catalyst/planning/QueryPlanner.scala | 2 +-
.../apache/spark/sql/execution/PlannerSuite.scala | 17 +++++++++++++++++
.../sql/execution/metric/SQLMetricsSuite.scala | 13 +++++++++++++
3 files changed, 31 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/12e3e9f1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala
index bc41dd0..6fa5203 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala
@@ -81,7 +81,7 @@ abstract class QueryPlanner[PhysicalPlan <: TreeNode[PhysicalPlan]] {
childPlans.map { childPlan =>
// Replace the placeholder by the child plan
candidateWithPlaceholders.transformUp {
- case p if p == placeholder => childPlan
+ case p if p.eq(placeholder) => childPlan
}
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/12e3e9f1/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 3db89ec..b10da6c 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
@@ -704,6 +704,23 @@ class PlannerSuite extends SharedSQLContext {
df.queryExecution.executedPlan.execute()
}
+ test("SPARK-25278: physical nodes should be different instances for same logical nodes") {
+ val range = Range(1, 1, 1, 1)
+ val df = Union(range, range)
+ val ranges = df.queryExecution.optimizedPlan.collect {
+ case r: Range => r
+ }
+ assert(ranges.length == 2)
+ // Ensure the two Range instances are equal according to their equal method
+ assert(ranges.head == ranges.last)
+ val execRanges = df.queryExecution.sparkPlan.collect {
+ case r: RangeExec => r
+ }
+ assert(execRanges.length == 2)
+ // Ensure the two RangeExec instances are different instances
+ assert(!execRanges.head.eq(execRanges.last))
+ }
+
test("SPARK-24556: always rewrite output partitioning in ReusedExchangeExec " +
"and InMemoryTableScanExec") {
def checkOutputPartitioningRewrite(
http://git-wip-us.apache.org/repos/asf/spark/blob/12e3e9f1/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
index a3a3f38..d45eb0c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
@@ -497,6 +497,19 @@ class SQLMetricsSuite extends SparkFunSuite with SQLMetricsTestUtils with Shared
}
}
+ test("SPARK-25278: output metrics are wrong for plans repeated in the query") {
+ val name = "demo_view"
+ withView(name) {
+ sql(s"CREATE OR REPLACE VIEW $name AS VALUES 1,2")
+ val view = spark.table(name)
+ val union = view.union(view)
+ testSparkPlanMetrics(union, 1, Map(
+ 0L -> ("Union" -> Map()),
+ 1L -> ("LocalTableScan" -> Map("number of output rows" -> 2L)),
+ 2L -> ("LocalTableScan" -> Map("number of output rows" -> 2L))))
+ }
+ }
+
test("writing data out metrics: parquet") {
testMetricsNonDynamicPartition("parquet", "t1")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org