[SPARK-7824][SQL]Collapsing operator reordering and constant folding into a single batch to push down the single side.

[pzzs]

SQL

select * from tableA join tableB on (a > 3 and b = d) or (a > 3 and b = e)

Plan before modify

== Optimized Logical Plan ==
Project [a#293,b#294,c#295,d#296,e#297]
 Join Inner, Some(((a#293 > 3) && ((b#294 = d#296) || (b#294 = e#297))))
  MetastoreRelation default, tablea, None
  MetastoreRelation default, tableb, None

Plan after modify

== Optimized Logical Plan ==
Project [a#293,b#294,c#295,d#296,e#297]
 Join Inner, Some(((b#294 = d#296) || (b#294 = e#297)))
  Filter (a#293 > 3)
   MetastoreRelation default, tablea, None
  MetastoreRelation default, tableb, None

CombineLimits ==> Limit(If(LessThan(ne, le), ne, le), grandChild) and LessThan is in BooleanSimplification , so CombineLimits must before BooleanSimplification and BooleanSimplification must before PushPredicateThroughJoin.

[pzzs]

This optimizer can void CartesianProduct
Tables

tableA        tableB        tableC
a   int       c    int      f    int    
b   int       d    int      g    int
              e    int

SQL

select * from tableA, tableB, tableC where (d = f and  a = c and b = d) or (d = f and a = c and b = e)

Plan before modify

== Optimized Logical Plan ==
Project [a#306,b#307,c#308,d#309,e#310,f#311,g#312]
 Join Inner, Some((((d#309 = f#311) && (a#306 = c#308)) && ((b#307 = d#309) || (b#307 = e#310))))
  Join Inner, None
   MetastoreRelation default, tablea, None
   MetastoreRelation default, tableb, None
  MetastoreRelation default, tablec, None

== Physical Plan ==
Project [a#306,b#307,c#308,d#309,e#310,f#311,g#312]
 Filter ((a#306 = c#308) && ((b#307 = d#309) || (b#307 = e#310)))
  ShuffledHashJoin [d#309], [f#311], BuildRight
   Exchange (HashPartitioning 5), []
    CartesianProduct
     HiveTableScan [a#306,b#307], (MetastoreRelation default, tablea, None), None
     HiveTableScan [c#308,d#309,e#310], (MetastoreRelation default, tableb, None), None
   Exchange (HashPartitioning 5), []
    HiveTableScan [f#311,g#312], (MetastoreRelation default, tablec, None), None

Plan after modify

== Optimized Logical Plan ==
Project [a#306,b#307,c#308,d#309,e#310,f#311,g#312]
 Join Inner, Some((d#309 = f#311))
  Join Inner, Some(((a#306 = c#308) && ((b#307 = d#309) || (b#307 = e#310))))
   MetastoreRelation default, tablea, None
   MetastoreRelation default, tableb, None
  MetastoreRelation default, tablec, None

== Physical Plan ==
Project [a#306,b#307,c#308,d#309,e#310,f#311,g#312]
 ShuffledHashJoin [d#309], [f#311], BuildRight
  Exchange (HashPartitioning 5), []
   Filter ((b#307 = d#309) || (b#307 = e#310))
    ShuffledHashJoin [a#306], [c#308], BuildRight
     Exchange (HashPartitioning 5), []
      HiveTableScan [a#306,b#307], (MetastoreRelation default, tablea, None), None
     Exchange (HashPartitioning 5), []
      HiveTableScan [c#308,d#309,e#310], (MetastoreRelation default, tableb, None), None
  Exchange (HashPartitioning 5), []
   HiveTableScan [f#311,g#312], (MetastoreRelation default, tablec, None), None

[pzzs]

@marmbrus @scwf

[scwf]

Get it. The root cause here is we moved Filter Pushdown before ConstantFolding in 4459514, then we can not simplify boolean condition before filter push down through join, which leads to CartesianProduct.

i think this change is reasonable.

[marmbrus]

Would you get the same result (and possibly more) by collapsing operator reordering and constant folding into a single batch? This seems like a very one-off fix to me.

[pzzs]

yes, it can work. but this batch will have two different types of optimizers. @marmbrus

[marmbrus]

That is not fundamentally a problem. Honestly some more thought probably needs to be put into the batches. Really the only reasons for splitting are the following:

• Large batches are inherently more costly as you must go through every rule, even if only a small number are making changes. So if rules will never interact they can be in separate batches.
• However, large batches are more powerful as there is more opportunity for rules to interact
• Its possible for rules to undo the result of other rules. In this case they must be in separate batches or it will go back and forth till the limit is reached
• Another reason for batches is satisfying preconditions. (i.e. a plan must be analyzed before optimizing it).

[pzzs]

@marmbrus /cc

[marmbrus]

ok to test

[SparkQA]

Test build #33867 has finished for PR 6351 at commit a04ffae.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[scwf]

how about "Operator Reordering and ConstantFolding"

[SparkQA]

Test build #33869 has finished for PR 6351 at commit ae3af6d.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #33875 has finished for PR 6351 at commit f8b9314.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[pzzs]

@marmbrus

[pzzs]

@marmbrus @yhuai @scwf /cc

[marmbrus]

ok to test

[SparkQA]

Test build #34717 has finished for PR 6351 at commit 20de7be.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[marmbrus]

Thanks! Merged to master.

[cloud-fan]

I think we don't need to change the rule order inside a batch with fixed point. Rules within a single batch shouldn't be sensitive to execution order.