[SPARK-28148][SQL] Repartition after join is not optimized away #27096

bmarcott · 2020-01-05T03:02:29Z

What changes were proposed in this pull request?

Extra shuffling was not eliminated after inner joins because they produce PartitioningCollection Partitioning and the current logic only matched on HashPartitioning.

Nothing was present in EnsureRequirements to eliminate parent sorting (within partitions) which was unnecessary when the same sort order was introduced by sortmergejoin

Copied from jira:
Partitioning & sorting is usually retained after join.

spark.conf.set('spark.sql.shuffle.partitions', '42')

df1 = spark.range(5000000, numPartitions=5)
df2 = spark.range(10000000, numPartitions=5)
df3 = spark.range(20000000, numPartitions=5)

# Reuse previous partitions & sort.
df1.join(df2, on='id').join(df3, on='id').explain()
# == Physical Plan ==
# *(8) Project [id#367L]
# +- *(8) SortMergeJoin [id#367L], [id#374L], Inner
#    :- *(5) Project [id#367L]
#    :  +- *(5) SortMergeJoin [id#367L], [id#369L], Inner
#    :     :- *(2) Sort [id#367L ASC NULLS FIRST], false, 0
#    :     :  +- Exchange hashpartitioning(id#367L, 42)
#    :     :     +- *(1) Range (0, 5000000, step=1, splits=5)
#    :     +- *(4) Sort [id#369L ASC NULLS FIRST], false, 0
#    :        +- Exchange hashpartitioning(id#369L, 42)
#    :           +- *(3) Range (0, 10000000, step=1, splits=5)
#    +- *(7) Sort [id#374L ASC NULLS FIRST], false, 0
#       +- Exchange hashpartitioning(id#374L, 42)
#          +- *(6) Range (0, 20000000, step=1, splits=5)

However here: Partitions persist through left join, sort doesn't.

df1.join(df2, on='id', how='left').repartition('id').sortWithinPartitions('id').explain()
# == Physical Plan ==
# *(5) Sort [id#367L ASC NULLS FIRST], false, 0
# +- *(5) Project [id#367L]
#    +- SortMergeJoin [id#367L], [id#369L], LeftOuter
#       :- *(2) Sort [id#367L ASC NULLS FIRST], false, 0
#       :  +- Exchange hashpartitioning(id#367L, 42)
#       :     +- *(1) Range (0, 5000000, step=1, splits=5)
#       +- *(4) Sort [id#369L ASC NULLS FIRST], false, 0
#          +- Exchange hashpartitioning(id#369L, 42)
#             +- *(3) Range (0, 10000000, step=1, splits=5)

Also here: Partitions do not persist though inner join.

df1.join(df2, on='id').repartition('id').sortWithinPartitions('id').explain()
# == Physical Plan ==
# *(6) Sort [id#367L ASC NULLS FIRST], false, 0
# +- Exchange hashpartitioning(id#367L, 42)
#    +- *(5) Project [id#367L]
#       +- *(5) SortMergeJoin [id#367L], [id#369L], Inner
#          :- *(2) Sort [id#367L ASC NULLS FIRST], false, 0
#          :  +- Exchange hashpartitioning(id#367L, 42)
#          :     +- *(1) Range (0, 5000000, step=1, splits=5)
#          +- *(4) Sort [id#369L ASC NULLS FIRST], false, 0
#             +- Exchange hashpartitioning(id#369L, 42)
#                +- *(3) Range (0, 10000000, step=1, splits=5)

dbtsai · 2020-01-06T20:34:59Z

Jenkins, ok to test.

SparkQA · 2020-01-07T00:34:27Z

Test build #116187 has finished for PR 27096 at commit ff573e8.

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

dongjoon-hyun · 2020-01-07T21:28:24Z

Thank you for making a PR, @bmarcott .

dbtsai · 2020-01-07T21:36:38Z

also ping @viirya

viirya

Currently EnsureRequirements ensures children distribution/ordering satisfies parent's requirement. That said, it checks if a child's distribution/ordering is good for its parent op. It not, it adds shuffle/sort on top of the child.

The cases failed to optimize I've seen so far, are incorrect or mismatched distribution/ordering output/requirement at child/parent sides.

For this part in EnsureRequirements:

def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
  // TODO: remove this after we create a physical operator for `RepartitionByExpression`.
  case operator @ ShuffleExchangeExec(upper: HashPartitioning, child, _) =>
    child.outputPartitioning match {
      case lower: HashPartitioning if upper.semanticEquals(lower) => child
      case _ => operator
    }
}

Read from the TODO, looks like it is supposed to have physical RepartitionByExpression that can skip shuffle if its child already satisfies the required distribution?

This PR targets a different scenario as the repartition/sort is added explicitly by the user. We have optimization at Optimizer that deals with logical plan, for example to remove Sort after sorted child plan.

So maybe EnsureRequirements not a good place to do that optimization? Maybe we need a separate rule for that? cc @cloud-fan

maropu · 2020-01-09T01:28:19Z

(The @viirya suggestion soulds resonable to me)

cloud-fan · 2020-01-09T02:10:31Z

an issue is that at logical phase we don't know the physical partitioning/sorting info (e.g. SMJ), so we can't optimize the user-specified sort/repartition well at logical phase.

I see 2 options here:

run an extra physical rule after EnsureRequirements, to remove useless shuffles
do as the TODO says, create a new physical node for RepartitionByExpression then we can go through the main logic in the ensureDistributionAndOrdering method, instead of a small hack in the apply method.

bmarcott · 2020-01-09T06:18:47Z

Thanks for taking a look!
Yes, the reason it is here is because the shuffle/sorting is introduced by EnsureRequirements itself, making the user added sorts/shuffles unnecessary.

Yea it felt a little hacky for optimization code to be in a rule called EnsureRequirements. Although then again, the new shuffles/sorts were introduced by it.

I'd like someone familiar with overall planner design to suggest whether I go through with 1st or 2nd option.
For 2nd option, won't I need to create a new physical node for both the repartition and sort, each of which is kinda a dummy physical node which relies on EnsureRequirements to add the necessary sorts/partitioning based on requiredChildDistribution and requiredChildOrdering

bmarcott · 2020-01-13T06:19:59Z

Any more suggestions?
I'm starting to think the new physical node is a bit of a hack as well just to reuse the code in ensureDistributionAndOrdering. We wouldn't want this "no op" SparkPlan to show up in the plan right?

cloud-fan · 2020-01-13T08:06:59Z

can we try option 1? Seems like we need to do some experiments here. I'm not sure which option is better without seeing the concrete code.

ulysses-you · 2020-01-14T02:38:39Z

+1, how about add a rule named like PruneShuffleAndSort.
At least include:

// TODO: remove this after we create a physical operator for `RepartitionByExpression`.
case operator @ ShuffleExchangeExec(upper: HashPartitioning, child, _) =>
  child.outputPartitioning match {
    case lower: HashPartitioning if upper.semanticEquals(lower) => child
    case _ => operator
  }

this pr code
PR-26946

bmarcott · 2020-01-14T05:17:42Z

i'll update soon with the new rule PruneShuffleAndSort
I'm iffy on including PR-26946.
EnsureRequirements should probably only introduce shuffles/ordering if it needs to, rather than adding blindly and relying on later rule to prune. Or maybe it is better to include so it'll apply to any case (haven't thought deeply on the cases here)?

bmarcott · 2020-01-15T03:10:42Z

I started to try including #26946, but it feels a little messy.

Notice that the numPartitions of the previous shuffle are maintained, but the distribution/partitioning is changed:
case (ShuffleExchangeExec(partitioning, child, _), distribution: OrderedDistribution) => ShuffleExchangeExec(distribution.createPartitioning(partitioning.numPartitions), child)

If we try to remove this code from EnsureRequirements, it'll create a new shuffle node with defaultNumPreShufflePartitions. I'm not sure it makes sense to have a general rule that if we have a shuffle with range partitioning with another shuffle as child, we eliminate the child but reuse it's numpartitions.

SparkQA · 2020-01-15T04:06:51Z

Test build #116740 has finished for PR 27096 at commit 1916ef9.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class PruneShuffleAndSort() extends Rule[SparkPlan]

cloud-fan · 2020-01-15T05:51:29Z

In general, shuffles are added by Spark and Spark can pick the best # of partitions. However, for user-specified shuffles (df.repartition), the semantic should be respecting the user request and keep the # of partitions.

SparkQA · 2020-01-15T12:47:30Z

Test build #116772 has finished for PR 27096 at commit 85003be.

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

bmarcott · 2020-01-16T06:16:57Z

updated with a new sparkplan rule and added a test which makes sure a user's repartition with a different numPartitions would not be eliminated (don't want to change expected numPartitions).

Review when you get a chance :).

SparkQA · 2020-01-16T07:52:55Z

Test build #116821 has finished for PR 27096 at commit 9e895bd.

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

SparkQA · 2020-01-16T08:05:02Z

Test build #116820 has finished for PR 27096 at commit 2517bea.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-01-16T08:05:02Z

Test build #116815 has finished for PR 27096 at commit 00b082a.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

bmarcott · 2020-01-19T04:22:51Z

retest this please

bmarcott · 2020-01-19T05:21:06Z

do the tests run post merge with master, or do I need to rebase to pick up fix #27242?

maropu · 2020-01-20T00:39:59Z

retest this please

SparkQA · 2020-01-20T04:57:22Z

Test build #117038 has finished for PR 27096 at commit 9e895bd.

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

bmarcott · 2020-01-21T19:04:13Z

rebased on latest master

SparkQA · 2020-01-21T23:06:50Z

Test build #117198 has finished for PR 27096 at commit 3ecdccb.

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

bmarcott · 2020-01-22T02:23:23Z

any remaining comments/concerns?

SparkQA · 2020-01-22T06:33:32Z

Test build #117211 has finished for PR 27096 at commit bb794db.

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

bmarcott · 2020-01-23T02:59:12Z

@cloud-fan and/or others please review when convenient

bmarcott · 2020-01-28T04:02:39Z

mind taking another look?
@cloud-fan
@viirya

dongjoon-hyun · 2020-01-31T18:56:35Z

Retest this please.

SparkQA · 2020-01-31T23:49:22Z

Test build #117679 has finished for PR 27096 at commit bb794db.

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

bmarcott · 2020-02-10T04:56:42Z

@cloud-fan @viirya
bump 😃

maryannxue · 2020-02-12T17:01:44Z

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

+    plan.transformUp {
+      case operator @ ShuffleExchangeExec(upper: HashPartitioning, child, _) =>
+        child.outputPartitioning match {
+          case lower: HashPartitioning if upper.semanticEquals(lower) => child


Why doesn't this apply to RangeParititoning? Is it just we assume repartition() would only do HashPartitioning?
And what happens if someone does df1.join(df2, Seq("id"), "left").repartition(100, df("some_other_column")).repartition(20, df1("id")) or df1.join(df2, Seq("id"), "left").sortWithinPartition(df1("some_other_column")).sortWithinPartition(df1("id")) ? We should be able to optimize that out too, right? It would be nice to make this rule more general and cover a wider range of cases.

Thanks for providing feedback.
Let me take a look into your specific examples and think a little more about it.

@maryannxue
This PR focused on fixing removing unnecessary sorting and shuffling after a join, which potentially includes its own ShuffleExchangeExec with HashPartioning. Both cases you mentioned are already optimized properly: the shuffling on "some_other_column" is removed and all sortWithinPartitions are removed (due to previous optimizations in logical plan, and the optimizations introduced here)

I wouldn't mind generalizing to all Partitioning types of the ShuffleExchangeExec, but I am not sure how to compare two partitioning types for equality. You can see the special case for HashPartitioning in this PR.

@maryannxue ping

bmarcott · 2020-04-23T06:12:58Z

Fixed latest conflicts/rebased on latest master

@cloud-fan could you take a look when you get a chance?

SparkQA · 2020-04-23T07:05:02Z

Test build #121654 has finished for PR 27096 at commit 9db29e6.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

bmarcott · 2020-06-05T02:32:58Z

@cloud-fan @viirya could you help review or add a suggested reviewer here?

bmarcott · 2020-06-14T11:21:08Z

@cloud-fan @viirya friendly ping

github-actions · 2020-09-23T00:45:48Z

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!

bmarcott changed the title ~~SPARK-28148: repartition after join is not optimized away~~ [SPARK-28148] [CORE]: repartition after join is not optimized away Jan 5, 2020

bmarcott changed the title ~~[SPARK-28148] [CORE]: repartition after join is not optimized away~~ [SPARK-28148][CORE]: repartition after join is not optimized away Jan 5, 2020

dbtsai changed the title ~~[SPARK-28148][CORE]: repartition after join is not optimized away~~ [SPARK-28148][SQL] Repartition after join is not optimized away Jan 6, 2020

dongjoon-hyun added the SQL label Jan 7, 2020

viirya reviewed Jan 9, 2020

View reviewed changes

bmarcott force-pushed the nmarcott-reuse-sort-order branch from ff573e8 to 1916ef9 Compare January 15, 2020 02:55

bmarcott force-pushed the nmarcott-reuse-sort-order branch from 2517bea to 9e895bd Compare January 16, 2020 06:13

bmarcott force-pushed the nmarcott-reuse-sort-order branch from 9e895bd to 3ecdccb Compare January 21, 2020 19:04

cloud-fan mentioned this pull request Jan 31, 2020

[SPARK-30036][SQL] Fix: REPARTITION hint does not work with order by #26946

Closed

maryannxue reviewed Feb 12, 2020

View reviewed changes

bmarcott added 5 commits April 23, 2020 15:09

create PruneShuffleAndSort physical rule

adc2933

fix match error

7d3e180

add test

b177973

add doc

8f0679e

update doc

9db29e6

bmarcott force-pushed the nmarcott-reuse-sort-order branch from bb794db to 9db29e6 Compare April 23, 2020 06:11

github-actions bot added the Stale label Sep 23, 2020

github-actions bot closed this Sep 24, 2020

maropu mentioned this pull request Sep 29, 2020

[SPARK-32820][SQL] Remove redundant shuffle exchanges inserted by EnsureRequirements #29677

Closed

[SPARK-28148][SQL] Repartition after join is not optimized away #27096

[SPARK-28148][SQL] Repartition after join is not optimized away #27096

Uh oh!

Conversation

bmarcott commented Jan 5, 2020 • edited by dongjoon-hyun Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Uh oh!

dbtsai commented Jan 6, 2020

Uh oh!

SparkQA commented Jan 7, 2020

Uh oh!

dongjoon-hyun commented Jan 7, 2020

Uh oh!

dbtsai commented Jan 7, 2020

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

maropu commented Jan 9, 2020

Uh oh!

cloud-fan commented Jan 9, 2020

Uh oh!

bmarcott commented Jan 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

bmarcott commented Jan 13, 2020

Uh oh!

cloud-fan commented Jan 13, 2020

Uh oh!

ulysses-you commented Jan 14, 2020

Uh oh!

bmarcott commented Jan 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

bmarcott commented Jan 15, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jan 15, 2020

Uh oh!

cloud-fan commented Jan 15, 2020

Uh oh!

SparkQA commented Jan 15, 2020

Uh oh!

bmarcott commented Jan 16, 2020

Uh oh!

SparkQA commented Jan 16, 2020

Uh oh!

SparkQA commented Jan 16, 2020

Uh oh!

SparkQA commented Jan 16, 2020

Uh oh!

bmarcott commented Jan 19, 2020

Uh oh!

bmarcott commented Jan 19, 2020

Uh oh!

maropu commented Jan 20, 2020

Uh oh!

SparkQA commented Jan 20, 2020

Uh oh!

bmarcott commented Jan 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

bmarcott commented Jan 22, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jan 22, 2020

Uh oh!

bmarcott commented Jan 23, 2020

Uh oh!

bmarcott commented Jan 28, 2020

Uh oh!

dongjoon-hyun commented Jan 31, 2020

Uh oh!

SparkQA commented Jan 31, 2020

Uh oh!

bmarcott commented Feb 10, 2020

Uh oh!

bmarcott commented Jan 5, 2020 •

edited by dongjoon-hyun

Loading

bmarcott commented Jan 9, 2020 •

edited

Loading

bmarcott commented Jan 14, 2020 •

edited

Loading

bmarcott commented Jan 15, 2020 •

edited

Loading

bmarcott commented Jan 21, 2020 •

edited

Loading

bmarcott commented Jan 22, 2020 •

edited

Loading

bmarcott Feb 24, 2020 •

edited

Loading