[SPARK-9814][SQL] EqualNotNull not passing to data sources

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala

@@ -349,6 +349,11 @@ private[sql] object DataSourceStrategy extends Strategy with Logging {
       case expressions.EqualTo(Literal(v, _), a: Attribute) =>
         Some(sources.EqualTo(a.name, v))
 
+      case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) =>
+        Some(sources.EqualNullSafe(a.name, v))
+      case expressions.EqualNullSafe(Literal(v, _), a: Attribute) =>
+        Some(sources.EqualNullSafe(a.name, v))
+
       case expressions.GreaterThan(a: Attribute, Literal(v, _)) =>
         Some(sources.GreaterThan(a.name, v))
       case expressions.GreaterThan(Literal(v, _), a: Attribute) =>

sql/core/src/main/scala/org/apache/spark/sql/sources/filters.scala

@@ -36,6 +36,15 @@ abstract class Filter
  */
 case class EqualTo(attribute: String, value: Any) extends Filter
 
+/**
+ * Performs equality comparison, similar to [[EqualTo]]. However, this differs from [[EqualTo]]
+ * in that it returns `true` (rather than NULL) if both inputs are NULL, and `false`
+ * (rather than NULL) if one of the input is NULL and the other is not NULL.
+ *
+ * @since 1.5.0
+ */
+case class EqualNullSafe(attribute: String, value: Any) extends Filter
+
 /**
  * A filter that evaluates to `true` iff the attribute evaluates to a value
  * greater than `value`.

sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala

@@ -56,6 +56,7 @@ case class SimpleFilteredScan(from: Int, to: Int)(@transient val sqlContext: SQL
     // Predicate test on integer column
     def translateFilterOnA(filter: Filter): Int => Boolean = filter match {
       case EqualTo("a", v) => (a: Int) => a == v
+      case EqualNullSafe("a", v) => (a: Int) => a == v
       case LessThan("a", v: Int) => (a: Int) => a < v
       case LessThanOrEqual("a", v: Int) => (a: Int) => a <= v
       case GreaterThan("a", v: Int) => (a: Int) => a > v