Add logical to support pushdown the join filter

Author: chenghao-intel

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -260,13 +260,14 @@ object PushPredicateThroughProject extends Rule[LogicalPlan] {
  * evaluated using only the attributes of the left or right side of a join.  Other
  * [[catalyst.plans.logical.Filter Filter]] conditions are moved into the `condition` of the
  * [[catalyst.plans.logical.Join Join]].
+ * And also Pushes down the join filter, where the `condition` can be evaluated using only the 
+ * attributes of the left or right side of sub query when applicable. 
  * 
- * The basic condition (JoinCondition & Filter) push down should conform to the rule:
- * https://cwiki.apache.org/confluence/display/Hive/OuterJoinBehavior
+ * Check https://cwiki.apache.org/confluence/display/Hive/OuterJoinBehavior for more details
  */
 object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
   // split the condition expression into 3 parts, 
-  // (canEvaluateInLeft, canEvaluateInRight, haveToEvaluateInBoth) 
+  // (canEvaluateInLeftSide, canEvaluateInRightSide, haveToEvaluateWithBothSide) 
   private def split(condition: Seq[Expression], left: LogicalPlan, right: LogicalPlan) = {
     val (leftEvaluateCondition, rest) =
         condition.partition(_.references subsetOf left.outputSet)
@@ -277,53 +278,76 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
   }
 
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    // push the where condition down into join filter
     case f @ Filter(filterCondition, Join(left, right, joinType, joinCondition)) =>
-      val splittedFilterCondition = splitConjunctivePredicates(filterCondition)
-      val splittedJoinCondition = joinCondition.map(splitConjunctivePredicates).getOrElse(Nil)
-
-      // Split the predicates into those that can be evaluated on the left, right, and those that
-      // must be evaluated after the join.
       val (leftFilterConditions, rightFilterConditions, commonFilterCondition) = 
-        split(splittedFilterCondition, left, right)
-      val (leftJoinConditions, rightJoinConditions, commonJoinCondition) = 
-        split(splittedJoinCondition, left, right)
+        split(splitConjunctivePredicates(filterCondition), left, right)
 
       joinType match {
         case Inner =>
-          // Treat the Condition / Filter in the same way
-          val newLeft = (leftFilterConditions ++ leftJoinConditions).
+          // push down the single side `where` condition into respective sides
+          val newLeft = leftFilterConditions.
             reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
-          val newRight = (rightFilterConditions ++ rightJoinConditions).
+          val newRight = rightFilterConditions.
             reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
-          val newJoinCond = (commonFilterCondition ++ commonJoinCondition).reduceLeftOption(And)
+          val newJoinCond = (commonFilterCondition ++ joinCondition).reduceLeftOption(And)
 
           Join(newLeft, newRight, Inner, newJoinCond)
         case RightOuter =>
-          // Push Down the Right Only Filter & Push Down the Left Only Join Condition
-          val newLeft = leftJoinConditions.
-            reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
+          // push down the right side only `where` condition
+          val newLeft = left
           val newRight = rightFilterConditions.
             reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
-          val newJoinCond = (rightJoinConditions ++ commonJoinCondition).reduceLeftOption(And)
+          val newJoinCond = joinCondition
           val newJoin = Join(newLeft, newRight, RightOuter, newJoinCond)
 
           (leftFilterConditions ++ commonFilterCondition).
             reduceLeftOption(And).map(Filter(_, newJoin)).getOrElse(newJoin)
         case LeftOuter =>
-          // Push Down the Left Only Filter & Push Down the Right Only Join Condition
+          // push down the left side only `where` condition
           val newLeft = leftFilterConditions.
             reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
-          val newRight = rightJoinConditions.
-            reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
-          val newJoinCond = (leftJoinConditions ++ commonJoinCondition).reduceLeftOption(And)
+          val newRight = right
+          val newJoinCond = joinCondition
           val newJoin = Join(newLeft, newRight, LeftOuter, newJoinCond)
 
           (rightFilterConditions ++ commonFilterCondition).
             reduceLeftOption(And).map(Filter(_, newJoin)).getOrElse(newJoin)
-        case FullOuter =>
-          // DO Nothing for Full Outer Join
+        case FullOuter => f // DO Nothing for Full Outer Join
+      }
+
+    // push down the join filter into sub query scanning if applicable
+    case f @ Join(left, right, joinType, joinCondition) =>
+      val (leftJoinConditions, rightJoinConditions, commonJoinCondition) = 
+        split(joinCondition.map(splitConjunctivePredicates).getOrElse(Nil), left, right)
+
+      joinType match {
+        case Inner =>
+          // push down the single side only join filter for both sides sub queries
+          val newLeft = leftJoinConditions.
+            reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
+          val newRight = rightJoinConditions.
+            reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
+          val newJoinCond = commonJoinCondition.reduceLeftOption(And)
+
+          Join(newLeft, newRight, Inner, newJoinCond)
+        case RightOuter =>
+          // push down the left side only join filter for left side sub query
+          val newLeft = leftJoinConditions.
+            reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
+          val newRight = right
+          val newJoinCond = (rightJoinConditions ++ commonJoinCondition).reduceLeftOption(And)
+
+          Join(newLeft, newRight, RightOuter, newJoinCond)
+        case LeftOuter =>
+          // push down the right side only join filter for right sub query
+          val newLeft = left
+          val newRight = rightJoinConditions.
+            reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
+          val newJoinCond = (leftJoinConditions ++ commonJoinCondition).reduceLeftOption(And)
 
-          f
+          Join(newLeft, newRight, LeftOuter, newJoinCond)
+        case FullOuter => f
       }
   }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala

@@ -20,11 +20,14 @@ package org.apache.spark.sql.catalyst.optimizer
 import org.apache.spark.sql.catalyst.analysis
 import org.apache.spark.sql.catalyst.analysis.EliminateAnalysisOperators
 import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.catalyst.plans.FullOuter
+import org.apache.spark.sql.catalyst.plans.LeftOuter
+import org.apache.spark.sql.catalyst.plans.RightOuter
 import org.apache.spark.sql.catalyst.rules._
-
-/* Implicit conversions */
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.junit.Test
 
 class FilterPushdownSuite extends OptimizerTest {
 
@@ -161,6 +164,184 @@ class FilterPushdownSuite extends OptimizerTest {
 
     comparePlans(optimized, correctAnswer)
   }
+  
+  test("joins: push down left outer join #1") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, LeftOuter)
+        .where("x.b".attr === 1 && "y.b".attr === 2)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.where('b === 1)
+    val correctAnswer =
+      left.join(y, LeftOuter).where("y.b".attr === 2).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down right outer join #1") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, RightOuter)
+        .where("x.b".attr === 1 && "y.b".attr === 2)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val right = testRelation.where('b === 2).subquery('d)
+    val correctAnswer =
+      x.join(right, RightOuter).where("x.b".attr === 1).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down left outer join #2") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, LeftOuter, Some("x.b".attr === 1))
+        .where("x.b".attr === 2 && "y.b".attr === 2)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.where('b === 2).subquery('d)
+    val correctAnswer =
+      left.join(y, LeftOuter, Some("d.b".attr === 1)).where("y.b".attr === 2).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down right outer join #2") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, RightOuter, Some("y.b".attr === 1))
+        .where("x.b".attr === 2 && "y.b".attr === 2)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val right = testRelation.where('b === 2).subquery('d)
+    val correctAnswer =
+      x.join(right, RightOuter, Some("d.b".attr === 1)).where("x.b".attr === 2).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down left outer join #3") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, LeftOuter, Some("y.b".attr === 1))
+        .where("x.b".attr === 2 && "y.b".attr === 2)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.where('b === 2).subquery('l)
+    val right = testRelation.where('b === 1).subquery('r)
+    val correctAnswer =
+      left.join(right, LeftOuter).where("r.b".attr === 2).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down right outer join #3") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, RightOuter, Some("y.b".attr === 1))
+        .where("x.b".attr === 2 && "y.b".attr === 2)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val right = testRelation.where('b === 2).subquery('r)
+    val correctAnswer =
+      x.join(right, RightOuter, Some("r.b".attr === 1)).where("x.b".attr === 2).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down left outer join #4") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, LeftOuter, Some("y.b".attr === 1))
+        .where("x.b".attr === 2 && "y.b".attr === 2 && "x.c".attr === "y.c".attr)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.where('b === 2).subquery('l)
+    val right = testRelation.where('b === 1).subquery('r)
+    val correctAnswer =
+      left.join(right, LeftOuter).where("r.b".attr === 2 && "l.c".attr === "r.c".attr).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down right outer join #4") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, RightOuter, Some("y.b".attr === 1))
+        .where("x.b".attr === 2 && "y.b".attr === 2 && "x.c".attr === "y.c".attr)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.subquery('l)
+    val right = testRelation.where('b === 2).subquery('r)
+    val correctAnswer =
+      left.join(right, RightOuter, Some("r.b".attr === 1)).
+        where("l.b".attr === 2 && "l.c".attr === "r.c".attr).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down left outer join #5") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, LeftOuter, Some("y.b".attr === 1 && "x.a".attr === 3))
+        .where("x.b".attr === 2 && "y.b".attr === 2 && "x.c".attr === "y.c".attr)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.where('b === 2).subquery('l)
+    val right = testRelation.where('b === 1).subquery('r)
+    val correctAnswer =
+      left.join(right, LeftOuter, Some("l.a".attr===3)).
+        where("r.b".attr === 2 && "l.c".attr === "r.c".attr).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("joins: push down right outer join #5") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+
+    val originalQuery = {
+      x.join(y, RightOuter, Some("y.b".attr === 1 && "x.a".attr === 3))
+        .where("x.b".attr === 2 && "y.b".attr === 2 && "x.c".attr === "y.c".attr)
+    }
+
+    val optimized = Optimize(originalQuery.analyze)
+    val left = testRelation.where('a === 3).subquery('l)
+    val right = testRelation.where('b === 2).subquery('r)
+    val correctAnswer =
+      left.join(right, RightOuter, Some("r.b".attr === 1)).
+        where("l.b".attr === 2 && "l.c".attr === "r.c".attr).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
 
   test("joins: can't push down") {
     val x = testRelation.subquery('x)