[SPARK-17897] [SQL] [BACKPORT-2.0] Fixed IsNotNull Constraint Inference Rule

[gatorsmile]

What changes were proposed in this pull request?

This PR is to backport #16067 to Spark 2.0

---

The constraints of an operator is the expressions that evaluate to true for all the rows produced. That means, the expression result should be neither false nor unknown (NULL). Thus, we can conclude that IsNotNull on all the constraints, which are generated by its own predicates or propagated from the children. The constraint can be a complex expression. For better usage of these constraints, we try to push down IsNotNull to the lowest-level expressions (i.e., Attribute). IsNotNull can be pushed through an expression when it is null intolerant. (When the input is NULL, the null-intolerant expression always evaluates to NULL.)

Below is the existing code we have for IsNotNull pushdown.

  private def scanNullIntolerantExpr(expr: Expression): Seq[Attribute] = expr match {
    case a: Attribute => Seq(a)
    case _: NullIntolerant | IsNotNull(_: NullIntolerant) =>
      expr.children.flatMap(scanNullIntolerantExpr)
    case _ => Seq.empty[Attribute]
  }

IsNotNull itself is not null-intolerant. It converts null to false. If the expression does not include any Not-like expression, it works; otherwise, it could generate a wrong result. This PR is to fix the above function by removing the IsNotNull from the inference. After the fix, when a constraint has a IsNotNull expression, we infer new attribute-specific IsNotNull constraints if and only if IsNotNull appears in the root.

Without the fix, the following test case will return empty.

val data = Seq[java.lang.Integer](1, null).toDF("key")
data.filter("not key is not null").show()

Before the fix, the optimized plan is like

== Optimized Logical Plan ==
Project [value#1 AS key#3]
+- Filter (isnotnull(value#1) && NOT isnotnull(value#1))
   +- LocalRelation [value#1]

After the fix, the optimized plan is like

== Optimized Logical Plan ==
Project [value#1 AS key#3]
+- Filter NOT isnotnull(value#1)
   +- LocalRelation [value#1]

How was this patch tested?

Added a test

[SparkQA]

Test build #72734 has started for PR 16894 at commit 11d2684.

[hvanhovell]

retest this please

[SparkQA]

Test build #72735 has finished for PR 16894 at commit 11d2684.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

retest this please

[SparkQA]

Test build #72743 has finished for PR 16894 at commit 11d2684.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

thanks, merging to 2.0

[yhuai]

thanks!