You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Apache Spark (JIRA)" <ji...@apache.org> on 2018/04/24 09:17:00 UTC
[jira] [Commented] (SPARK-24066) Add a window exchange rule to
eliminate redundant physical plan SortExec
[ https://issues.apache.org/jira/browse/SPARK-24066?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16449512#comment-16449512 ]
Apache Spark commented on SPARK-24066:
--------------------------------------
User 'heary-cao' has created a pull request for this issue:
https://github.com/apache/spark/pull/21139
> 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
> Priority: Major
>
> 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