[SPARK-36638][SQL] Generalize OptimizeSkewedJoin by zhengruifeng · Pull Request #33893 · apache/spark

zhengruifeng · 2021-09-01T12:19:25Z

What changes were proposed in this pull request?

This PR aims to generalize OptimizeSkewedJoin to support all patterns that can be handled by current split-duplicate strategy:

1, find the splittable shuffle query stages by the semantics of internal nodes;

2, for each splittable shuffle query stage, check whether skew partitions exists, if true, split them into specs;

3, handle Combinatorial Explosion: for each skew partition, check whether the combination number is too large, if so, re-split the stages to keep a reasonable number of combinations. For example, for partition 0, stage A/B/C are split into 100/100/100 specs, respectively. Then there are 1M combinations, which is too large, and will cause performance regression.

4, attach new specs to shuffle query stages;

Why are the changes needed?

to Generalize OptimizeSkewedJoin

Does this PR introduce any user-facing change?

one additional config added

How was this patch tested?

existing testsuites, added testsuites, some cases on our productive system

zhengruifeng · 2021-09-01T12:22:16Z

added test("General Skew Join: 3-table join")

== Physical Plan ==
AdaptiveSparkPlan (52)
+- == Final Plan ==
   CollectLimit (33)
   +- * HashAggregate (32)
      +- AQEShuffleRead (31)
         +- ShuffleQueryStage (30)
            +- Exchange (29)
               +- * HashAggregate (28)
                  +- * Project (27)
                     +- * SortMergeJoin(skew=true) LeftOuter (26)
                        :- * Project (15)
                        :  +- * SortMergeJoin(skew=true) Inner (14)
                        :     :- Window(skew=true) (7)
                        :     :  +- * Sort (6)
                        :     :     +- AQEShuffleRead (5)
                        :     :        +- ShuffleQueryStage (4)
                        :     :           +- Exchange (3)
                        :     :              +- * Project (2)
                        :     :                 +- * Range (1)
                        :     +- * Sort (13)
                        :        +- AQEShuffleRead (12)
                        :           +- ShuffleQueryStage (11)
                        :              +- Exchange (10)
                        :                 +- * Project (9)
                        :                    +- * Range (8)
                        +- * Sort (25)
                           +- * HashAggregate(skew=true) (24)
                              +- AQEShuffleRead (23)
                                 +- ShuffleQueryStage (22)
                                    +- Exchange (21)
                                       +- * HashAggregate (20)
                                          +- ShuffleQueryStage (19)
                                             +- Exchange (18)
                                                +- * Project (17)
                                                   +- * Range (16)

zhengruifeng · 2021-09-01T12:24:38Z

added test("General Skew Join: 3-table join UNION 2-table join")

== Physical Plan ==
AdaptiveSparkPlan (68)
+- == Final Plan ==
   CollectLimit (43)
   +- * HashAggregate (42)
      +- AQEShuffleRead (41)
         +- ShuffleQueryStage (40)
            +- Exchange (39)
               +- * HashAggregate (38)
                  +- Union (37)
                     :- * Project (24)
                     :  +- * SortMergeJoin(skew=true) LeftOuter (23)
                     :     :- * Project (14)
                     :     :  +- * SortMergeJoin(skew=true) Inner (13)
                     :     :     :- * Sort (6)
                     :     :     :  +- AQEShuffleRead (5)
                     :     :     :     +- ShuffleQueryStage (4)
                     :     :     :        +- Exchange (3)
                     :     :     :           +- * Project (2)
                     :     :     :              +- * Range (1)
                     :     :     +- * Sort (12)
                     :     :        +- AQEShuffleRead (11)
                     :     :           +- ShuffleQueryStage (10)
                     :     :              +- Exchange (9)
                     :     :                 +- * Project (8)
                     :     :                    +- * Range (7)
                     :     +- * Sort (22)
                     :        +- * HashAggregate(skew=true) (21)
                     :           +- AQEShuffleRead (20)
                     :              +- ShuffleQueryStage (19)
                     :                 +- Exchange (18)
                     :                    +- * HashAggregate (17)
                     :                       +- * Project (16)
                     :                          +- * Range (15)
                     +- * Project (36)
                        +- * SortMergeJoin(skew=true) LeftOuter (35)
                           :- * Sort (28)
                           :  +- AQEShuffleRead (27)
                           :     +- ShuffleQueryStage (26)
                           :        +- ReusedExchange (25)
                           +- * Sort (34)
                              +- AQEShuffleRead (33)
                                 +- ShuffleQueryStage (32)
                                    +- Exchange (31)
                                       +- * Project (30)
                                          +- * Range (29)

zhengruifeng · 2021-09-01T12:26:01Z

added test("General Skew Join: 5-table join")

== Physical Plan ==
AdaptiveSparkPlan (68)
+- == Final Plan ==
   CollectLimit (45)
   +- * HashAggregate (44)
      +- AQEShuffleRead (43)
         +- ShuffleQueryStage (42)
            +- Exchange (41)
               +- * HashAggregate (40)
                  +- * Project (39)
                     +- * SortMergeJoin(skew=true) Inner (38)
                        :- * Project (22)
                        :  +- * SortMergeJoin(skew=true) Cross (21)
                        :     :- * Project (14)
                        :     :  +- * SortMergeJoin(skew=true) LeftOuter (13)
                        :     :     :- * Sort (6)
                        :     :     :  +- AQEShuffleRead (5)
                        :     :     :     +- ShuffleQueryStage (4)
                        :     :     :        +- Exchange (3)
                        :     :     :           +- * Project (2)
                        :     :     :              +- * Range (1)
                        :     :     +- * Sort (12)
                        :     :        +- AQEShuffleRead (11)
                        :     :           +- ShuffleQueryStage (10)
                        :     :              +- Exchange (9)
                        :     :                 +- * Project (8)
                        :     :                    +- * Range (7)
                        :     +- * Sort (20)
                        :        +- AQEShuffleRead (19)
                        :           +- ShuffleQueryStage (18)
                        :              +- Exchange (17)
                        :                 +- * Project (16)
                        :                    +- * Range (15)
                        +- * Sort (37)
                           +- * Project (36)
                              +- * ShuffledHashJoin(skew=true) Inner BuildLeft (35)
                                 :- * HashAggregate(skew=true) (29)
                                 :  +- AQEShuffleRead (28)
                                 :     +- ShuffleQueryStage (27)
                                 :        +- Exchange (26)
                                 :           +- * HashAggregate (25)
                                 :              +- * Project (24)
                                 :                 +- * Range (23)
                                 +- AQEShuffleRead (34)
                                    +- ShuffleQueryStage (33)
                                       +- Exchange (32)
                                          +- * Project (31)
                                             +- * Range (30)

zhengruifeng · 2021-09-02T06:37:13Z

friendly ping @cloud-fan

zhengruifeng · 2021-09-06T02:23:26Z

friendly ping @JkSelf @cloud-fan @yaooqinn @ulysses-you . Could you please take a look in your spare time? Thanks!

zhengruifeng · 2021-09-26T06:27:29Z

When developing this method, I used some tests like #34108 to check correctness. It should be helpful for reviewing.

zhengruifeng · 2022-02-14T08:51:39Z

this skew test case newly added in #34908 can be optimized by this PR without extra shuffle:

Master this PR

zhengruifeng · 2022-03-23T00:50:53Z

retest this please

sigmod · 2022-03-23T17:48:11Z

cc @maryannxue

drop last str arg format sql in ut support both ShuffleQueryStageExec and BroadcastQueryStageExec as leaves nit reslove conflicts narrow valid operator whitelist move agg stringArgs to subclasses add doc && resolve conflicts del sample node resolve conflicts nit

nit

github-actions · 2022-09-30T00:36:25Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

github-actions Bot added the SQL label Sep 1, 2021

advancedxy reviewed Sep 3, 2021

View reviewed changes

Comment thread sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala Outdated

zhengruifeng force-pushed the generalize_skew_join branch 2 times, most recently from 545bf9b to cd3c449 Compare September 13, 2021 12:38

zhengruifeng force-pushed the generalize_skew_join branch from cd3c449 to 5e64c4e Compare September 16, 2021 02:39

zhengruifeng force-pushed the generalize_skew_join branch from 6b434b0 to c1d3e48 Compare September 26, 2021 06:19

zhengruifeng mentioned this pull request Sep 28, 2021

[SPARK-36638][SQL][TEST] Generalize OptimizeSkewedJoin - correctness #34108

Closed

zhengruifeng force-pushed the generalize_skew_join branch 2 times, most recently from dc3b6ea to d7c6678 Compare November 19, 2021 08:40

zhengruifeng mentioned this pull request Dec 15, 2021

[SPARK-37328][SQL] Fix bug that OptimizeSkewedJoin may not work after it was moved from queryStageOptimizerRules to queryStagePreparationRules. #34602

Closed

zhengruifeng force-pushed the generalize_skew_join branch from d7c6678 to 3cdd96e Compare December 15, 2021 07:40

zhengruifeng force-pushed the generalize_skew_join branch from 3d4d7fa to c8bfb0c Compare December 22, 2021 10:19

zhengruifeng force-pushed the generalize_skew_join branch 2 times, most recently from 36a5b6a to 1f88310 Compare January 25, 2022 06:28

zhengruifeng force-pushed the generalize_skew_join branch from 1f88310 to 65bb496 Compare January 27, 2022 14:29

zhengruifeng force-pushed the generalize_skew_join branch from 65bb496 to 169c2ca Compare February 14, 2022 08:19

zhengruifeng commented Feb 14, 2022

View reviewed changes

zhengruifeng force-pushed the generalize_skew_join branch from 169c2ca to 2b3a2fd Compare March 14, 2022 11:06

zhengruifeng force-pushed the generalize_skew_join branch from 2b3a2fd to 35d2212 Compare March 22, 2022 15:57

zhengruifeng force-pushed the generalize_skew_join branch from 35d2212 to 6305685 Compare March 23, 2022 01:23

apache deleted a comment from SparkQA Mar 23, 2022

zhengruifeng and others added 12 commits May 4, 2022 22:27

init

02c2969

drop last str arg format sql in ut support both ShuffleQueryStageExec and BroadcastQueryStageExec as leaves nit reslove conflicts narrow valid operator whitelist move agg stringArgs to subclasses add doc && resolve conflicts del sample node resolve conflicts nit

support intermediate skew join

b8f0141

simplify collecting nodes

3a60275

rebase

2273484

del maxJoins

fbdb8e1

nit

fcb2e1e

nit

fix conflict

7336062

nit

1c580a7

fix SPARK-37652 case

b328f2b

resolve conflict

bb80853

add ut

cb493c4

rebase, update version, fix ut

0d250b0

zhengruifeng force-pushed the generalize_skew_join branch from 6305685 to 0d250b0 Compare May 4, 2022 15:07

github-actions Bot added the Stale label Sep 30, 2022

github-actions Bot closed this Oct 1, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-36638][SQL] Generalize OptimizeSkewedJoin#33893

[SPARK-36638][SQL] Generalize OptimizeSkewedJoin#33893
zhengruifeng wants to merge 12 commits into
apache:masterfrom
zhengruifeng:generalize_skew_join

zhengruifeng commented Sep 1, 2021 •

edited

Loading

Uh oh!

zhengruifeng commented Sep 1, 2021

Uh oh!

zhengruifeng commented Sep 1, 2021

Uh oh!

zhengruifeng commented Sep 1, 2021

Uh oh!

zhengruifeng commented Sep 2, 2021

Uh oh!

Uh oh!

zhengruifeng commented Sep 6, 2021

Uh oh!

zhengruifeng commented Sep 26, 2021

Uh oh!

zhengruifeng Feb 14, 2022 •

edited

Loading

Uh oh!

zhengruifeng commented Mar 23, 2022

Uh oh!

sigmod commented Mar 23, 2022

Uh oh!

github-actions Bot commented Sep 30, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

zhengruifeng commented Sep 1, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

zhengruifeng commented Sep 1, 2021

Uh oh!

zhengruifeng commented Sep 1, 2021

Uh oh!

zhengruifeng commented Sep 1, 2021

Uh oh!

zhengruifeng commented Sep 2, 2021

Uh oh!

Uh oh!

zhengruifeng commented Sep 6, 2021

Uh oh!

zhengruifeng commented Sep 26, 2021

Uh oh!

zhengruifeng Feb 14, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhengruifeng commented Mar 23, 2022

Uh oh!

sigmod commented Mar 23, 2022

Uh oh!

github-actions Bot commented Sep 30, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

zhengruifeng commented Sep 1, 2021 •

edited

Loading

zhengruifeng Feb 14, 2022 •

edited

Loading