Skip to content

[SPARK-30528][SQL] Turn off DPP subquery duplication by default #27551

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
maryannxue wants to merge 2 commits into from
Closed

[SPARK-30528][SQL] Turn off DPP subquery duplication by default #27551

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3514cf4
[SPARK-30528] Potential performance regression with DPP subquery dupl…
maryannxue Feb 12, 2020
29f5ae8
address review comments
maryannxue Feb 12, 2020
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
12 changes: 6 additions & 6 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,11 +259,11 @@ object SQLConf {
.doubleConf
.createWithDefault(0.5)

val DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST =
buildConf("spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcast")
val DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY =
buildConf("spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly")
.internal()
.doc("When true, dynamic partition pruning will seek to reuse the broadcast results from " +
"a broadcast hash join operation.")
.doc("When true, dynamic partition pruning will only apply when the broadcast exchange of " +
"a broadcast hash join operation can be reused as the dynamic pruning filter.")
.booleanConf
.createWithDefault(true)

Expand Down Expand Up @@ -2303,8 +2303,8 @@ class SQLConf extends Serializable with Logging {
def dynamicPartitionPruningFallbackFilterRatio: Double =
getConf(DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO)

def dynamicPartitionPruningReuseBroadcast: Boolean =
getConf(DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST)
def dynamicPartitionPruningReuseBroadcastOnly: Boolean =
getConf(DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY)

def stateStoreProviderClass: String = getConf(STATE_STORE_PROVIDER_CLASS)

Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/dynamicpruning/PartitionPruning.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper {
filteringPlan: LogicalPlan,
joinKeys: Seq[Expression],
hasBenefit: Boolean): LogicalPlan = {
val reuseEnabled = SQLConf.get.dynamicPartitionPruningReuseBroadcast
val reuseEnabled = SQLConf.get.exchangeReuseEnabled
val index = joinKeys.indexOf(filteringKey)
if (hasBenefit || reuseEnabled) {
// insert a DynamicPruning wrapper to identify the subquery during query planning
Expand All @@ -96,7 +96,7 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper {
filteringPlan,
joinKeys,
index,
!hasBenefit),
!hasBenefit || SQLConf.get.dynamicPartitionPruningReuseBroadcastOnly),
pruningPlan)
} else {
// abort dynamic partition pruning
Expand Down
5 changes: 1 addition & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/dynamicpruning/PlanDynamicPruningFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,9 +36,6 @@ import org.apache.spark.sql.internal.SQLConf
case class PlanDynamicPruningFilters(sparkSession: SparkSession)
extends Rule[SparkPlan] with PredicateHelper {

private def reuseBroadcast: Boolean =
SQLConf.get.dynamicPartitionPruningReuseBroadcast && SQLConf.get.exchangeReuseEnabled

/**
* Identify the shape in which keys of a given plan are broadcasted.
*/
Expand All @@ -59,7 +56,7 @@ case class PlanDynamicPruningFilters(sparkSession: SparkSession)
sparkSession, sparkSession.sessionState.planner, buildPlan)
// Using `sparkPlan` is a little hacky as it is based on the assumption that this rule is
// the first to be applied (apart from `InsertAdaptiveSparkPlan`).
val canReuseExchange = reuseBroadcast && buildKeys.nonEmpty &&
val canReuseExchange = SQLConf.get.exchangeReuseEnabled && buildKeys.nonEmpty &&
plan.find {
case BroadcastHashJoinExec(_, _, _, BuildLeft, _, left, _) =>
left.sameResult(sparkPlan)
Expand Down
Loading