[SPARK-32332][SQL] Make it possible to implement columnar exchanges #29262

cloud-fan · 2020-07-27T18:18:35Z

What changes were proposed in this pull request?

This PR adds abstract classes for shuffle and broadcast, so that users can provide their columnar implementations.

This PR updates several places to use the abstract exchange classes, and also update AdaptiveSparkPlanExec so that the columnar rules can see exchange nodes.

This is an alternative of #29134 .
Close #29134

Why are the changes needed?

To allow columnar exchanges.

Does this PR introduce any user-facing change?

no

How was this patch tested?

new tests

cloud-fan · 2020-07-27T18:19:24Z

cc @andygrove @tgravescs @maryannxue

cloud-fan · 2020-07-27T18:21:07Z

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

  }

  private def newQueryStage(e: Exchange): QueryStageExec = {
    val optimizedPlan = applyPhysicalRules(e.child, queryStageOptimizerRules)


it's important to hide the exchange nodes from the stage optimizer rules, as that's the assumption of these rules.

Then we don't need https://github.com/apache/spark/pull/29134/files#diff-a30c7a6fcdcdd13e57135fd04d05f3b7R115

andygrove · 2020-07-27T18:33:44Z

Thanks @cloud-fan I will test our AQE POC with these changes.

SparkQA · 2020-07-27T22:48:00Z

Test build #126660 has finished for PR 29262 at commit c76bf9c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait BroadcastExchangeLike extends Exchange
trait ShuffleExchangeLike extends Exchange

andygrove · 2020-07-28T14:16:15Z

Thanks @cloud-fan. I have tested these changes both with Spark 3.1 and also back ported to the 3.0 branch and everything is working well, so LGTM.

I wish I had thought to separate the rules out into stage creation and post-stage creation. That made things much simpler.

cc @tgravescs

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

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

SparkQA · 2020-07-28T17:28:42Z

Test build #126728 has finished for PR 29262 at commit b90ea90.

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

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

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

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

SparkQA · 2020-07-29T07:05:02Z

Test build #126750 has finished for PR 29262 at commit 8aff7a8.

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

tgravescs · 2020-07-29T13:41:42Z

test this please

tgravescs

looks good, pending jenkins

SparkQA · 2020-07-29T16:21:51Z

Test build #126775 has finished for PR 29262 at commit 8aff7a8.

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

tgravescs · 2020-07-29T16:25:24Z

test this please

SparkQA · 2020-07-29T19:10:50Z

Test build #126779 has finished for PR 29262 at commit 50175d7.

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

tgravescs · 2020-07-29T19:25:13Z

I merged this to master, unfortunately wouldn't pick clean to branch-3.0.

@cloud-fan would you want to put up PR for branch-3.0? Otherwise Andy or myself can.

SparkQA · 2020-07-29T20:49:05Z

Test build #126785 has finished for PR 29262 at commit 50175d7.

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

cloud-fan · 2020-07-30T06:39:28Z

Does it qualify a backport? It's kind of a new feature.

tgravescs · 2020-07-30T13:16:00Z

I went back and forth on that as I can see it both ways. I ended up filing it as a bug as it prevents us from properly doing columnar processing (introduced in 3.0.0) with AQE. I also thought it was all internal and a fairly isolated path and was hoping to get it into 3.0.1 if others didn't disagree since users are starting to to use AQE in 3.0. what do you think?

cloud-fan · 2020-07-30T13:46:49Z

I agree it's not that risky(the code is the same as before, just with one more abstraction layer: exchange-like). I don't have a strong opinion here. @andygrove can you help to create a backport PR?

andygrove · 2020-07-30T14:13:58Z

Sure, I'll create a PR this morning. Thanks @cloud-fan and @tgravescs.

support columnar shuffle

c76bf9c

probot-autolabeler bot added SQL STRUCTURED STREAMING labels Jul 27, 2020

cloud-fan commented Jul 27, 2020

View reviewed changes

This was referenced Jul 28, 2020

[SPARK-32332][SQL] Support columnar exchanges when AQE is enabled #29134

Closed

[FEA] Support Adaptive Execution NVIDIA/spark-rapids#5

Closed

[WIP] [POC] Enable broadcast and shuffle exchanges to stay on GPU when AQE is enabled NVIDIA/spark-rapids#361

Closed

tgravescs reviewed Jul 28, 2020

View reviewed changes

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

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

address comments

b90ea90

tgravescs reviewed Jul 28, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala Show resolved Hide resolved

viirya reviewed Jul 29, 2020

View reviewed changes

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

viirya reviewed Jul 29, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala Outdated Show resolved Hide resolved

address comments

8aff7a8

fix typo

50175d7

tgravescs approved these changes Jul 29, 2020

View reviewed changes

viirya approved these changes Jul 29, 2020

View reviewed changes

asfgit closed this in a025a89 Jul 29, 2020

cloud-fan changed the title ~~[SPARK-32332][SQL] Support columnar exchanges~~ [SPARK-32332][SQL] Make it possible to implement columnar exchanges Aug 5, 2020

[SPARK-32332][SQL] Make it possible to implement columnar exchanges #29262

[SPARK-32332][SQL] Make it possible to implement columnar exchanges #29262

Uh oh!

Conversation

cloud-fan commented Jul 27, 2020

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!

cloud-fan commented Jul 27, 2020

Uh oh!

cloud-fan Jul 27, 2020

Choose a reason for hiding this comment

Uh oh!

andygrove commented Jul 27, 2020

Uh oh!

SparkQA commented Jul 27, 2020

Uh oh!

andygrove commented Jul 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Jul 28, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Jul 29, 2020

Uh oh!

tgravescs commented Jul 29, 2020

Uh oh!

tgravescs left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 29, 2020

Uh oh!

tgravescs commented Jul 29, 2020

Uh oh!

SparkQA commented Jul 29, 2020

Uh oh!

tgravescs commented Jul 29, 2020

Uh oh!

SparkQA commented Jul 29, 2020

Uh oh!

cloud-fan commented Jul 30, 2020

Uh oh!

tgravescs commented Jul 30, 2020

Uh oh!

cloud-fan commented Jul 30, 2020

Uh oh!

andygrove commented Jul 30, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

andygrove commented Jul 28, 2020 •

edited

Loading