removed isnotnull of compound expressions from constraints.

added a support for operator Not for all the binary operators.

Author: gatorsmile

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

@@ -608,12 +608,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
-      // 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))
+      // 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))
+
       if (newIsNotNullConstraints.nonEmpty) {
         Filter(And(newIsNotNullConstraints.reduce(And), condition), child)
       } else {
@@ -622,11 +622,11 @@ object NullFiltering extends Rule[LogicalPlan] with PredicateHelper {
 
     case join @ Join(left, right, joinType, condition) =>
       val leftIsNotNullConstraints = join.constraints
-        .filter(isNullFilteringForNonCompoundExpr)
+        .filter(_.isInstanceOf[IsNotNull])
         .filter(_.references.subsetOf(left.outputSet)) -- left.constraints
       val rightIsNotNullConstraints =
         join.constraints
-          .filter(isNullFilteringForNonCompoundExpr)
+          .filter(_.isInstanceOf[IsNotNull])
           .filter(_.references.subsetOf(right.outputSet)) -- right.constraints
       val newLeftChild = if (leftIsNotNullConstraints.nonEmpty) {
         Filter(leftIsNotNullConstraints.reduce(And), left)
@@ -644,12 +644,6 @@ 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/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -46,22 +46,16 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]] extends TreeNode[PlanT
   private def constructIsNotNullConstraints(constraints: Set[Expression]): Set[Expression] = {
     // Currently we only propagate constraints if the condition consists of equality
     // and ranges. For all other cases, we return an empty set of constraints
-    constraints.map {
-      case EqualTo(l, r) =>
-        Set(IsNotNull(l), IsNotNull(r))
-      case GreaterThan(l, r) =>
-        Set(IsNotNull(l), IsNotNull(r))
-      case GreaterThanOrEqual(l, r) =>
-        Set(IsNotNull(l), IsNotNull(r))
-      case LessThan(l, r) =>
-        Set(IsNotNull(l), IsNotNull(r))
-      case LessThanOrEqual(l, r) =>
-        Set(IsNotNull(l), IsNotNull(r))
-      case Not(EqualTo(l, r)) =>
-        Set(IsNotNull(l), IsNotNull(r))
-      case _ =>
-        Set.empty[Expression]
-    }.foldLeft(Set.empty[Expression])(_ union _.toSet)
+    var isNotNullConstraints = Set.empty[Expression]
+    constraints.collect {
+      case b @ BinaryComparison(l, r) if !b.isInstanceOf[EqualNullSafe] =>
+        if (l.isInstanceOf[AttributeReference]) isNotNullConstraints += IsNotNull(l)
+        if (r.isInstanceOf[AttributeReference]) isNotNullConstraints += IsNotNull(r)
+      case Not(b @ BinaryComparison(l, r)) if !b.isInstanceOf[EqualNullSafe] =>
+        if (l.isInstanceOf[AttributeReference]) isNotNullConstraints += IsNotNull(l)
+        if (r.isInstanceOf[AttributeReference]) isNotNullConstraints += IsNotNull(r)
+    }
+    isNotNullConstraints
   }
 
   /**

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

@@ -217,4 +217,19 @@ class ConstraintPropagationSuite extends SparkFunSuite {
         IsNotNull(resolveColumn(tr, "a")),
         IsNotNull(resolveColumn(tr, "b")))))
   }
+
+  test("IsNotNull constraints of compound expressions in filters") {
+    val tr = LocalRelation('a.int, 'b.string, 'c.int)
+    verifyConstraints(tr
+      .where('a.attr + 'c.attr > 10).analyze.constraints,
+      ExpressionSet(Seq(resolveColumn(tr, "a") + resolveColumn(tr, "c") > 10)))
+  }
+
+  test("IsNotNull constraints of BinaryComparison in Not in filters") {
+    val tr = LocalRelation('a.int, 'b.string, 'c.int)
+    verifyConstraints(tr
+      .where(!('a.attr < 10)).analyze.constraints,
+      ExpressionSet(Seq(IsNotNull(resolveColumn(tr, "a")),
+        Not(resolveColumn(tr, "a") < 10))))
+  }
 }