Skip to content

[SPARK-33302][SQL] Push down filters through Expand #30278

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
AngersZhuuuu wants to merge 4 commits into from
Closed

[SPARK-33302][SQL] Push down filters through Expand #30278

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c3ec848
[SPARK-33302][SQL] Failed to push down filters through Expand
AngersZhuuuu Nov 6, 2020
803bf0a
Update FilterPushdownSuite.scala
AngersZhuuuu Nov 9, 2020
77d6e45
Update LeftSemiAntiJoinPushDownSuite.scala
AngersZhuuuu Nov 9, 2020
a09f836
Update FilterPushdownSuite.scala
AngersZhuuuu Nov 9, 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
1 change: 1 addition & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1269,6 +1269,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
case _: Sort => true
case _: BatchEvalPython => true
case _: ArrowEvalPython => true
case _: Expand => true
Copy link
Member

@maropu maropu Nov 9, 2020

Choose a reason for hiding this comment

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

This change affects the PushDownLeftSemiAntiJoin rule, too. So, could you add tests for the case?

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

This change affects the PushDownLeftSemiAntiJoin rule, too. So, could you add tests for the case?

Double check the case, seems current master fix this case by some pr, but 3.0 is still as jira desc.

Copy link
Contributor

@cloud-fan cloud-fan Nov 9, 2020

Choose a reason for hiding this comment

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

seems current master fix this case

What do you mean by "fix this case"?

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

seems current master fix this case

What do you mean by "fix this case"?

I have found the pr #29673
Before this pr, SQL

SELECT
       years,
       appversion,                                               
       SUM(uusers) AS users                                      
FROM   (SELECT
               Date_trunc('year', dt)          AS years,
               CASE                                              
                 WHEN h.pid = 3 THEN 'iOS'           
                 WHEN h.pid = 4 THEN 'Android'       
                 ELSE 'Other'                                    
               END                             AS viewport,      
               h.vs                            AS appversion,
               Count(DISTINCT u.uid)           AS uusers
               ,Count(DISTINCT u.suid)         AS srcusers
        FROM   t1 u                                   
               join t2 h                              
                 ON h.uid = u.uid            
        GROUP  BY 1,                                             
                  2,                                             
                  3) AS a
WHERE  viewport = 'iOS'                                          
GROUP  BY 1,                                                     
          2

Optimized plan is

== Optimized Logical Plan ==
Aggregate [years#0, appversion#2], [years#0, appversion#2, sum(uusers#3L) AS users#5L]
+- Aggregate [date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#24, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#25, vs#17], [date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#24 AS years#0, vs#17 AS appversion#2, count(if ((gid#23 = 1)) u.`uid`#26 else null) AS uusers#3L]
   +- Aggregate [date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#24, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#25, vs#17, u.`uid`#26, u.`suid`#27, gid#23], [date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#24, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#25, vs#17, u.`uid`#26, gid#23]
      +- Filter (CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#25 = iOS)
         +- Expand [ArrayBuffer(date_trunc(year, cast(dt#14 as timestamp), Some(Asia/Shanghai)), CASE WHEN (pid#16 = 3) THEN iOS WHEN (pid#16 = 4) THEN Android ELSE Other END, vs#17, uid#12, null, 1), ArrayBuffer(date_trunc(year, cast(dt#14 as timestamp), Some(Asia/Shanghai)), CASE WHEN (pid#16 = 3) THEN iOS WHEN (pid#16 = 4) THEN Android ELSE Other END, vs#17, null, suid#15, 2)], [date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#24, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#25, vs#17, u.`uid`#26, u.`suid`#27, gid#23]
            +- Project [uid#12, dt#14, suid#15, pid#16, vs#17]
               +- Join Inner, (uid#18 = uid#12)
                  :- Project [uid#12, dt#14, suid#15]
                  :  +- Filter isnotnull(uid#12)
                  :     +- Relation[pid#11,uid#12,sid#13,dt#14,suid#15] parquet
                  +- Project [pid#16, vs#17, uid#18]
                     +- Filter isnotnull(uid#18)
                        +- Relation[pid#16,vs#17,uid#18,csid#19] parquet

After that pr, Optimized plan is

== Optimized Logical Plan ==
Aggregate [years#0, appversion#2], [years#0, appversion#2, sum(uusers#3L) AS users#5L]
+- Aggregate [date_trunc(year, cast(dt#14 as timestamp), Some(Asia/Shanghai)), CASE WHEN (pid#16 = 3) THEN iOS WHEN (pid#16 = 4) THEN Android ELSE Other END, vs#17], [date_trunc(year, cast(dt#14 as timestamp), Some(Asia/Shanghai)) AS years#0, vs#17 AS appversion#2, count(distinct uid#12) AS uusers#3L]
   +- Project [uid#12, dt#14, pid#16, vs#17]
      +- Join Inner, (uid#18 = uid#12)
         :- Project [uid#12, dt#14]
         :  +- Filter isnotnull(uid#12)
         :     +- Relation[pid#11,uid#12,sid#13,dt#14,suid#15] parquet
         +- Project [pid#16, vs#17, uid#18]
            +- Filter ((CASE WHEN (pid#16 = 3) THEN iOS WHEN (pid#16 = 4) THEN Android ELSE Other END = iOS) AND isnotnull(uid#18))
               +- Relation[pid#16,vs#17,uid#18,csid#19] parquet

Filter((CASE WHEN (pid#16 = 3) THEN iOS WHEN (pid#16 = 4) THEN Android ELSE Other END = iOS)) is pushed down and won't generate Expand

Copy link
Contributor

@cloud-fan cloud-fan Nov 9, 2020

Choose a reason for hiding this comment

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

how does it related to left semi join?

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

This change affects the PushDownLeftSemiAntiJoin rule, too. So, could you add tests for the case?

So can we add such a test?

With test case in LeftSemiPushdownSuite

  test("Unary: LeftSemi join push down through expand") {
    val expand = Expand(Seq(Seq('a, 'b, "null"), Seq('a, "null", 'c)),
      Seq('a, 'b, 'c), testRelation)
    val originalQuery = expand
      .join(testRelation1, joinType = LeftSemi, condition = Some('b === 'd && 'b === 1))

    val optimized = Optimize.execute(originalQuery.analyze)
    val correctAnswer = Expand(Seq(Seq('a, 'b, "null"), Seq('a, "null", 'c)),
      Seq('a, 'b, 'c), Filter(EqualTo('b, 1), testRelation))
        .join(testRelation1, joinType = LeftSemi, condition = Some('b === 'd))
        .analyze

    comparePlans(optimized, correctAnswer)
  }

originalQuery is

'Join LeftSemi, (('b = 'd) AND ('b = 1))
:- 'Expand [List('a, 'b, null), List('a, null, 'c)], ['a, 'b, 'c]
:  +- LocalRelation <empty>, [a#0, b#1, c#2]
+- LocalRelation <empty>, [d#3]

Test result is

== FAIL: Plans do not match ===
!'Expand [List(a#0, b#0, null), List(a#0, null, c#0)], [a#0, b#0, c#0]   'Join LeftSemi, (b#0 = d#0)
!+- 'Join LeftSemi, ((b#0 = 1) AND (b#0 = d#0))                          :- Expand [List(a#0, b#0, null), List(a#0, null, c#0)], [a#0, b#0, c#0]
!   :- LocalRelation <empty>, [a#0, b#0, c#0]                            :  +- Filter (b#0 = 1)
!   +- LocalRelation <empty>, [d#0]                                      :     +- LocalRelation <empty>, [a#0, b#0, c#0]
!                                                                        +- LocalRelation <empty>, [d#0]

Expand will be promoted below Join, so should we ignore this case or add a parameter in canPushThrough like below

 def canPushThrough(p: UnaryNode, isFilterPushDown: Boolean = false): Boolean = p match {
    // Note that some operators (e.g. project, aggregate, union) are being handled separately
    // (earlier in this rule).
    case _: AppendColumns => true
    case _: Distinct => true
    case _: Generate => true
    case _: Pivot => true
    case _: RepartitionByExpression => true
    case _: Repartition => true
    case _: ScriptTransformation => true
    case _: Sort => true
    case _: BatchEvalPython => true
    case _: ArrowEvalPython => true
    case _: Expand => isFilterPushDown
    case _ => false
  }

Copy link
Contributor

@cloud-fan cloud-fan Nov 9, 2020

Choose a reason for hiding this comment

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

I'm sorry I didn't get it. What's the issue here? We can't pushdown left-semi join through expand?

Copy link
Member

@maropu maropu Nov 9, 2020

Choose a reason for hiding this comment

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

The optimized (left-side) plan above looks correct to me...

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

I'm sorry I didn't get it. What's the issue here? We can't pushdown left-semi join through expand?

oh..my mistake, I misunderstood some code about PushDownLeftSemiAntiJoin

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

The optimized (left-side) plan above looks correct to me...

My fault, I misunderstand some code about PushDownLeftSemiAntiJoin, test case added ==

case _ => false
}

Expand Down
26 changes: 25 additions & 1 deletion ...catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{BooleanType, IntegerType, TimestampType}
import org.apache.spark.sql.types.{BooleanType, IntegerType, StringType, TimestampType}
import org.apache.spark.unsafe.types.CalendarInterval

class FilterPushdownSuite extends PlanTest {
Expand Down Expand Up @@ -1208,6 +1208,30 @@ class FilterPushdownSuite extends PlanTest {
checkAnalysis = false)
}


Copy link
Member

@maropu maropu Nov 9, 2020

Choose a reason for hiding this comment

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

nit: unnecessary blank.

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

nit: unnecessary blank.

Done

test("push down predicate through expand") {
val input = LocalRelation('a.int, 'b.string, 'c.double)
Copy link
Member

@maropu maropu Nov 9, 2020

Choose a reason for hiding this comment

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

nit: could you use testRelation instead?

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Nov 9, 2020

Choose a reason for hiding this comment

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

nit: could you use testRelation instead?

Done

val query =
Filter('a > 1,
Expand(
Seq(
Seq('a, 'b, 'c, Literal.create(null, StringType), 1),
Seq('a, 'b, 'c, 'a, 2)),
Seq('a, 'b, 'c),
input)).analyze
val optimized = Optimize.execute(query)

val expected =
Expand(
Seq(
Seq('a, 'b, 'c, Literal.create(null, StringType), 1),
Seq('a, 'b, 'c, 'a, 2)),
Seq('a, 'b, 'c),
Filter('a > 1, input)).analyze

comparePlans(optimized, expected)
}

test("SPARK-28345: PythonUDF predicate should be able to pushdown to join") {
val pythonUDFJoinCond = {
val pythonUDF = PythonUDF("pythonUDF", null,
Expand Down
15 changes: 15 additions & 0 deletions ...rc/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,21 @@ class LeftSemiPushdownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("Unary: LeftSemi join push down through Expand") {
val expand = Expand(Seq(Seq('a, 'b, "null"), Seq('a, "null", 'c)),
Seq('a, 'b, 'c), testRelation)
val originalQuery = expand
.join(testRelation1, joinType = LeftSemi, condition = Some('b === 'd && 'b === 1))

val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer = Expand(Seq(Seq('a, 'b, "null"), Seq('a, "null", 'c)),
Seq('a, 'b, 'c), testRelation
.join(testRelation1, joinType = LeftSemi, condition = Some('b === 'd && 'b === 1)))
.analyze

comparePlans(optimized, correctAnswer)
}

Seq(Some('d === 'e), None).foreach { case innerJoinCond =>
Seq(LeftSemi, LeftAnti).foreach { case outerJT =>
Seq(Inner, LeftOuter, Cross, RightOuter).foreach { case innerJT =>
Expand Down