Address comment. Consider more general case.

Author: viirya

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

@@ -89,8 +89,6 @@ abstract class Optimizer(sessionCatalog: SessionCatalog, conf: CatalystConf)
       CombineFilters,
       CombineLimits,
       CombineUnions,
-      // Push down Filters again after combination
-      PushDownPredicate,
       // Constant folding and strength reduction
       NullPropagation,
       FoldablePropagation,
@@ -588,15 +586,33 @@ object CombineUnions extends Rule[LogicalPlan] {
  * one conjunctive predicate.
  */
 object CombineFilters extends Rule[LogicalPlan] with PredicateHelper {
+  private def toCNF(predicate: Expression): Expression = {
+    val disjunctives = splitDisjunctivePredicates(predicate)
+    var finalPredicates = splitConjunctivePredicates(disjunctives.head)
+    disjunctives.tail.foreach { cond =>
+      val predicates = new ArrayBuffer[Expression]()
+      splitConjunctivePredicates(cond).map { p =>
+        predicates ++= finalPredicates.map(Or(_, p))
+      }
+      finalPredicates = predicates.toSeq
+    }
+    finalPredicates.reduce(And)
+  }
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case Filter(fc, nf @ Filter(nc, grandChild)) =>
-      (ExpressionSet(splitConjunctivePredicates(fc)) --
-        ExpressionSet(splitConjunctivePredicates(nc))).reduceOption(And) match {
+      val fcCNF = toCNF(fc)
+      val ncCNF = toCNF(nc)
+      val combinedFilter = (ExpressionSet(splitConjunctivePredicates(fcCNF)) --
+        ExpressionSet(splitConjunctivePredicates(ncCNF))).reduceOption(And) match {
         case Some(ac) =>
           Filter(And(nc, ac), grandChild)
         case None =>
           nf
       }
+      // [[Filter]] can't pushdown through another [[Filter]]. Once they are combined,
+      // [[BooleanSimplification]] rule will possibly simplify the predicate to the form that
+      // will not be able to pushdown. So we pushdown the combined [[Filter]] immediately.
+      PushDownPredicate(combinedFilter)
   }
 }
 

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

@@ -36,7 +36,6 @@ class FilterPushdownSuite extends PlanTest {
       Batch("Filter Pushdown", FixedPoint(10),
         PushDownPredicate,
         CombineFilters,
-        PushDownPredicate,
         BooleanSimplification,
         PushPredicateThroughJoin,
         CollapseProject) :: Nil
@@ -193,6 +192,26 @@ class FilterPushdownSuite extends PlanTest {
     comparePlans(optimized, correctAnswer)
   }
 
+  test("disjunctive predicates which are able to pushdown should be pushed down after converted") {
+    // (('a === 2) || ('c > 10 || 'a === 3)) can't be pushdown due to the disjunctive form.
+    // However, its conjunctive normal form can be pushdown.
+    val originalQuery = testRelation
+      .select('a, 'b, ('c + 1) as 'cc)
+      .groupBy('a)('a, count('cc) as 'c)
+      .where('c > 10)
+      .where(('a === 2) || ('c > 10 && 'a === 3))
+
+    val optimized = Optimize.execute(originalQuery.analyze)
+    val correctAnswer =
+      testRelation
+        .where('a === 2 || 'a === 3)
+        .select('a, 'b, ('c + 1) as 'cc)
+        .groupBy('a)('a, count('cc) as 'c)
+        .where('c > 10).analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
   test("joins: push to either side") {
     val x = testRelation.subquery('x)
     val y = testRelation.subquery('y)