[SPARK-33832][SQL] Support optimize skewed join even if introduce extra shuffle #32816

ulysses-you · 2021-06-08T09:04:44Z

What changes were proposed in this pull request?

move the rule OptimizeSkewedJoin from stage optimization phase to stage preparation phase.
run the rule EnsureRequirements one more time after the OptimizeSkewedJoin rule in the stage preparation phase.
add SkewJoinAwareCost to support estimate skewed join cost
add new config to decide if force optimize skewed join
in OptimizeSkewedJoin, we generate 2 physical plans, one with skew join optimization and one without. Then we use the cost evaluator w.r.t. the force-skew-join flag and pick the plan with lower cost.

Why are the changes needed?

In general, skewed join has more impact on performance than once more shuffle. It makes sense to force optimize skewed join even if introduce extra shuffle.

A common case:

HashAggregate
  SortMergJoin
    Sort
      Exchange
    Sort
      Exchange

and after this PR, the plan looks like:

HashAggregate
  Exchange
    SortMergJoin (isSkew=true)
      Sort
        Exchange
      Sort
        Exchange

Note that, the new introduced shuffle also can be optimized by AQE.

Does this PR introduce any user-facing change?

Yes, a new config.

How was this patch tested?

Add new test
pass exists test SPARK-30524: Do not optimize skew join if introduce additional shuffle
pass exists test SPARK-33551: Do not use custom shuffle reader for repartition

SparkQA · 2021-06-08T10:13:33Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43989/

SparkQA · 2021-06-08T10:21:44Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43989/

SparkQA · 2021-06-08T10:24:23Z

Test build #139471 has finished for PR 32816 at commit b02bc0f.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-08T11:09:35Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43994/

SparkQA · 2021-06-08T11:44:02Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43994/

SparkQA · 2021-06-08T13:03:54Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44003/

SparkQA · 2021-06-08T13:36:07Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44003/

SparkQA · 2021-06-08T13:44:03Z

Test build #139466 has finished for PR 32816 at commit 368d1e0.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class SkewJoinAwareCost(numShuffles: Int, numSkewJoins: Int) extends Cost

cloud-fan · 2021-06-08T14:04:50Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala

@@ -111,6 +115,19 @@ case class AdaptiveSparkPlanExec(
    CollapseCodegenStages()
  )

+  // OptimizeSkewedJoin has moved into preparation rules, so we should make
+  // finalPreparationStageRules same as finalStageOptimizerRules
+  private def finalPreparationStageRules: Seq[Rule[SparkPlan]] = {


is it mainly to exclude OptimizeSkewedJoin in the final stage?

yea, currently it's only for OptimizeSkewedJoin

SparkQA · 2021-06-08T15:01:34Z

Test build #139480 has finished for PR 32816 at commit ccc45ab.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-09T04:34:39Z

Test build #139532 has finished for PR 32816 at commit 4236958.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-09T05:31:48Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44056/

SparkQA · 2021-06-09T06:06:27Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44056/

SparkQA · 2021-06-09T07:19:07Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44066/

SparkQA · 2021-06-09T07:56:12Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44066/

SparkQA · 2021-06-09T09:58:48Z

Test build #139541 has finished for PR 32816 at commit e5526c5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-10T04:21:32Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44135/

SparkQA · 2021-06-10T04:57:48Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44135/

SparkQA · 2021-06-10T05:37:01Z

Test build #139608 has finished for PR 32816 at commit a80bb5c.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public final class CompositeReadLimit implements ReadLimit
public final class ReadMinRows implements ReadLimit
trait InvokeLike extends Expression with NonSQLExpression with ImplicitCastInputTypes
case class LateralSubquery(
case class LateralJoin(
case class CommandResultExec(
class RocksDBFileManager(
sealed trait SchemaReader
class SchemaV1Reader extends SchemaReader
class SchemaV2Reader extends SchemaReader
trait SchemaWriter
class SchemaV1Writer extends SchemaWriter
class SchemaV2Writer extends SchemaWriter
case class CommandResult(

SparkQA · 2021-06-10T07:33:13Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44148/

SparkQA · 2021-06-10T08:08:20Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44148/

SparkQA · 2021-06-10T11:19:27Z

Test build #139621 has finished for PR 32816 at commit 094ee0c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2021-07-01T17:53:12Z

I think using the CostEvaluator to accept extra shuffles introduced by skew join handling is a good idea. However, the current framework is too simple: we just give up the entire re-optimization if the cost becomes higher, while we should try to interact with the re-optimization process to get the plan with the lowest cost. For example, if skew join handling adds extra shuffles, we should only give up skew join handling, instead of the entire re-optimization.

My idea is, the re-optimization can produce 2 result plans: one with skew join handling and one without. Then we compare the cost: if the plan with skew join handling has a higher cost and force-apply is not enabled, we pick the other plan. Otherwise, we pick the plan with skew join handling. Then we don't need to change the CostEvaluator.

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala

SparkQA · 2021-09-09T03:03:22Z

Test build #143101 has finished for PR 32816 at commit ca63321.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

ulysses-you · 2021-09-09T03:10:04Z

retest this please

SparkQA · 2021-09-09T03:37:58Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47604/

SparkQA · 2021-09-09T04:22:15Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47605/

SparkQA · 2021-09-09T04:31:01Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47607/

SparkQA · 2021-09-09T04:32:35Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47605/

SparkQA · 2021-09-09T04:36:19Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47604/

SparkQA · 2021-09-09T04:40:28Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47607/

SparkQA · 2021-09-09T08:25:26Z

Test build #143103 has finished for PR 32816 at commit ca63321.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

maryannxue · 2021-09-10T14:44:19Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala


+  private def ensureDistributionAndOrdering(
+      originChildren: Seq[SparkPlan],


nit: originalChildren

Still, i'd prefer using children here and newChildren below.

maryannxue · 2021-09-10T14:48:38Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala

@@ -35,15 +36,48 @@ case class SimpleCost(value: Long) extends Cost {
 }

 /**
- * A simple implementation of [[CostEvaluator]], which counts the number of
- * [[ShuffleExchangeLike]] nodes in the plan.
+ * A skew join aware implementation of [[Cost]], which consider shuffle number and skew join number


Can we add more description on how the cost is calculated in the presence of skew joins? What happens if there's two or more extra shuffles by adding a skew join optimization?

yea, add more comment to show how we pick the cost with skew join and shuffle.

SparkQA · 2021-09-11T11:21:44Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47668/

SparkQA · 2021-09-11T11:30:35Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47668/

SparkQA · 2021-09-11T18:25:33Z

Test build #143164 has finished for PR 32816 at commit f5e4b91.

This patch fails from timeout after a configured wait of 500m.
This patch merges cleanly.
This patch adds no public classes.

ulysses-you · 2021-09-12T01:31:33Z

retest this please

SparkQA · 2021-09-12T02:23:50Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47678/

SparkQA · 2021-09-12T02:32:47Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47678/

SparkQA · 2021-09-12T06:20:33Z

Test build #143175 has finished for PR 32816 at commit f5e4b91.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2021-09-13T09:21:23Z

thanks, merging to master!

ulysses-you · 2021-09-13T11:15:27Z

thank you all !

### What changes were proposed in this pull request? This is a followup of #32816 to simplify and improve the code: 1. Add a `SkewJoinChildWrapper` to wrap the skew join children, so that `EnsureRequirements` rule will skip them and save time 2. Remove `SkewJoinAwareCost` and keep using `SimpleCost`. We can put `numSkews` in the first 32 bits. ### Why are the changes needed? code simplification and improvement ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? existing tests Closes #34080 from cloud-fan/follow. Authored-by: Wenchen Fan <wenchen@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

… with multiple joins ### What changes were proposed in this pull request? In #32816 , we moved the rule `OptimizeSkewedJoin` from `queryStageOptimizerRules` to `queryStagePreparationRules`. This means that the input plan of `OptimizeSkewedJoin` will be the entire query plan, while before the input plan was a shuffle stage which is usually a small part of the query plan. However, to simplify the implementation, `OptimizeSkewedJoin` has a check that it will be skipped if the input plan has more than 2 shuffle stages. This is obviously problematic now, as the input plan is the entire query plan and is very likely to have more than 2 shuffles. This PR proposes to remove the check from `OptimizeSkewedJoin` completely, as it's no longer needed. 1. We can and should optimize all the skewed joins in the query plan. 2. If it introduces extra shuffles, we can return the original input plan, or accept it if the force-apply config is true. ### Why are the changes needed? Fix a performance regression in the master branch (not released yet) ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? a new test Closes #34974 from cloud-fan/aqe. Authored-by: Wenchen Fan <wenchen@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

github-actions bot added the SQL label Jun 8, 2021

cloud-fan reviewed Jun 8, 2021

View reviewed changes

cloud-fan mentioned this pull request Jun 17, 2021

[SPARK-35794][SQL] Allow custom plugin for AQE cost evaluator #32944

Closed

cloud-fan reviewed Jul 1, 2021

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala Outdated Show resolved Hide resolved

ulysses-you added 4 commits July 2, 2021 09:51

Support optimize skew join even if introduce extra shuffle

49e8bd9

EnsureRequirements

db77ddd

make a new rules

a63cd72

fix local reader number

59a5e4a

pull out shuffle origin

ca63321

maryannxue reviewed Sep 10, 2021

View reviewed changes

address comment

f5e4b91

cloud-fan approved these changes Sep 13, 2021

View reviewed changes

cloud-fan closed this in 4a6b2b9 Sep 13, 2021

ulysses-you deleted the support-extra-shuffle branch September 13, 2021 11:13

cloud-fan mentioned this pull request Sep 23, 2021

[SPARK-33832][SQL] Force skew join code simplification and improvement #34080

Closed

cloud-fan mentioned this pull request Dec 21, 2021

[SPARK-37328][SQL] OptimizeSkewedJoin should work for complex queries with multiple joins #34974

Closed

FireFoxAhri mentioned this pull request Dec 3, 2022

[Bug] kyuubi spark extension throws java.lang.IllegalStateException: UnspecifiedDistribution does not have default partitioning. apache/kyuubi#3893

Closed

5 tasks


		private def ensureDistributionAndOrdering(
		originChildren: Seq[SparkPlan],

[SPARK-33832][SQL] Support optimize skewed join even if introduce extra shuffle #32816

[SPARK-33832][SQL] Support optimize skewed join even if introduce extra shuffle #32816

Conversation

ulysses-you commented Jun 8, 2021 • edited Loading

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?

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

SparkQA commented Jun 8, 2021

cloud-fan Jun 8, 2021

Choose a reason for hiding this comment

ulysses-you Jun 8, 2021

Choose a reason for hiding this comment

SparkQA commented Jun 8, 2021

SparkQA commented Jun 9, 2021

SparkQA commented Jun 9, 2021

SparkQA commented Jun 9, 2021

SparkQA commented Jun 9, 2021

SparkQA commented Jun 9, 2021

SparkQA commented Jun 9, 2021

SparkQA commented Jun 10, 2021

SparkQA commented Jun 10, 2021

SparkQA commented Jun 10, 2021

SparkQA commented Jun 10, 2021

SparkQA commented Jun 10, 2021

SparkQA commented Jun 10, 2021

cloud-fan commented Jul 1, 2021

SparkQA commented Sep 9, 2021

ulysses-you commented Sep 9, 2021

SparkQA commented Sep 9, 2021

SparkQA commented Sep 9, 2021

SparkQA commented Sep 9, 2021

SparkQA commented Sep 9, 2021

SparkQA commented Sep 9, 2021

SparkQA commented Sep 9, 2021

SparkQA commented Sep 9, 2021

maryannxue Sep 10, 2021

Choose a reason for hiding this comment

maryannxue Sep 10, 2021

Choose a reason for hiding this comment

maryannxue Sep 10, 2021

Choose a reason for hiding this comment

ulysses-you Sep 11, 2021

Choose a reason for hiding this comment

SparkQA commented Sep 11, 2021

SparkQA commented Sep 11, 2021

SparkQA commented Sep 11, 2021

ulysses-you commented Sep 12, 2021

SparkQA commented Sep 12, 2021

SparkQA commented Sep 12, 2021

SparkQA commented Sep 12, 2021

cloud-fan commented Sep 13, 2021

ulysses-you commented Sep 13, 2021

ulysses-you commented Jun 8, 2021 •

edited

Loading