address comments.

Author: gatorsmile

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

@@ -95,8 +95,7 @@ case class FilterExec(condition: Expression, child: SparkPlan)
 
   // If one expression and its children are null intolerant, it is null intolerant.
   private def isNullIntolerant(expr: Expression): Boolean = expr match {
-    case e: NullIntolerant =>
-      if (e.isInstanceOf[LeafExpression]) true else e.children.forall(isNullIntolerant)
+    case e: NullIntolerant => e.children.forall(isNullIntolerant)
     case _ => false
   }
 

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala

@@ -1667,6 +1667,10 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
       expr = "_1 + _2", expectedNonNullableColumns = Seq("_1", "_2"))
     verifyNullabilityInFilterExec(df,
       expr = "_1", expectedNonNullableColumns = Seq("_1"))
+    // `constructIsNotNullConstraints` infers the IsNotNull(_2) from IsNotNull(_2 + Rand())
+    // Thus, we are able to set nullability of _2 to false.
+    // If IsNotNull(_2) is not given from `constructIsNotNullConstraints`, the impl of
+    // isNullIntolerant in `FilterExec` needs an update for more advanced inference.
     verifyNullabilityInFilterExec(df,
       expr = "_2 + Rand()", expectedNonNullableColumns = Seq("_2"))
     verifyNullabilityInFilterExec(df,