You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Wan Kun (Jira)" <ji...@apache.org> on 2023/09/20 06:55:00 UTC
[jira] [Updated] (SPARK-45230) Plan sorter for Aggregate after SMJ
[ https://issues.apache.org/jira/browse/SPARK-45230?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Wan Kun updated SPARK-45230:
----------------------------
Description:
If the aggregate operator comes after SMJ and the grouping expressions of aggregate operator contains all the join keys of the streamed side, we can add a sorter of the streamed side of the SMJ, so the aggregate can be convert to SortAggregate which will be faster than HashAggregate.
For example, with table t1(a, b, c) and t2(x, y, z):
{code}
SELECT a, b, sum(c)
FROM t1
JOIN t2
ON t1.b = t2.y
GROUP BY a, b
{code}
The optimized plan:
{code}
Scan(t1) Scan(t2)
| |
| |
Exchange 1 Exchange 2
\ /
\ /
\ /
SMJ (t1.b = t2.y)
|
|
Aggregate
{code}
Before this PR, spark EnsureReqirement will add Sorter(t1.b) to the left side of SMJ.
{code}
Scan(t1) Scan(t2)
| |
| |
Exchange 1 Exchange 2
\ /
Sort(t1.b) Sort(t2.y)
\ /
SMJ (t1.b = t2.y)
|
|
HashAggregate
{code}
If we add a Sort(t1.b, t1.a) to the left side of the SMJ, the following aggregate could be convert to SortAggregate, will be faster.
{code}
Scan(t1) Scan(t2)
| |
| |
Exchange 1 Exchange 2
\ /
Sort(t1.b, t1.a) Sort(t2.y)
\ /
SMJ (t1.b = t2.y)
|
|
SortAggregate
{code}
was:
If the aggregate operator comes after SMJ and the grouping expressions of aggregate operator contains all the join keys of the streamed side, we can add a sorter of the streamed side of the SMJ, so the aggregate can be convert to SortAggregate which will be faster than HashAggregate.
For example, with table t1(a, b, c) and t2(x, y, z):
{code}
SELECT a, b, sum(c)
FROM t1
JOIN t2
ON t1.b = t2.y
GROUP BY a, b
{code}
The optimized plan:
{code}
Scan(t1) Scan(t2)
| |
| |
Exchange 1 Exchange 2
\ /
\ /
\ /
SMJ (t1.b = t2.y)
|
|
Aggregate
{code}
Before this PR, spark EnsureReqirement will add Sorter(t1.b) to the left side of SMJ.
{code}
Scan(t1) Scan(t2)
| |
| |
Exchange 1 Exchange 2
\ /
Sort(t1.b) Sort(t2.y)
\ /
SMJ (t1.b = t2.y)
|
|
HashAggregate
{code}
If we add a Sort(t1.b, t1.a) to the left side of the SMJ, the following aggregate could be convert to SortAggregate, will be faster.
{code}
Scan(t1) Scan(t2)
| |
| |
Exchange 1 Exchange 2
\ /
Sort(t1.b, t1.a) Sort(t2.y)
\ /
SMJ (t1.b = t2.y)
|
|
SortAggregate
{code}
> Plan sorter for Aggregate after SMJ
> -----------------------------------
>
> Key: SPARK-45230
> URL: https://issues.apache.org/jira/browse/SPARK-45230
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 4.0.0
> Reporter: Wan Kun
> Priority: Major
>
> If the aggregate operator comes after SMJ and the grouping expressions of aggregate operator contains all the join keys of the streamed side, we can add a sorter of the streamed side of the SMJ, so the aggregate can be convert to SortAggregate which will be faster than HashAggregate.
> For example, with table t1(a, b, c) and t2(x, y, z):
> {code}
> SELECT a, b, sum(c)
> FROM t1
> JOIN t2
> ON t1.b = t2.y
> GROUP BY a, b
> {code}
> The optimized plan:
> {code}
> Scan(t1) Scan(t2)
> | |
> | |
> Exchange 1 Exchange 2
> \ /
> \ /
> \ /
> SMJ (t1.b = t2.y)
> |
> |
> Aggregate
> {code}
> Before this PR, spark EnsureReqirement will add Sorter(t1.b) to the left side of SMJ.
> {code}
> Scan(t1) Scan(t2)
> | |
> | |
> Exchange 1 Exchange 2
> \ /
> Sort(t1.b) Sort(t2.y)
> \ /
> SMJ (t1.b = t2.y)
> |
> |
> HashAggregate
> {code}
> If we add a Sort(t1.b, t1.a) to the left side of the SMJ, the following aggregate could be convert to SortAggregate, will be faster.
> {code}
> Scan(t1) Scan(t2)
> | |
> | |
> Exchange 1 Exchange 2
> \ /
> Sort(t1.b, t1.a) Sort(t2.y)
> \ /
> SMJ (t1.b = t2.y)
> |
> |
> SortAggregate
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org