Skip to content

Commit

Permalink
[SPARK-37828][SQL] Push down filters through RebalancePartitions
Browse files Browse the repository at this point in the history
### What changes were proposed in this pull request?

Push down filters through RebalancePartitions. How to reproduce this issue:
```scala
spark.sql("SELECT * FROM (SELECT /*+ REBALANCE */ * FROM range(10)) t1 WHERE id = 3").explain(true)
```
Output:
```
== Optimized Logical Plan ==
Filter (id#0L = 3)
+- RebalancePartitions
   +- Range (0, 10, step=1, splits=None)
```
After this pr:
```
RebalancePartitions
+- Filter (id#0L = 3)
   +- Range (0, 10, step=1, splits=None)
```

### Why are the changes needed?

Improve query performance.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Unit test.

Closes apache#35118 from wangyum/SPARK-37828.

Authored-by: Yuming Wang <yumwang@ebay.com>
Signed-off-by: Yuming Wang <yumwang@ebay.com>
  • Loading branch information
wangyum authored and dchvn committed Jan 19, 2022
1 parent f97c68d commit 6d8426e
Show file tree
Hide file tree
Showing 2 changed files with 9 additions and 0 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -1615,6 +1615,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
case _: Pivot => true
case _: RepartitionByExpression => true
case _: Repartition => true
case _: RebalancePartitions => true
case _: ScriptTransformation => true
case _: Sort => true
case _: BatchEvalPython => true
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1407,4 +1407,12 @@ class FilterPushdownSuite extends PlanTest {
condition = Some("x.a".attr === "z.a".attr)).analyze
comparePlans(optimized, correctAnswer)
}

test("SPARK-37828: Push down filters through RebalancePartitions") {
val originalQuery = RebalancePartitions(Seq.empty, testRelation).where('a > 3)
val optimized = Optimize.execute(originalQuery.analyze)

val correctAnswer = RebalancePartitions(Seq.empty, testRelation.where('a > 3)).analyze
comparePlans(optimized, correctAnswer)
}
}

0 comments on commit 6d8426e

Please sign in to comment.