title: Spark Sql UDF throwing NullPointer when adding a filter on a columns that uses that UDF
body: <p>SPARK_VERSION = 2.2.0</p>
  
  <p>I ran into an interesting issue when trying to do a <code>filter</code> on a dataframe that has columns that were added using a UDF. I am able to replicate the problem with a smaller set of data.</p>
  
  <p>Given the dummy case classes:</p>
  
  <pre><code>case class Info(number: Int, color: String)
  case class Record(name: String, infos: Seq[Info])
  </code></pre>
  
  <p>and the following data:</p>
  
  <pre><code>val blue = Info(1, "blue")
  val black = Info(2, "black")
  val yellow = Info(3, "yellow")
  val orange = Info(4, "orange")
  val white = Info(5, "white")
  
  val a  = Record("a", Seq(blue, black, white))
  val a2 = Record("a", Seq(yellow, white, orange))
  val b = Record("b", Seq(blue, black))
  val c = Record("c", Seq(white, orange))
  val d = Record("d", Seq(orange, black))
  </code></pre>
  
  <p>do the following...</p>
  
  <p>Create two dataframes (we will call them left and right)</p>
  
  <pre><code>val left = Seq(a, b).toDF
  val right = Seq(a2, c, d).toDF
  </code></pre>
  
  <p>Join those dataframes using a <code>full_outer</code> join, and take only what is on the right side</p>
  
  <pre><code>val rightOnlyInfos  = left.alias("l")
    .join(right.alias("r"), Seq("name"), "full_outer")
    .filter("l.infos is null")
    .select($"name", $"r.infos".as("r_infos"))
  </code></pre>
  
  <p>This results in the following:</p>
  
  <pre><code>rightOnlyInfos.show(false)
  +----+-----------------------+
  |name|r_infos                |
  +----+-----------------------+
  |c   |[[5,white], [4,orange]]|
  |d   |[[4,orange], [2,black]]|
  +----+-----------------------+
  </code></pre>
  
  <p>Using the following udf, add a new column that is a boolean and represents whether or not one of the <code>r_infos</code> contains the color <code>black</code></p>
  
  <pre><code>def hasBlack = (s: Seq[Row]) =&gt; {
    s.exists{ case Row(num: Int, color: String) =&gt;
      color == "black"
    }
  }
  
  val joinedBreakdown = rightOnlyInfos.withColumn("has_black", udf(hasBlack).apply($"r_infos"))
  </code></pre>
  
  <p>This is where I am seeing problems now. If I do the following, I get no errors:</p>
  
  <pre><code>joinedBreakdown.show(false)
  </code></pre>
  
  <p>and it results (like expected) in:</p>
  
  <pre><code>+----+-----------------------+---------+
  |name|r_infos                |has_black|
  +----+-----------------------+---------+
  |c   |[[5,white], [4,orange]]|false    |
  |d   |[[4,orange], [2,black]]|true     |
  +----+-----------------------+---------+
  </code></pre>
  
  <p>and the schema</p>
  
  <pre><code>joinedBreakdown.printSchema
  </code></pre>
  
  <p>shows</p>
  
  <pre><code>root 
    |-- name: string (nullable = true) 
    |-- r_infos: array (nullable = true) 
    | |-- element: struct (containsNull = true)
    | | |-- number: integer (nullable = false) 
    | | |-- color: string (nullable = true)
    |-- has_black: boolean (nullable = true)
  </code></pre>
  
  <p>However, when I try to filter by that results, I get an error:</p>
  
  <pre><code>joinedBreakdown.filter("has_black == true").show(false)
  </code></pre>
  
  <p>With the following error:</p>
  
  <pre><code>org.apache.spark.SparkException: Failed to execute user defined function($anonfun$hasBlack$1: (array&lt;struct&lt;number:int,color:string&gt;&gt;) =&gt; boolean)
    at org.apache.spark.sql.catalyst.expressions.ScalaUDF.eval(ScalaUDF.scala:1075)
    at org.apache.spark.sql.catalyst.expressions.BinaryExpression.eval(Expression.scala:411)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin.org$apache$spark$sql$catalyst$optimizer$EliminateOuterJoin$$canFilterOutNull(joins.scala:127)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$$anonfun$rightHasNonNullPredicate$lzycompute$1$1.apply(joins.scala:138)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$$anonfun$rightHasNonNullPredicate$lzycompute$1$1.apply(joins.scala:138)
    at scala.collection.LinearSeqOptimized$class.exists(LinearSeqOptimized.scala:93)
    at scala.collection.immutable.List.exists(List.scala:84)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin.rightHasNonNullPredicate$lzycompute$1(joins.scala:138)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin.rightHasNonNullPredicate$1(joins.scala:138)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin.org$apache$spark$sql$catalyst$optimizer$EliminateOuterJoin$$buildNewJoinType(joins.scala:145)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$$anonfun$apply$2.applyOrElse(joins.scala:152)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$$anonfun$apply$2.applyOrElse(joins.scala:150)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
    at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin.apply(joins.scala:150)
    at org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin.apply(joins.scala:116)
    at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:85)
    at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:82)
    at scala.collection.LinearSeqOptimized$class.foldLeft(LinearSeqOptimized.scala:124)
    at scala.collection.immutable.List.foldLeft(List.scala:84)
    at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:82)
    at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:74)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:74)
    at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:78)
    at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:78)
    at org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:84)
    at org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:80)
    at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:89)
    at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:89)
    at org.apache.spark.sql.Dataset.withAction(Dataset.scala:2832)
    at org.apache.spark.sql.Dataset.head(Dataset.scala:2153)
    at org.apache.spark.sql.Dataset.take(Dataset.scala:2366)
    at org.apache.spark.sql.Dataset.showString(Dataset.scala:245)
    at org.apache.spark.sql.Dataset.show(Dataset.scala:646)
    at org.apache.spark.sql.Dataset.show(Dataset.scala:623)
    ... 58 elided
  Caused by: java.lang.NullPointerException
    at $anonfun$hasBlack$1.apply(&lt;console&gt;:41)
    at $anonfun$hasBlack$1.apply(&lt;console&gt;:40)
    at org.apache.spark.sql.catalyst.expressions.ScalaUDF$$anonfun$2.apply(ScalaUDF.scala:92)
    at org.apache.spark.sql.catalyst.expressions.ScalaUDF$$anonfun$2.apply(ScalaUDF.scala:91)
    at org.apache.spark.sql.catalyst.expressions.ScalaUDF.eval(ScalaUDF.scala:1072)
    ... 114 more
  </code></pre>
  
  <p><strong>EDIT:</strong> opened up a jira issue. Pasting here for tracking purposes:
  <a href="https://issues.apache.org/jira/browse/SPARK-22942" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/SPARK-22942</a></p>