[
https://issues.apache.org/jira/browse/SPARK-42935?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jeff Min updated SPARK-42935:
-----------------------------
Labels: pull-request-available (was: )
> Optimze shuffle for union spark plan
> ------------------------------------
>
> Key: SPARK-42935
> URL: https://issues.apache.org/jira/browse/SPARK-42935
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 3.5.0
> Reporter: Jeff Min
> Priority: Major
> Labels: pull-request-available
> Fix For: 3.5.0
>
>
> Union plan does not take full advantage of children plan output partitionings
> when output partitoning can't match parent plan's required distribution. For
> example, Table1 and table2 are all bucketed table with bucket column id and
> bucket number 100. We will do row_number window function after union the two
> tables.
> {code:sql}
> create table table1 (id int, name string) using csv CLUSTERED BY (id) INTO
> 100 BUCKETS;
> insert into table1 values(1, "s1");
> insert into table1 values(2, "s2");
>
> create table table2 (id int, name string) using csv CLUSTERED BY (id) INTO
> 100 BUCKETS;
> insert into table2 values(1, "s3");
>
> set spark.sql.shuffle.partitions=100;
> explain select *, row_number() over(partition by id order by name desc)
> id_row_number from (select * from table1 union all select * from
> table2);{code}
> The physical plan is
> {code:bash}
> AdaptiveSparkPlan isFinalPlan=false
> +- Window row_number() windowspecdefinition(id#35, name#36 DESC NULLS LAST,
> specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS
> id_row_number#28, id#35, name#36 DESC NULLS LAST
> +- Sort id#35 ASC NULLS FIRST, name#36 DESC NULLS LAST, false, 0
> +- Exchange hashpartitioning(id#35, 100), ENSURE_REQUIREMENTS,
> [plan_id=88]
> +- Union
> :- FileScan csv spark_catalog.default.table1id#35,name#36
> +- FileScan csv spark_catalog.default.table2id#37,name#38 {code}
>
> Although the two tables are bucketed by id column, there's still a exchange
> plan after union.The reason is that union plan's output partitioning is null.
> We can indroduce a new idea to optimize exchange plan:
> # First introduce a new RDD, it consists of parent rdds that has the same
> partition size. The ith parttition corresponds to ith partition of each
> parent rdd.
> # Then push the required distribution to union plan's children. If any child
> output partitioning matches the required distribution , we can reduce this
> child shuffle operation.
> After doing these, the physical plan does not contain exchange shuffle plan
> {code:bash}
> AdaptiveSparkPlan isFinalPlan=false
> +- Window row_number() windowspecdefinition(id#7, name#8 DESC NULLS LAST,
> specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS
> id_row_number#0, id#7, name#8 DESC NULLS LAST
> +- Sort id#7 ASC NULLS FIRST, name#8 DESC NULLS LAST, false, 0
> +- UnionZip ClusteredDistribution(ArrayBuffer(id#7),false,None),
> ClusteredDistribution(ArrayBuffer(id#9),false,None), hashpartitioning(id#7,
> 200)
> :- FileScan csv spark_catalog.default.table1id#7,name#8
> +- FileScan csv spark_catalog.default.table2id#9,name#10 {code}
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
