Merge remote-tracking branch 'upstream/master' into add-more-not-null…

…-attrs

Conflicts:
	sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala

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

@@ -79,7 +79,7 @@ case class Filter(condition: Expression, child: SparkPlan)
 
   // Split out all the IsNotNulls from condition.
   private val (notNullPreds, otherPreds) = splitConjunctivePredicates(condition).partition {
-    case IsNotNull(a) if a.references.subsetOf(child.outputSet) => true
+    case IsNotNull(a: NullIntolerant) if a.references.subsetOf(child.outputSet) => true
     case _ => false
   }
 
@@ -114,15 +114,12 @@ case class Filter(condition: Expression, child: SparkPlan)
   override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
     val numOutput = metricTerm(ctx, "numOutputRows")
 
-    // filter out the nulls
-    val filterOutNull = notNullAttributes.map { a =>
-      val idx = child.output.map(_.exprId).indexOf(a)
-      if (idx != -1) {
-        s"if (${input(idx).isNull}) continue;"
-      } else {
-        ""
-      }
-    }.mkString("\n")
+    /**
+     * Generates code for `c`, using `in` for input attributes and `attrs` for nullability.
+     */
+    def genPredicate(c: Expression, in: Seq[ExprCode], attrs: Seq[Attribute]): String = {
+      val bound = BindReferences.bindReference(c, attrs)
+      val evaluated = evaluateRequiredVariables(child.output, in, c.references)
 
       // Generate the code for the predicate.
       val ev = ExpressionCanonicalizer.execute(bound).gen(ctx)