[WIP][SPARK-19712][SQL] Move PullupCorrelatedPredicates and RewritePredicateSubquery after OptimizeSubqueries

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/subquery.scala

@@ -245,6 +245,17 @@ object ScalarSubquery {
       case _ => false
     }.isDefined
   }
+
+  def hasScalarSubquery(e: Expression): Boolean = {
+    e.find {
+      case s: ScalarSubquery => true
+      case _ => false
+    }.isDefined
+  }
+
+  def hasScalarSubquery(e: Seq[Expression]): Boolean = {
+    e.find(hasScalarSubquery(_)).isDefined
+  }
 }
 
 /**

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

@@ -119,7 +119,11 @@ case class EliminateOuterJoin(conf: SQLConf) extends Rule[LogicalPlan] with Pred
    * Returns whether the expression returns null or false when all inputs are nulls.
    */
   private def canFilterOutNull(e: Expression): Boolean = {
-    if (!e.deterministic || SubqueryExpression.hasCorrelatedSubquery(e)) return false
+    if (!e.deterministic ||
+        SubqueryExpression.hasCorrelatedSubquery(e) ||
+        SubExprUtils.containsOuter(e)) {
+      return false
+    }
     val attributes = e.references.toSeq
     val emptyRow = new GenericInternalRow(attributes.length)
     val boundE = BindReferences.bindReference(e, attributes)
@@ -147,9 +151,42 @@ case class EliminateOuterJoin(conf: SQLConf) extends Rule[LogicalPlan] with Pred
     }
   }
 
+  private def buildNewJoinType(upperJoin: Join, lowerJoin: Join, otherTableOutput: AttributeSet):
+    JoinType = {
+    val conditions = upperJoin.constraints
+    // Find the predicates reference only on the other table.
+    val localConditions = conditions.filter(_.references.subsetOf(otherTableOutput))
+    // Find the predicates reference either the left table or the join predicates
+    // between the left table and the other table.
+    val leftConditions = conditions.filter(_.references.
+      subsetOf(lowerJoin.left.outputSet ++ otherTableOutput)).diff(localConditions)
+    // Find the predicates reference either the right table or the join predicates
+    // between the right table and the other table.
+    val rightConditions = conditions.filter(_.references.
+      subsetOf(lowerJoin.right.outputSet ++ otherTableOutput)).diff(localConditions)
+
+    val leftHasNonNullPredicate = leftConditions.exists(canFilterOutNull)
+    val rightHasNonNullPredicate = rightConditions.exists(canFilterOutNull)
+
+    lowerJoin.joinType match {
+      case RightOuter if leftHasNonNullPredicate => Inner
+      case LeftOuter if rightHasNonNullPredicate => Inner
+      case FullOuter if leftHasNonNullPredicate && rightHasNonNullPredicate => Inner
+      case FullOuter if leftHasNonNullPredicate => LeftOuter
+      case FullOuter if rightHasNonNullPredicate => RightOuter
+      case o => o
+    }
+  }
+
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case f @ Filter(condition, j @ Join(_, _, RightOuter | LeftOuter | FullOuter, _)) =>
       val newJoinType = buildNewJoinType(f, j)
       if (j.joinType == newJoinType) f else Filter(condition, j.copy(joinType = newJoinType))
+    case j @ Join(child @ Join(_, _, RightOuter | LeftOuter | FullOuter, _),
+      subquery, LeftSemiOrAnti(joinType), joinCond) =>
+      val newJoinType = buildNewJoinType(j, child, subquery.outputSet)
+      if (newJoinType == child.joinType) j else {
+        Join(child.copy(joinType = newJoinType), subquery, joinType, joinCond)
+      }
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala

@@ -112,3 +112,10 @@ object LeftExistence {
     case _ => None
   }
 }
+
+object LeftSemiOrAnti {
+  def unapply(joinType: JoinType): Option[JoinType] = joinType match {
+    case LeftSemi | LeftAnti => Some(joinType)
+    case _ => None
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -319,7 +319,7 @@ case class Join(
         left.constraints
           .union(right.constraints)
           .union(splitConjunctivePredicates(condition.get).toSet)
-      case LeftSemi if condition.isDefined =>
+      case LeftSemi | LeftAnti if condition.isDefined =>
         left.constraints
           .union(splitConjunctivePredicates(condition.get).toSet)
       case j: ExistenceJoin =>

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

@@ -33,11 +33,14 @@ class FilterPushdownSuite extends PlanTest {
     val batches =
       Batch("Subqueries", Once,
         EliminateSubqueryAliases) ::
+      Batch("Subquery", Once,
+        RewritePredicateSubquery) ::
       Batch("Filter Pushdown", FixedPoint(10),
         CombineFilters,
         PushDownPredicate,
         BooleanSimplification,
         PushPredicateThroughJoin,
+        PushLeftSemiLeftAntiThroughJoin,
         CollapseProject) :: Nil
   }
 
@@ -855,8 +858,9 @@ class FilterPushdownSuite extends PlanTest {
       .where(Exists(z.where("x.a".attr === "z.a".attr)))
       .join(y, Inner, Option("x.a".attr === "y.a".attr))
       .analyze
-    val optimized = Optimize.execute(Optimize.execute(query))
-    comparePlans(optimized, answer)
+    val optimized = Optimize.execute(query)
+    val expected = Optimize.execute(answer)
+    comparePlans(optimized, expected)
   }
 
   test("predicate subquery: push down complex") {
@@ -875,8 +879,9 @@ class FilterPushdownSuite extends PlanTest {
       .join(x, Inner, Option("w.a".attr === "x.a".attr))
       .join(y, LeftOuter, Option("x.a".attr === "y.a".attr))
       .analyze
-    val optimized = Optimize.execute(Optimize.execute(query))
-    comparePlans(optimized, answer)
+    val optimized = Optimize.execute(query)
+    val expected = Optimize.execute(answer)
+    comparePlans(optimized, expected)
   }
 
   test("SPARK-20094: don't push predicate with IN subquery into join condition") {
@@ -890,13 +895,14 @@ class FilterPushdownSuite extends PlanTest {
         ("x.a".attr > 1 || "z.c".attr.in(ListQuery(w.select("w.d".attr)))))
       .analyze
 
-    val expectedPlan = x
+    val answer = x
       .join(z, Inner, Some("x.b".attr === "z.b".attr))
       .where("x.a".attr > 1 || "z.c".attr.in(ListQuery(w.select("w.d".attr))))
       .analyze
 
     val optimized = Optimize.execute(queryPlan)
-    comparePlans(optimized, expectedPlan)
+    val expected = Optimize.execute(answer)
+    comparePlans(optimized, expected)
   }
 
   test("Window: predicate push down -- basic") {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/PlanTest.scala

@@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
 
 /**
  * Provides helper methods for comparing plans.
@@ -71,7 +72,11 @@ abstract class PlanTest extends SparkFunSuite with PredicateHelper {
         val newCondition =
           splitConjunctivePredicates(condition.get).map(rewriteEqual(_)).sortBy(_.hashCode())
             .reduce(And)
-        Join(left, right, joinType, Some(newCondition))
+        val maskedJoinType = if (joinType.isInstanceOf[ExistenceJoin]) {
+          val exists = AttributeReference("exists", BooleanType, false)(exprId = ExprId(0))
+          ExistenceJoin(exists)
+        } else joinType
+        Join(left, right, maskedJoinType, Some(newCondition))
     }
   }