Skip to content

[SPARK-35455][SQL] Unify empty relation optimization between normal and AQE optimizer #32602

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
ulysses-you wants to merge 23 commits into from
Closed

[SPARK-35455][SQL] Unify empty relation optimization between normal and AQE optimizer #32602

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
4514d27
Enhance EliminateUnnecessaryJoin
ulysses-you May 20, 2021
5097247
fix
ulysses-you May 20, 2021
8df68d9
LocalRelation early
ulysses-you May 20, 2021
165077b
fix
ulysses-you May 20, 2021
2f5fa20
split PropagateEmptyRelation
ulysses-you May 21, 2021
7b80db0
fix
ulysses-you May 21, 2021
05e074c
split PropagateEmptyRelationBase
ulysses-you May 21, 2021
48dd92a
fix
ulysses-you May 21, 2021
8f4dc80
merge
ulysses-you May 23, 2021
1220087
nit
ulysses-you May 23, 2021
e26df96
fix
ulysses-you May 24, 2021
f7a14cf
fix
ulysses-you May 24, 2021
2c0dfb0
empty
ulysses-you May 24, 2021
0e151d9
naaj
ulysses-you May 24, 2021
c086f72
comment
ulysses-you May 24, 2021
9b78ac0
nit
ulysses-you May 24, 2021
767dd92
inline
ulysses-you May 24, 2021
47e0c3a
test
ulysses-you May 24, 2021
d9ca6da
comment
ulysses-you May 24, 2021
2fead86
Relations
ulysses-you May 24, 2021
0754936
indentation
ulysses-you May 24, 2021
a6213ea
test name
ulysses-you May 25, 2021
624e45e
nit
ulysses-you May 25, 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
8 changes: 7 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.sql.execution.adaptive

import org.apache.spark.sql.catalyst.analysis.UpdateAttributeNullability
import org.apache.spark.sql.catalyst.optimizer.PropagateEmptyRelation
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, LogicalPlanIntegrity, PlanHelper}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.internal.SQLConf
Expand All @@ -27,7 +29,11 @@ import org.apache.spark.util.Utils
*/
class AQEOptimizer(conf: SQLConf) extends RuleExecutor[LogicalPlan] {
private val defaultBatches = Seq(
Batch("Eliminate Unnecessary Join", Once, EliminateUnnecessaryJoin),
Batch("LocalRelation early", Once,
ConvertToLocalRelation,
EliminateUnnecessaryJoin,
PropagateEmptyRelation,
UpdateAttributeNullability),
Comment thread
ulysses-you marked this conversation as resolved.
Batch("Demote BroadcastHashJoin", Once, DemoteBroadcastHashJoin)
)

Expand Down
32 changes: 32 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ConvertToLocalRelation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.spark.sql.execution.adaptive

import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule

/**
* Converts empty query stage to empty `LocalRelation`
*/
object ConvertToLocalRelation extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case l @ LogicalQueryStage(_, stage: QueryStageExec) if stage.resultOption.get().isDefined &&

@cloud-fan cloud-fan May 20, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Sorry I made the wrong decision. This may change the output partitioning and is not always safe/beneficial (we may add extra shuffles in the planning phase later).

Looking at the optimizations for empty local relations, some of them are likely beneficial and we can always do: eliminate join, aggregate, limit, repartition, sort, generate

Some may not that beneficial and we shouldn't do: simplify union, eliminate project/filter/sample.

My new idea:

  1. Create a new rule PropagateEmptyRelationBasic, which deals with local relaion only, and is not very beneficial (eliminate project, filter, etc), and runs in the normal optimizer only
  2. Create a new rule PropagateEmptyRelationAdvanced, which deals with both local relation and query stage, and is very beneficial (eliminate join, aggregate, etc.), and runs in both normal and AQE optimizer

The old EliminateUnnecessaryJoin and PropagateEmptyRelation rules should be removed and merged into the new rules.

PropagateEmptyRelationAdvanced may not be able to access QueryStageExec which is in sql/core. We can let this rule take checkRowCount functiton as a parameter.

@ulysses-you ulysses-you May 20, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I see the issue. If we worry about the LocalRelation output partitioning, we can just mark LocalRelationExec output partitioning as SinglePartition to avoid extra shuffle. But it doesn't work with optimization like non empty left semi/anti elimintaion and multi-join case.

we can always do: eliminate join, aggregate

Not sure we can always do this if we don't want to introduce extra shuffle. And the issue it has already existed in current EliminateUnnecessaryJoin, like this plan

Aggregate (same key with join)
  Join Inner
    LocalRelation
    xxx

An another idea, if we plan to support extra shuffle later and don't expect introduce shuffle at AQE optimzier side, then is it better to check the physical plan requiredChildDistribution ? We can only allow one node which has a valid requiredChildDistribution (not UnspecifiedDistribution) in one query stage, and skip optimize if one query stage has two or more valid requiredChildDistribution nodes. Thus we can run PropagateEmptyRelation safely.

@cloud-fan cloud-fan May 20, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

If we turn a broadcast stage into local relation without changing other plan parts, seems we will broadcast the local relation again. SinglePartition can't help here.

@ulysses-you ulysses-you May 20, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

If it's hard to avoid introduce shuffle at AQE optimizer, how about add extra shuffle check between AQE optimizer and stage preparation ? Then it won't affect the extra shuffle in stage preparation.

@cloud-fan cloud-fan May 20, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

We can do that, but it seems not worth the complexity. It's not very helpful to turn query stage to local relation if we can't use it to eliminate expensive operators like join, agg, sort, etc.

I think my proposal is simpler and effective enough. The EliminateUnnecessaryJoin today does not consider extra shuffles either and blindly eliminate joins (so as the query stages) if possible

stage.getRuntimeStatistics.rowCount.contains(0) =>
LocalRelation(l.output, Seq.empty)
}
}
38 changes: 11 additions & 27 deletions ...ore/src/main/scala/org/apache/spark/sql/execution/adaptive/EliminateUnnecessaryJoin.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.spark.sql.execution.adaptive

import org.apache.spark.sql.catalyst.planning.ExtractSingleColumnNullAwareAntiJoin
import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftSemi}
import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
import org.apache.spark.sql.catalyst.plans.logical.{Join, LocalRelation, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.joins.HashedRelationWithAllNullKeys
Expand All @@ -28,21 +28,12 @@ import org.apache.spark.sql.execution.joins.HashedRelationWithAllNullKeys
* 1. Join is single column NULL-aware anti join (NAAJ), and broadcasted [[HashedRelation]]
* is [[HashedRelationWithAllNullKeys]]. Eliminate join to an empty [[LocalRelation]].
*
* 2. Join is inner join, and either side of join is empty. Eliminate join to an empty
* [[LocalRelation]].
* 2. Join is left semi join
* Join right side is non-empty and condition is empty. Eliminate join to its left side.
*
* 3. Join is left semi join
* 3.1. Join right side is empty. Eliminate join to an empty [[LocalRelation]].
* 3.2. Join right side is non-empty and condition is empty. Eliminate join to its left side.
*
* 4. Join is left anti join
* 4.1. Join right side is empty. Eliminate join to its left side.
* 4.2. Join right side is non-empty and condition is empty. Eliminate join to an empty
* 3. Join is left anti join
* Join right side is non-empty and condition is empty. Eliminate join to an empty
* [[LocalRelation]].
*
* This applies to all joins (sort merge join, shuffled hash join, broadcast hash join, and
* broadcast nested loop join), because sort merge join and shuffled hash join will be changed
* to broadcast hash join with AQE at the first place.
*/
object EliminateUnnecessaryJoin extends Rule[LogicalPlan] {

Expand All @@ -53,36 +44,29 @@ object EliminateUnnecessaryJoin extends Rule[LogicalPlan] {
case _ => false
}

private def checkRowCount(plan: LogicalPlan, hasRow: Boolean): Boolean = plan match {
private def checkRowCount(plan: LogicalPlan): Boolean = plan match {
case LogicalQueryStage(_, stage: QueryStageExec) if stage.resultOption.get().isDefined =>
stage.getRuntimeStatistics.rowCount match {
case Some(count) => hasRow == (count > 0)
case Some(count) => count > 0
case _ => false
}

case _ => false
}

def apply(plan: LogicalPlan): LogicalPlan = plan.transformDown {
def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
case j @ ExtractSingleColumnNullAwareAntiJoin(_, _) if isRelationWithAllNullKeys(j.right) =>
LocalRelation(j.output, data = Seq.empty, isStreaming = j.isStreaming)

case j @ Join(_, _, Inner, _, _) if checkRowCount(j.left, hasRow = false) ||
checkRowCount(j.right, hasRow = false) =>
LocalRelation(j.output, data = Seq.empty, isStreaming = j.isStreaming)

case j @ Join(_, _, LeftSemi, condition, _) =>
if (checkRowCount(j.right, hasRow = false)) {
LocalRelation(j.output, data = Seq.empty, isStreaming = j.isStreaming)
} else if (condition.isEmpty && checkRowCount(j.right, hasRow = true)) {
if (condition.isEmpty && checkRowCount(j.right)) {
j.left
} else {
j
}

case j @ Join(_, _, LeftAnti, condition, _) =>
if (checkRowCount(j.right, hasRow = false)) {
j.left
} else if (condition.isEmpty && checkRowCount(j.right, hasRow = true)) {
if (condition.isEmpty && checkRowCount(j.right)) {
LocalRelation(j.output, data = Seq.empty, isStreaming = j.isStreaming)
} else {
j
Expand Down
2 changes: 1 addition & 1 deletion 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 @@ -1382,7 +1382,7 @@ abstract class DynamicPartitionPruningSuiteBase
withSQLConf(
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key -> EliminateUnnecessaryJoin.ruleName) {
SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key -> ConvertToLocalRelation.ruleName) {
val df = sql(
"""
|SELECT * FROM fact_sk f
Expand Down
66 changes: 65 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,8 @@ class AdaptiveQueryExecSuite
test("Empty stage coalesced to 1-partition RDD") {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true") {
SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key -> ConvertToLocalRelation.ruleName) {
val df1 = spark.range(10).withColumn("a", 'id)
val df2 = spark.range(10).withColumn("b", 'id)
withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
Expand Down Expand Up @@ -1307,6 +1308,69 @@ class AdaptiveQueryExecSuite
}
}

test("SPARK-35455: Enhance EliminateUnnecessaryJoin - single join") {

@cloud-fan cloud-fan May 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

let's update the test name and PR title: Unify empty relation optimization between normal and AQE optimizer

@ulysses-you ulysses-you May 25, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Updated it and also updated the PR title.

withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
Seq(
// left semi join and empty left side

@cloud-fan cloud-fan May 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

We can't optimize this before this PR?

@ulysses-you ulysses-you May 24, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

yeah, we cann't. Before we only check right side with LeftSemi/LeftAnti.

And the test should use different column to do filter and join in case of InferFiltersFromConstraints make right side empty. Updated it.

("SELECT * FROM (SELECT * FROM testData WHERE key = 0)t1 LEFT SEMI JOIN testData2 t2 ON " +
"t1.key = t2.a", true),
// left anti join and empty left side
("SELECT * FROM (SELECT * FROM testData WHERE key = 0)t1 LEFT ANTI JOIN testData2 t2 ON " +
"t1.key = t2.a", true),
// left outer join and empty left side
("SELECT * FROM (SELECT * FROM testData WHERE key = 0)t1 LEFT JOIN testData2 t2 ON " +
"t1.key = t2.a", true),
// left outer join and non-empty left side
("SELECT * FROM testData t1 LEFT JOIN testData2 t2 ON " +
"t1.key = t2.a", false),
// right outer join and empty right side
("SELECT * FROM testData t1 RIGHT JOIN (SELECT * FROM testData2 WHERE b = 0)t2 ON " +
"t1.key = t2.a", true),
// right outer join and non-empty right side
("SELECT * FROM testData t1 RIGHT JOIN testData2 t2 ON " +
"t1.key = t2.a", false),
// full outer join and both side empty
("SELECT * FROM (SELECT * FROM testData WHERE key = 0)t1 FULL JOIN " +
"(SELECT * FROM testData2 WHERE b = 0)t2 ON t1.key = t2.a", true),
// full outer join and left side empty right side non-empty
("SELECT * FROM (SELECT * FROM testData WHERE key = 0)t1 FULL JOIN " +
"testData2 t2 ON t1.key = t2.a", true)
).foreach { case (query, isEliminated) =>
val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(query)
assert(findTopLevelBaseJoin(plan).size == 1)
assert(findTopLevelBaseJoin(adaptivePlan).isEmpty == isEliminated, adaptivePlan)
}
}
}

test("SPARK-35455: Enhance EliminateUnnecessaryJoin - multi join") {

@cloud-fan cloud-fan May 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

ditto

withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
Seq(
"""
|SELECT * FROM testData t1
| JOIN (SELECT * FROM testData2 WHERE b = 0) t2 ON t1.key = t2.a
| LEFT JOIN testData2 t3 ON t1.key = t3.a
|""".stripMargin,
"""
|SELECT * FROM (SELECT * FROM testData WHERE key = 0) t1
| LEFT ANTI JOIN testData2 t2
| FULL JOIN (SELECT * FROM testData2 WHERE b = 0) t3 ON t1.key = t3.a
|""".stripMargin,
"""
|SELECT * FROM testData t1
| LEFT SEMI JOIN (SELECT * FROM testData2 WHERE b = 0)
| RIGHT JOIN testData2 t3 on t1.key = t3.a
|""".stripMargin
).foreach { query =>
val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(query)
assert(findTopLevelBaseJoin(plan).size == 2)
assert(findTopLevelBaseJoin(adaptivePlan).isEmpty)
}
}
}

test("SPARK-32753: Only copy tags to node with no tags") {
withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
withTempView("v1") {
Expand Down