You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "caoxuewen (JIRA)" <ji...@apache.org> on 2018/04/24 09:08:00 UTC
[jira] [Created] (SPARK-24066) Add a window exchange rule to
eliminate redundant physical plan SortExec
caoxuewen created SPARK-24066:
---------------------------------
Summary: Add a window exchange rule to eliminate redundant physical plan SortExec
Key: SPARK-24066
URL: https://issues.apache.org/jira/browse/SPARK-24066
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 2.4.0
Reporter: caoxuewen
Currently, when the order field of window function has a subset relationship, SparkSQL will randomly generate different physical plan.
Similar like:
case class DistinctAgg(a: Int, b: Float, c: Double, d: Int, e: String)
val df = spark.sparkContext.parallelize(
DistinctAgg(8, 2, 3, 4, "a") ::
DistinctAgg(9, 3, 4, 5, "b") ::
DistinctAgg(3, 4, 5, 6, "c") ::
DistinctAgg(3, 4, 5, 7, "c") ::
DistinctAgg(3, 4, 5, 8, "c") ::
DistinctAgg(3, 6, 6, 9, "d") ::
DistinctAgg(30, 40, 50, 60, "e") ::
DistinctAgg(41, 51, 61, 71, null) ::
DistinctAgg(42, 52, 62, 72, null) ::
DistinctAgg(43, 53, 63, 73, "k") ::Nil).toDF()
df.createOrReplaceTempView("distinctAgg")
select a, b, c,
avg(b) over(partition by a order by b) as sumIb,
sum(d) over(partition by a order by b, c) as sumId, d
from distinctAgg
The physics plan will produce different results randomly.
One: there is only one sort of physical plan
== Physical Plan ==
*(3) Project [a#181, b#182, c#183, sumId#210L, sumIb#209L, d#184]
+- Window [sum(cast(b#182 as bigint)) windowspecdefinition(a#181, b#182 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS sumIb#209L], [a#181], [b#182 ASC NULLS FIRST]
+- Window [sum(cast(d#184 as bigint)) windowspecdefinition(a#181, b#182 ASC NULLS FIRST, c#183 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS sumId#210L], [a#181], [b#182 ASC NULLS FIRST, c#183 ASC NULLS FIRST]
+- *(2) Sort [a#181 ASC NULLS FIRST, b#182 ASC NULLS FIRST, c#183 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(a#181, 5)
+- *(1) Project [a#181, b#182, c#183, d#184]
+- *(1) SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).a AS a#181, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).b AS b#182, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).c AS c#183, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).d AS d#184, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).e, true, false) AS e#185]
+- Scan ExternalRDDScan[obj#180]
Another one: there is two sort of physical plans
== Physical Plan ==
*(4) Project [a#181, b#182, c#183, sumId#210L, sumIb#209L, d#184]
+- Window [sum(cast(d#184 as bigint)) windowspecdefinition(a#181, b#182 ASC NULLS FIRST, c#183 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS sumId#210L], [a#181], [b#182 ASC NULLS FIRST, c#183 ASC NULLS FIRST]
+- *(3) Sort [a#181 ASC NULLS FIRST, b#182 ASC NULLS FIRST, c#183 ASC NULLS FIRST], false, 0
+- Window [sum(cast(b#182 as bigint)) windowspecdefinition(a#181, b#182 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS sumIb#209L], [a#181], [b#182 ASC NULLS FIRST]
+- *(2) Sort [a#181 ASC NULLS FIRST, b#182 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(a#181, 5)
+- *(1) Project [a#181, b#182, c#183, d#184]
+- *(1) SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).a AS a#181, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).b AS b#182, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).c AS c#183, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).d AS d#184, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$DistinctAgg, true]).e, true, false) AS e#185]
+- Scan ExternalRDDScan[obj#180]
this PR add an exchange rule to ensure that no redundant physical plan SortExec is generated.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org