no nulll filtering is added for compound expressions.

Author: gatorsmile

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

@@ -606,11 +606,12 @@ object NullPropagation extends Rule[LogicalPlan] {
 object NullFiltering extends Rule[LogicalPlan] with PredicateHelper {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case filter @ Filter(condition, child) =>
-      // We generate a list of additional isNotNull filters from the operator's existing constraints
-      // but remove those that are either already part of the filter condition or are part of the
-      // operator's child constraints.
-      val newIsNotNullConstraints = filter.constraints.filter(_.isInstanceOf[IsNotNull]) --
-        (child.constraints ++ splitConjunctivePredicates(condition))
+      // We generate a list of additional isNotNull filters from the operator's existing
+      // non-compound constraints but remove those that are either already part of the filter
+      // condition or are part of the operator's child constraints.
+      val newIsNotNullConstraints =
+        filter.constraints.filter(isNullFilteringForNonCompoundExpr) --
+          (child.constraints ++ splitConjunctivePredicates(condition))
       if (newIsNotNullConstraints.nonEmpty) {
         Filter(And(newIsNotNullConstraints.reduce(And), condition), child)
       } else {
@@ -619,11 +620,11 @@ object NullFiltering extends Rule[LogicalPlan] with PredicateHelper {
 
     case join @ Join(left, right, joinType, condition) =>
       val leftIsNotNullConstraints = join.constraints
-        .filter(_.isInstanceOf[IsNotNull])
+        .filter(isNullFilteringForNonCompoundExpr)
         .filter(_.references.subsetOf(left.outputSet)) -- left.constraints
       val rightIsNotNullConstraints =
         join.constraints
-          .filter(_.isInstanceOf[IsNotNull])
+          .filter(isNullFilteringForNonCompoundExpr)
           .filter(_.references.subsetOf(right.outputSet)) -- right.constraints
       val newLeftChild = if (leftIsNotNullConstraints.nonEmpty) {
         Filter(leftIsNotNullConstraints.reduce(And), left)
@@ -641,6 +642,12 @@ object NullFiltering extends Rule[LogicalPlan] with PredicateHelper {
         join
       }
   }
+  private def isNullFilteringForNonCompoundExpr(exp: Expression): Boolean = {
+    exp match {
+      case c: IsNotNull if c.child.isInstanceOf[AttributeReference] => true
+      case _ => false
+    }
+  }
 }
 
 /**

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

@@ -40,6 +40,12 @@ class NullFilteringSuite extends PlanTest {
     comparePlans(optimized, correctAnswer)
   }
 
+  test("filter: do not push Null-filtering of compound expressions") {
+    val originalQuery = testRelation.where('a + 'b === 1).analyze
+    val optimized = Optimize.execute(originalQuery)
+    comparePlans(optimized, originalQuery)
+  }
+
   test("single inner join: filter out nulls on either side on equi-join keys") {
     val x = testRelation.subquery('x)
     val y = testRelation.subquery('y)
@@ -83,6 +89,15 @@ class NullFilteringSuite extends PlanTest {
     comparePlans(optimized, correctAnswer)
   }
 
+  test("single inner join: no null filters are generated for compound expression") {
+    val x = testRelation.subquery('x)
+    val y = testRelation.subquery('y)
+    val originalQuery = x.join(y,
+      condition = Some("x.a".attr * 2 === "y.a".attr - 4)).analyze
+    val optimized = Optimize.execute(originalQuery)
+    comparePlans(optimized, originalQuery)
+  }
+
   test("single outer join: no null filters are generated") {
     val x = testRelation.subquery('x)
     val y = testRelation.subquery('y)