resolve saveToEs saves case classes fields with NULL values #998 (#1478)

This PR fix the Scala Writer when it processes case classes with null fields.
Closes #998

Author: lucebert

mr/src/main/java/org/elasticsearch/hadoop/serialization/builder/FilteringValueWriter.java

@@ -27,11 +27,13 @@
 import org.elasticsearch.hadoop.serialization.field.FieldFilter;
 import org.elasticsearch.hadoop.serialization.field.FieldFilter.NumberedInclude;
 import org.elasticsearch.hadoop.util.StringUtils;
+import static org.elasticsearch.hadoop.cfg.ConfigurationOptions.ES_SPARK_DATAFRAME_WRITE_NULL_VALUES_DEFAULT;
 
 public abstract class FilteringValueWriter<T> implements ValueWriter<T>, SettingsAware {
 
     private List<NumberedInclude> includes;
     private List<String> excludes;
+    private Boolean writeNullValues = Boolean.parseBoolean(ES_SPARK_DATAFRAME_WRITE_NULL_VALUES_DEFAULT);
 
     @Override
     public void setSettings(Settings settings) {
@@ -41,10 +43,15 @@ public void setSettings(Settings settings) {
             includes.add(new NumberedInclude(include));
         }
         excludes = StringUtils.tokenize(settings.getMappingExcludes());
+        writeNullValues = settings.getDataFrameWriteNullValues();
     }
 
     protected boolean shouldKeep(String parentField, String name) {
         name = StringUtils.hasText(parentField) ? parentField + "." + name : name;
         return FieldFilter.filter(name, includes, excludes).matched;
     }
+
+    protected boolean hasWriteNullValues() {
+        return writeNullValues;
+    }
 }

spark/core/src/main/scala/org/elasticsearch/spark/serialization/ScalaValueWriter.scala

@@ -61,10 +61,12 @@ class ScalaValueWriter(writeUnknownTypes: Boolean = false) extends JdkValueWrite
         generator.writeBeginObject()
         for ((k, v) <- m) {
           if (shouldKeep(parentField, k.toString)) {
-            generator.writeFieldName(k.toString)
-            val result = doWrite(v, generator, k.toString)
-            if (!result.isSuccesful) {
-              return result
+            if (v != None && v!= null && v != () || hasWriteNullValues) {
+              generator.writeFieldName(k.toString)
+              val result = doWrite(v, generator, k.toString)
+              if (!result.isSuccesful) {
+                return result
+              }
             }
           }
         }

spark/core/src/test/scala/org/elasticsearch/spark/serialization/ScalaValueWriterTest.scala

@@ -154,6 +154,58 @@ class ScalaValueWriterTest {
     println(serialize(node1))
   }
 
+  @Test
+  def testCaseClassWithNone(): Unit = {
+
+    case class TestCaseClass(option : Option[String])
+    val caseClass = TestCaseClass(None)
+
+    assertEquals("""{}""", serialize(caseClass))
+  }
+
+  @Test
+  def testCaseClassWithSome(): Unit = {
+
+    case class TestCaseClass(option : Option[String])
+    val caseClass = TestCaseClass(Some("value"))
+
+    assertEquals("""{"option":"value"}""", serialize(caseClass))
+  }
+
+
+  @Test
+  def testCaseClassWithSomeAndNone(): Unit = {
+
+    case class TestCaseClass(option1 : Option[String], option2 : Option[String])
+    val caseClass = TestCaseClass(None, Some("value2"))
+
+    assertEquals("""{"option2":"value2"}""", serialize(caseClass))
+  }
+
+
+  @Test
+  def testCaseClassWithInnerObject(): Unit = {
+
+    case class TestCaseClass(option1 : Option[String], option2 : Option[TestCaseClassInner])
+    case class TestCaseClassInner(option1 : Option[String], option2 : Option[String])
+    val caseClass = TestCaseClass(None, Some(TestCaseClassInner(option1 = Some("value1") , option2 = None)))
+
+    assertEquals("""{"option2":{"option1":"value1"}}""", serialize(caseClass))
+  }
+
+  @Test
+  def testCaseClassWithInnerObjectAndNullSetting(): Unit = {
+
+    case class TestCaseClass(option1: Option[String], option2: Option[TestCaseClassInner])
+    case class TestCaseClassInner(option1: Option[String], option2: Option[String])
+    val caseClass = TestCaseClass(None, Some(TestCaseClassInner(option1 = Some("value1"), option2 = None)))
+
+    val settings = new TestSettings()
+    settings.setProperty(ConfigurationOptions.ES_SPARK_DATAFRAME_WRITE_NULL_VALUES, "true")
+
+    assertEquals("""{"option1":null,"option2":{"option1":"value1","option2":null}}""", serialize(caseClass, settings))
+  }
+
   @Test
   def testDate(): Unit = {
     val date = new Date(1420114230123l)