[SPARK] Polishing plus tests

closes elastic#569
relates elastic#572

spark/sql-13/src/itest/scala/org/elasticsearch/spark/integration/AbstractScalaEsSparkSQL.scala

@@ -465,6 +465,20 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals("feb", filter.select("tag").take(1)(0)(0))
   }
 
+  @Test
+  def testDataSourcePushDown015NullSafeEqualTo() {
+    val df = esDataSource("pd_nullsafeequalto")
+    val filter = df.filter(df("airport").eqNullSafe("OTP"))
+    if (strictPushDown) {
+      assertEquals(0, filter.count())
+      // however if we change the arguments to be lower cased, it will be Spark who's going to filter out the data
+      return
+    }
+
+    assertEquals(1, filter.count())
+    assertEquals("feb", filter.select("tag").take(1)(0)(0))
+  }
+
   @Test
   def testDataSourcePushDown02GT() {
     val df = esDataSource("pd_gt")

spark/sql-13/src/main/scala/org/elasticsearch/spark/sql/DefaultSource.scala

@@ -148,6 +148,11 @@ private[sql] case class ElasticsearchRelation(parameters: Map[String, String], @
     filter match {
 
       case EqualTo(attribute, value)            => {
+        // if we get a null, translate it into a missing query (we're extra careful - Spark should translate the equals into isMissing anyway)
+        if (value == null || value == None || value == Unit) {
+          return s"""{"missing":{"field":"$attribute"}}"""
+        }
+
         if (strictPushDown) s"""{"term":{"$attribute":${extract(value)}}}"""
         else s"""{"query":{"match":{"$attribute":${extract(value)}}}}"""
       }
@@ -205,12 +210,6 @@ private[sql] case class ElasticsearchRelation(parameters: Map[String, String], @
 
       //
       // [[EqualNullSafe]] Filter notes:
-      //
-      // To work this function clearly alone, null-safety check should be added and
-      // return a query string using "missing" filter (identically with [[IsNull]]).
-      // However, Spark does not pass [[EqualNullSafe]] filter having null value but
-      // instead [[IsNull]]. To make sure, we might have to add null-safety check logic
-      // here as well as String filter. For now, it works identical with [[EqualTo]].
 
       case f:Product if isClass(f, "org.apache.spark.sql.sources.EqualNullSafe")   => {
         var arg = extract(f.productElement(1))