Add CombineFilters

Author: wangyum

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

@@ -177,6 +177,7 @@ abstract class Optimizer(sessionCatalog: SessionCatalog)
       InferFiltersFromConstraints,
       PushDownPredicate,
       CollapseProject,
+      CombineFilters,
       RemoveRedundantProject) :+
     Batch("UpdateAttributeReferences", Once,
       UpdateNullabilityInAttributeReferences)

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

@@ -37,17 +37,18 @@ class RewriteSubquerySuite extends PlanTest {
         InferFiltersFromConstraints,
         PushDownPredicate,
         CollapseProject,
+        CombineFilters,
         RemoveRedundantProject) :: Nil
   }
 
+  val relation = LocalRelation('a.int, 'b.int)
+  val relInSubquery = LocalRelation('x.int, 'y.int, 'z.int)
+
   test("Column pruning after rewriting predicate subquery") {
     withSQLConf(SQLConf.CONSTRAINT_PROPAGATION_ENABLED.key -> "false") {
-      val relation = LocalRelation('a.int, 'b.int)
-      val relInSubquery = LocalRelation('x.int, 'y.int, 'z.int)
-
       val query = relation.where('a.in(ListQuery(relInSubquery.select('x)))).select('a)
-
       val optimized = Optimize.execute(query.analyze)
+
       val correctAnswer = relation
         .select('a)
         .join(relInSubquery.select('x), LeftSemi, Some('a === 'x))
@@ -59,12 +60,9 @@ class RewriteSubquerySuite extends PlanTest {
 
   test("Infer filters and push down predicate after rewriting predicate subquery") {
     withSQLConf(SQLConf.CONSTRAINT_PROPAGATION_ENABLED.key -> "true") {
-      val relation = LocalRelation('a.int, 'b.int)
-      val relInSubquery = LocalRelation('x.int, 'y.int, 'z.int)
-
       val query = relation.where('a.in(ListQuery(relInSubquery.select('x)))).select('a)
-
       val optimized = Optimize.execute(query.analyze)
+
       val correctAnswer = relation
         .where(IsNotNull('a)).select('a)
         .join(relInSubquery.where(IsNotNull('x)).select('x), LeftSemi, Some('a === 'x))
@@ -74,4 +72,17 @@ class RewriteSubquerySuite extends PlanTest {
     }
   }
 
+  test("combine filters after rewriting predicate subquery") {
+    val query = relation.where('a.in(ListQuery(relInSubquery.select('x).where('y > 1)))).select('a)
+    val optimized = Optimize.execute(query.analyze)
+
+    val correctAnswer = relation
+      .where(IsNotNull('a)).select('a)
+      .join(relInSubquery.where(IsNotNull('x) && IsNotNull('y) && 'y > 1).select('x),
+        LeftSemi, Some('a === 'x))
+      .analyze
+
+    comparePlans(optimized, correctAnswer)
+  }
+
 }