outer join elimination by parent join.

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

@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.analysis.{CleanupAliases, EliminateSubQueri
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.planning.ExtractFiltersAndInnerJoins
-import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftOuter, LeftSemi, RightOuter}
+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.types._
@@ -44,6 +44,7 @@ abstract class Optimizer extends RuleExecutor[LogicalPlan] {
       // Operator push down
       SetOperationPushDown,
       SamplePushDown,
+      OuterJoinElimination,
       ReorderJoin,
       PushPredicateThroughJoin,
       PushPredicateThroughProject,
@@ -768,6 +769,107 @@ object ReorderJoin extends Rule[LogicalPlan] with PredicateHelper {
   }
 }
 
+/**
+ * Elimination of Outer Join by Parent Join Condition
+ *
+ * Given an outer join is involved in another join (called parent join), when the join type of the
+ * parent join is inner, left-semi, left-outer and right-outer, checking if the join condition of
+ * the parent join satisfies the following two conditions:
+ *
+ * 1) there exist null filtering predicates against the columns in the null-supplying side of
+ *    parent join.
+ * 2) these columns are from the child join.
+ *
+ * If having such join predicates, execute the elimination rules:
+ * - full outer -> inner if both sides of the child join have such predicates
+ * - left outer -> inner if the right side of the child join has such predicates
+ * - right outer -> inner if the left side of the child join has such predicates
+ * - full outer -> left outer if only the left side of the child join has such predicates
+ * - full outer -> right outer if only the right side of the child join has such predicates
+ *
+ */
+object OuterJoinElimination extends Rule[LogicalPlan] with PredicateHelper {
+
+  private def containsAttr(plan: LogicalPlan, attr: Attribute): Boolean =
+    plan.outputSet.exists(_.semanticEquals(attr))
+
+  private def hasNullFilteringPredicate(predicate: Expression, plan: LogicalPlan): Boolean = {
+    predicate match {
+      case EqualTo(ar: AttributeReference, _) if containsAttr(plan, ar) => true
+      case EqualTo(_, ar: AttributeReference) if containsAttr(plan, ar) => true
+      case EqualNullSafe(ar: AttributeReference, l)
+        if !l.nullable && containsAttr(plan, ar) => true
+      case EqualNullSafe(l, ar: AttributeReference)
+        if !l.nullable && containsAttr(plan, ar) => true
+      case GreaterThan(ar: AttributeReference, _) if containsAttr(plan, ar) => true
+      case GreaterThan(_, ar: AttributeReference) if containsAttr(plan, ar) => true
+      case GreaterThanOrEqual(ar: AttributeReference, _) if containsAttr(plan, ar) => true
+      case GreaterThanOrEqual(_, ar: AttributeReference) if containsAttr(plan, ar) => true
+      case LessThan(ar: AttributeReference, _) if containsAttr(plan, ar) => true
+      case LessThan(_, ar: AttributeReference) if containsAttr(plan, ar) => true
+      case LessThanOrEqual(ar: AttributeReference, _) if containsAttr(plan, ar) => true
+      case LessThanOrEqual(_, ar: AttributeReference) if containsAttr(plan, ar) => true
+      case In(ar: AttributeReference, _) if containsAttr(plan, ar) => true
+      case IsNotNull(ar: AttributeReference) if containsAttr(plan, ar) => true
+      case And(l, r) => hasNullFilteringPredicate(l, plan) || hasNullFilteringPredicate(r, plan)
+      case Or(l, r) => hasNullFilteringPredicate(l, plan) && hasNullFilteringPredicate(r, plan)
+      case Not(e) => !hasNullFilteringPredicate(e, plan)
+      case _ => false
+    }
+  }
+
+  private def buildNewJoin(
+      otherCondition: Expression,
+      left: LogicalPlan,
+      right: LogicalPlan,
+      joinType: JoinType,
+      condition: Option[Expression]): Join = {
+    val leftHasNonNullPredicate = hasNullFilteringPredicate(otherCondition, left)
+    val rightHasNonNullPredicate = hasNullFilteringPredicate(otherCondition, right)
+
+    joinType match {
+      case RightOuter if leftHasNonNullPredicate =>
+        Join(left, right, Inner, condition)
+      case LeftOuter if rightHasNonNullPredicate =>
+        Join(left, right, Inner, condition)
+      case FullOuter if leftHasNonNullPredicate && rightHasNonNullPredicate =>
+        Join(left, right, Inner, condition)
+      case FullOuter if leftHasNonNullPredicate =>
+        Join(left, right, LeftOuter, condition)
+      case FullOuter if rightHasNonNullPredicate =>
+        Join(left, right, RightOuter, condition)
+      case _ => Join(left, right, joinType, condition)
+    }
+  }
+
+  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+
+    // Case 1: when parent join is Inner|LeftSemi|LeftOuter and the child join is on the right side
+    case pj @ Join(
+        pLeft,
+        j @ Join(left, right, RightOuter|LeftOuter|FullOuter, condition),
+        Inner|LeftSemi|LeftOuter,
+        Some(pJoinCond)) =>
+      Join(
+          pLeft,
+          buildNewJoin(pJoinCond, left, right, j.joinType, condition),
+          pj.joinType,
+          Some(pJoinCond))
+
+    // Case 2: when parent join is Inner|LeftSemi|RightOuter and the child join is on the left side
+    case pj @ Join(
+        j @ Join(left, right, RightOuter|LeftOuter|FullOuter, condition),
+        pRight,
+        Inner|LeftSemi|RightOuter,
+        Some(pJoinCond)) =>
+      Join(
+          buildNewJoin(pJoinCond, left, right, j.joinType, condition),
+          pRight,
+          pj.joinType,
+          Some(pJoinCond))
+  }
+}
+
 /**
  * Pushes down [[Filter]] operators where the `condition` can be
  * evaluated using only the attributes of the left or right side of a join.  Other

sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql
 
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.logical.{Project, Join}
 import org.apache.spark.sql.execution.joins.BroadcastHashJoin
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.test.SharedSQLContext
@@ -140,4 +142,157 @@ class DataFrameJoinSuite extends QueryTest with SharedSQLContext {
       assert(df1.join(broadcast(pf1)).count() === 4)
     }
   }
+
+  test("join - left outer to inner by the parent join's join condition") {
+    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as("a")
+    val df2 = Seq((1, 3, "1"), (5, 6, "5")).toDF("int", "int2", "str").as("b")
+    val df3 = Seq((1, 3, "1"), (3, 6, "5")).toDF("int", "int2", "str").as("c")
+
+    // Left -> Inner
+    val right = df.join(df2, $"a.int" === $"b.int", "left")
+    val left2Inner =
+      df3.join(right, $"c.int" === $"b.int", "inner").select($"a.*", $"b.*", $"c.*")
+
+    left2Inner.explain(true)
+
+    // The order before conversion: Left Then Inner
+    assert(left2Inner.queryExecution.analyzed.collect {
+      case j@Join(_, Join(_, _, LeftOuter, _), Inner, _) => j
+    }.size === 1)
+
+    // The order after conversion: Inner Then Inner
+    assert(left2Inner.queryExecution.optimizedPlan.collect {
+      case j@Join(_, Join(_, _, Inner, _), Inner, _) => j
+    }.size === 1)
+
+    checkAnswer(
+      left2Inner,
+      Row(1, 2, "1", 1, 3, "1", 1, 3, "1") :: Nil)
+  }
+
+  test("join - right outer to inner by the parent join's join condition") {
+    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as("a")
+    val df2 = Seq((1, 3, "1"), (5, 6, "5")).toDF("int", "int2", "str").as("b")
+    val df3 = Seq((1, 9, "8"), (5, 0, "4")).toDF("int", "int2", "str").as("c")
+
+    // Right Then Inner -> Inner Then Right
+    val right2Inner = df.join(df2, $"a.int" === $"b.int", "right")
+      .join(df3, $"a.int" === $"b.int", "inner").select($"a.*", $"b.*", $"c.*")
+
+    // The order before conversion: Left Then Inner
+    assert(right2Inner.queryExecution.analyzed.collect {
+      case j@Join(Join(_, _, RightOuter, _), _, Inner, _) => j
+    }.size === 1)
+
+    // The order after conversion: Inner Then Inner
+    assert(right2Inner.queryExecution.optimizedPlan.collect {
+      case j@Join(Join(_, _, Inner, _), _, Inner, _) => j
+    }.size === 1)
+
+    checkAnswer(
+      right2Inner,
+      Row(1, 2, "1", 1, 3, "1", 1, 9, "8") ::
+      Row(1, 2, "1", 1, 3, "1", 5, 0, "4") :: Nil)
+  }
+
+  test("join - full outer to inner by the parent join's join condition") {
+    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as("a")
+    val df2 = Seq((1, 2, "1"), (5, 6, "5")).toDF("int", "int2", "str").as("b")
+    val df3 = Seq((1, 3, "1"), (3, 6, "5")).toDF("int", "int2", "str").as("c")
+
+    // Full -> Inner
+    val right = df.join(df2, $"a.int" === $"b.int", "full")
+    val full2Inner = df3.join(right, $"c.int" === $"a.int" && $"b.int" === 1, "inner")
+      .select($"a.*", $"b.*", $"c.*")
+
+    // The order before conversion: Left Then Inner
+    assert(full2Inner.queryExecution.analyzed.collect {
+      case j@Join(_, Join(_, _, FullOuter, _), Inner, _) => j
+    }.size === 1)
+
+    // The order after conversion: Inner Then Inner
+    assert(full2Inner.queryExecution.optimizedPlan.collect {
+      case j@Join(_, Join(_, _, Inner, _), Inner, _) => j
+    }.size === 1)
+
+    checkAnswer(
+      full2Inner,
+      Row(1, 2, "1", 1, 2, "1", 1, 3, "1") :: Nil)
+  }
+
+  test("join - full outer to right by the parent join's join condition") {
+    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as("a")
+    val df2 = Seq((1, 2, "1"), (5, 6, "5")).toDF("int", "int2", "str").as("b")
+    val df3 = Seq((1, 3, "1"), (3, 6, "5")).toDF("int", "int2", "str").as("c")
+
+    // Full -> Right
+    val right = df.join(df2, $"a.int" === $"b.int", "full")
+    val full2Right = df3.join(right, $"b.int" === 1, "leftsemi")
+
+    // The order before conversion: Left Then Inner
+    assert(full2Right.queryExecution.analyzed.collect {
+      case j@Join(_, Join(_, _, FullOuter, _), LeftSemi, _) => j
+    }.size === 1)
+
+    // The order after conversion: Inner Then Inner
+    assert(full2Right.queryExecution.optimizedPlan.collect {
+      case j@Join(_, Project(_, Join(_, _, RightOuter, _)), LeftSemi, _) => j
+    }.size === 1)
+
+    checkAnswer(
+      full2Right,
+      Row(1, 3, "1") :: Row(3, 6, "5") :: Nil)
+  }
+
+
+  test("join - full outer to left by the parent join's join condition #1") {
+    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as("a")
+    val df2 = Seq((1, 2, "1"), (5, 6, "5")).toDF("int", "int2", "str").as("b")
+    val df3 = Seq((1, 3, "1"), (4, 6, "5")).toDF("int", "int2", "str").as("c")
+
+    // Full -> Left
+    val right = df.join(df2, $"a.int" === $"b.int", "full")
+    val full2Left = df3.join(right, $"c.int" === $"a.int", "left")
+      .select($"a.*", $"b.*", $"c.*")
+
+    // The order before conversion: Full Then Left
+    assert(full2Left.queryExecution.analyzed.collect {
+      case j@Join(_, Join(_, _, FullOuter, _), LeftOuter, _) => j
+    }.size === 1)
+
+    // The order after conversion: Left Then Left
+    assert(full2Left.queryExecution.optimizedPlan.collect {
+      case j@Join(_, Join(_, _, LeftOuter, _), LeftOuter, _) => j
+    }.size === 1)
+
+    checkAnswer(
+      full2Left,
+      Row(1, 2, "1", 1, 2, "1", 1, 3, "1") ::
+      Row(null, null, null, null, null, null, 4, 6, "5") :: Nil)
+  }
+
+  test("join - full outer to left by the parent join's join condition #2") {
+    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as("a")
+    val df2 = Seq((1, 2, "1"), (5, 6, "5")).toDF("int", "int2", "str").as("b")
+    val df3 = Seq((1, 3, "1"), (4, 6, "5")).toDF("int", "int2", "str").as("c")
+
+    // Full -> Left
+    val full2Left = df.join(df2, $"a.int" === $"b.int", "full")
+      .join(df3, $"c.int" === $"a.int", "right").select($"a.*", $"b.*", $"c.*")
+
+    // The order before conversion: Full Then Right
+    assert(full2Left.queryExecution.analyzed.collect {
+      case j@Join(Join(_, _, FullOuter, _), _, RightOuter, _) => j
+    }.size === 1)
+
+    // The order after conversion: Left Then Right
+    assert(full2Left.queryExecution.optimizedPlan.collect {
+      case j@Join(Join(_, _, LeftOuter, _), _, RightOuter, _) => j
+    }.size === 1)
+
+    checkAnswer(
+      full2Left,
+      Row(1, 2, "1", 1, 2, "1", 1, 3, "1") ::
+      Row(null, null, null, null, null, null, 4, 6, "5") :: Nil)
+  }
 }