Skip to content

[SPARK-34637] [SQL] Support DPP + AQE when the broadcast exchange can be reused #31756

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
JkSelf wants to merge 6 commits into from
Closed

[SPARK-34637] [SQL] Support DPP + AQE when the broadcast exchange can be reused #31756

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3bc4baf
Support DPP + AQE when find the broadcast exchange can reuse
JkSelf Apr 15, 2021
657c61b
resolve the comments and fix the failed test
JkSelf Apr 15, 2021
c7b3735
add the new test to test DPP side broadcast query stage is created fi…
JkSelf Apr 16, 2021
6b07c84
update subquery symbols
JkSelf May 6, 2021
4ccd4b8
resolve the comments
JkSelf May 11, 2021
701f1c3
resolve the comments
JkSelf May 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
resolve the comments
  • Loading branch information
@JkSelf
JkSelf committed May 11, 2021
commit 4ccd4b8b1d2604f05a4564303d0b8aa65ea5dfa3
3 changes: 1 addition & 2 deletions ...ain/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,7 @@ case class PlanAdaptiveDynamicPruningFilters(rootPlan: SparkPlan) extends Rule[S

if (canReuseExchange) {
exchange.setLogicalLink(adaptivePlan.executedPlan.logicalLink.get)
val newAdaptivePlan = AdaptiveSparkPlanExec(
exchange, adaptivePlan.context, adaptivePlan.preprocessingRules, true)
val newAdaptivePlan = adaptivePlan.copy(inputPlan = exchange)

val broadcastValues = SubqueryBroadcastExec(
name, index, buildKeys, newAdaptivePlan)
Expand Down
13 changes: 7 additions & 6 deletions sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1475,14 +1475,14 @@ abstract class DynamicPartitionPruningSuiteBase
}
}

test("SPARK-34637: test DPP side broadcast query stage is created firstly") {
test("SPARK-34637: DPP side broadcast query stage is created firstly") {
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") {
val df = sql(
""" WITH view1 as (
""" WITH v as (
| SELECT f.store_id FROM fact_stats f WHERE f.units_sold = 70 group by f.store_id
| )
|
| SELECT * FROM view1 v1 join view1 v2 WHERE v1.store_id = v2.store_id
| SELECT * FROM v v1 join v v2 WHERE v1.store_id = v2.store_id
""".stripMargin)

// A possible resulting query plan:
Expand All @@ -1492,9 +1492,10 @@ abstract class DynamicPartitionPruningSuiteBase
// +- Exchange
// +- HashAggregate
// +- Filter
// +- FileScan
// Dynamicpruning Subquery
// +- AdaptiveSparkPlan
// +- FileScan [PartitionFilters: [isnotnull(store_id#3367),
// dynamicpruningexpression(store_id#3367 IN dynamicpruning#3385)]]
// +- SubqueryBroadcast dynamicpruning#3385
// +- AdaptiveSparkPlan
// +- BroadcastQueryStage
Copy link
Contributor

@cloud-fan cloud-fan Apr 19, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

there is no other place to reuse this broadcast, right?

Copy link
Contributor Author

@JkSelf JkSelf May 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This broadcast only be reused in the build side.

Copy link
Contributor

@tgravescs tgravescs Apr 28, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how is broadcast before the FileScan? what is being broadcast

Copy link
Contributor Author

@JkSelf JkSelf May 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This broadcast is in the DPP subquery of the FileScan. It will broadcast the results of the build side and then prune the dataset.

Copy link
Contributor

@cloud-fan cloud-fan May 10, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you make it more clear that this is the subquery in the file scan node, not the child of it?

// +- BroadcastExchange
//
Expand Down