Avoids converting predicate sets to lists

Author: liancheng

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

@@ -335,7 +335,7 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
             } else {
               // (a || b || c || ...) && (a || b || d || ...) =>
               // ((c || ...) && (d || ...)) || a || b
-              (And(ldiff.reduce(Or), rdiff.reduce(Or)) :: common.toList).reduce(Or)
+              (common + And(ldiff.reduce(Or), rdiff.reduce(Or))).reduce(Or)
             }
           }
       }  // end of And(left, right)
@@ -373,7 +373,7 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
             } else {
               // (a && b && c && ...) || (a && b && d && ...) =>
               // ((c && ...) || (d && ...)) && a && b
-              (Or(ldiff.reduce(And), rdiff.reduce(And)) :: common.toList).reduce(And)
+              (common + Or(ldiff.reduce(And), rdiff.reduce(And))).reduce(And)
             }
           }
       }  // end of Or(left, right)

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

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.optimizer
 
 import org.apache.spark.sql.catalyst.analysis.EliminateAnalysisOperators
-import org.apache.spark.sql.catalyst.expressions.{Literal, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Or, And, Literal, Expression}
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.PlanTest
 import org.apache.spark.sql.catalyst.rules._
@@ -40,11 +40,22 @@ class BooleanSimplificationSuite extends PlanTest {
 
   val testRelation = LocalRelation('a.int, 'b.int, 'c.int, 'd.string)
 
-  def checkCondition(originCondition: Expression, optimizedCondition: Expression): Unit = {
-    val originQuery = testRelation.where(originCondition).analyze
-    val optimized = Optimize(originQuery)
-    val expected = testRelation.where(optimizedCondition).analyze
-    comparePlans(optimized, expected)
+  def compareConditions(e1: Expression, e2: Expression): Boolean = (e1, e2) match {
+    case (And(l1, l2), And(r1, r2)) =>
+      compareConditions(l1, r1) && compareConditions(l2, r2) ||
+        compareConditions(l1, r2) && compareConditions(l2, r1)
+
+    case (Or(l1, l2), Or(r1, r2)) =>
+      compareConditions(l1, r1) && compareConditions(l2, r2) ||
+        compareConditions(l1, r2) && compareConditions(l2, r1)
+
+    case (l, r) => l == r
+  }
+
+  def checkCondition(input: Expression, expected: Expression): Unit = {
+    val plan = testRelation.where(input).analyze
+    val actual = Optimize(plan).expressions.head
+    compareConditions(actual, expected)
   }
 
   test("a && a => a") {