Fix basicPhysicalOperators usage.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala

@@ -121,12 +121,10 @@ trait PredicateHelper {
   }
 
   /**
-   * Given a sequence of attributes and a filter condition, update attributes' nullability
-   * when the condition implies that attributes cannot be null.
+   * Given an IsNotNull expression, returns the IDs of expressions whose not-nullness
+   * is implied by the IsNotNull expressions.
    */
-  protected def updateAttributeNullabilityFromNonNullConstraints(
-      attributes: Seq[Attribute],
-      condition: Expression): Seq[Attribute] = {
+  protected def getImpliedNotNullExprIds(isNotNullExpr: IsNotNull): Set[ExprId] = {
     // This logic is a little tricky, so we'll use an example to build some intuition.
     // Consider the expression IsNotNull(f(g(x), y)). By definition, its child is not null:
     //    f(g(x), y) is not null
@@ -142,38 +140,21 @@ trait PredicateHelper {
     // By recursively applying this logic, if g is NullIntolerant then x is not null.
     // However, if g is NOT NullIntolerant (e.g. if g(null) is non-null) then we cannot
     // conclude anything about x's nullability.
-    def getNonNullAttributes(isNotNull: IsNotNull): Set[ExprId] = {
-      def getExprIdIfNamed(expr: Expression): Set[ExprId] = expr match {
-        case ne: NamedExpression => Set(ne.toAttribute.exprId)
-        case _ => Set.empty
-      }
-      // Recurse through the IsNotNull expression's descendants, stopping
-      // once we encounter a null-tolerant expression.
-      def getNotNullDescendants(expr: Expression): Set[ExprId] = {
-        expr.children.map {
-          case child: NullIntolerant =>
-            getExprIdIfNamed(child) ++ getNotNullDescendants(child)
-          case child =>
-            getExprIdIfNamed(child)
-        }.reduce(_ ++ _)
-      }
-      getExprIdIfNamed(isNotNull) ++ getNotNullDescendants(isNotNull)
-    }
-
-    val notNullAttributes: Set[ExprId] = {
-      splitConjunctivePredicates(condition)
-        .collect { case isNotNull: IsNotNull => isNotNull }
-        .map(getNonNullAttributes)
-        .reduce(_ ++ _)
+    def getExprIdIfNamed(expr: Expression): Set[ExprId] = expr match {
+      case ne: NamedExpression => Set(ne.toAttribute.exprId)
+      case _ => Set.empty
     }
-
-    attributes.map { a =>
-      if (a.nullable && notNullAttributes.contains(a.exprId)) {
-        a.withNullability(false)
-      } else {
-        a
-      }
+    // Recurse through the IsNotNull expression's descendants, stopping
+    // once we encounter a null-tolerant expression.
+    def getNotNullDescendants(expr: Expression): Set[ExprId] = {
+      expr.children.map {
+        case child: NullIntolerant =>
+          getExprIdIfNamed(child) ++ getNotNullDescendants(child)
+        case child =>
+          getExprIdIfNamed(child)
+      }.reduce(_ ++ _)
     }
+    getExprIdIfNamed(isNotNullExpr) ++ getNotNullDescendants(isNotNullExpr)
   }
 }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -130,8 +130,21 @@ case class Generate(
 case class Filter(condition: Expression, child: LogicalPlan)
   extends OrderPreservingUnaryNode with PredicateHelper {
 
+  private val impliedNotNullExprIds: Set[ExprId] = {
+    splitConjunctivePredicates(condition)
+      .collect { case isNotNull: IsNotNull => isNotNull }
+      .map(getImpliedNotNullExprIds)
+      .reduce(_ ++ _)
+  }
+
   override def output: Seq[Attribute] = {
-    updateAttributeNullabilityFromNonNullConstraints(child.output, condition)
+    child.output.map { a =>
+      if (a.nullable && impliedNotNullExprIds.contains(a.exprId)) {
+        a.withNullability(false)
+      } else {
+        a
+      }
+    }
   }
 
   override def maxRows: Option[Long] = child.maxRows

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

@@ -91,8 +91,23 @@ case class FilterExec(condition: Expression, child: SparkPlan)
   // all the variables at the beginning to take advantage of short circuiting.
   override def usedInputs: AttributeSet = AttributeSet.empty
 
+  // Split out all the IsNotNulls from condition.
+  private val (notNullPreds, otherPreds) = splitConjunctivePredicates(condition).partition {
+    case IsNotNull(_) => true
+    case _ => false
+  }
+
+  private val impliedNotNullExprIds: Set[ExprId] =
+    notNullPreds.map { case n: IsNotNull => getImpliedNotNullExprIds(n) }.reduce(_ ++ _)
+
   override def output: Seq[Attribute] = {
-    updateAttributeNullabilityFromNonNullConstraints(child.output, condition)
+    child.output.map { a =>
+      if (a.nullable && impliedNotNullExprIds.contains(a.exprId)) {
+        a.withNullability(false)
+      } else {
+        a
+      }
+    }
   }
 
   override lazy val metrics = Map(
@@ -133,7 +148,7 @@ case class FilterExec(condition: Expression, child: SparkPlan)
 
     // To generate the predicates we will follow this algorithm.
     // For each predicate that is not IsNotNull, we will generate them one by one loading attributes
-    // as necessary. For each of both attributes, if there is an IsNotNull predicate we will
+    // as necessary. For each attribute, if there is an IsNotNull predicate we will
     // generate that check *before* the predicate. After all of these predicates, we will generate
     // the remaining IsNotNull checks that were not part of other predicates.
     // This has the property of not doing redundant IsNotNull checks and taking better advantage of
@@ -172,7 +187,7 @@ case class FilterExec(condition: Expression, child: SparkPlan)
     // Reset the isNull to false for the not-null columns, then the followed operators could
     // generate better code (remove dead branches).
     val resultVars = input.zipWithIndex.map { case (ev, i) =>
-      if (notNullAttributes.contains(child.output(i).exprId)) {
+      if (impliedNotNullExprIds.contains(child.output(i).exprId)) {
         ev.isNull = FalseLiteral
       }
       ev