Fixing support for empty and null arrays (#1827)

This commit makes it so that if an array is null in Elasticsearch, it comes into es-hadoop/spark as null. Previously
it showed up as either [] or [null], depending on spark version.
Closes #1527

Author: masseyke

mr/src/main/java/org/elasticsearch/hadoop/serialization/ScrollReader.java

@@ -895,7 +895,12 @@ else if (t == Token.START_OBJECT) {
             String rawValue = parser.text();
             try {
                 if (isArrayField(fieldMapping)) {
-                    return singletonList(fieldMapping, parseValue(parser, esType), parser);
+                    Object parsedValue = parseValue(parser, esType);
+                    if (parsedValue == null) {
+                        return null; //There is not a null element in the array. The array itself is null.
+                    } else {
+                        return singletonList(fieldMapping, parsedValue, parser);
+                    }
                 } else {
                     return parseValue(parser, esType);
                 }

mr/src/test/java/org/elasticsearch/hadoop/serialization/ScrollReaderTest.java

@@ -53,6 +53,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -431,6 +432,24 @@ public void testScrollWithNestedArrays() throws IOException {
         assertEquals(10L, JsonUtils.query("a").get(0).get(0).get(0).apply(scroll.getHits().get(2)[1]));
     }
 
+    @Test
+    public void testScrollWithEmptyrrays() throws IOException {
+        MappingSet mappings = getMappingSet("empty-list");
+        InputStream stream = getClass().getResourceAsStream(scrollData("empty-list"));
+        Settings testSettings = new TestSettings();
+        testSettings.setProperty(ConfigurationOptions.ES_READ_FIELD_AS_ARRAY_INCLUDE, "status_code");
+        testSettings.setProperty(ConfigurationOptions.ES_READ_METADATA, "" + readMetadata);
+        testSettings.setProperty(ConfigurationOptions.ES_READ_METADATA_FIELD, "" + metadataField);
+        testSettings.setProperty(ConfigurationOptions.ES_OUTPUT_JSON, "" + readAsJson);
+        JdkValueReader valueReader = ObjectUtils.instantiate(JdkValueReader.class.getName(), testSettings);
+        ScrollReader reader = new ScrollReader(ScrollReaderConfigBuilder.builder(valueReader, mappings.getResolvedView(), testSettings));
+        ScrollReader.Scroll scroll = reader.read(stream);
+        // The first entry is null. The second is an array with the single element '123'. And the third is an empty array
+        assertNull(JsonUtils.query("status_code").apply(scroll.getHits().get(0)[1]));
+        assertEquals(123L, JsonUtils.query("status_code").get(0).apply(scroll.getHits().get(1)[1]));
+        assertEquals(Collections.emptyList(), JsonUtils.query("status_code").apply(scroll.getHits().get(2)[1]));
+    }
+
     @Test(expected = EsHadoopParsingException.class)
     public void testScrollWithBreakOnInvalidMapping() throws IOException {
         MappingSet mappings = getMappingSet("numbers-as-strings");

mr/src/test/resources/org/elasticsearch/hadoop/serialization/scrollReaderTestData/empty-list/mapping.json

@@ -0,0 +1,29 @@
+{
+  "artists" : {
+    "mappings" : {
+      "properties" : {
+        "name" : {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword",
+              "ignore_above" : 256
+            }
+          }
+        },
+        "status_code" : {
+          "type" : "long",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword",
+              "ignore_above" : 256
+            }
+          }
+        },
+        "ts" : {
+          "type" : "long"
+        }
+      }
+    }
+  }
+}

mr/src/test/resources/org/elasticsearch/hadoop/serialization/scrollReaderTestData/empty-list/scroll.json

@@ -0,0 +1,55 @@
+{
+  "_scroll_id" : "FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFmJUOHBDZ0pjU1dPUm9CbThxUWp1M3cAAAAAAABtCxZzRFItZkxtWFRjZWxXZ3pZamhZZWZ3",
+  "took" : 0,
+  "timed_out" : false,
+  "_shards" : {
+    "total" : 1,
+    "successful" : 1,
+    "skipped" : 0,
+    "failed" : 0
+  },
+  "hits" : {
+    "total" : {
+      "value" : 3,
+      "relation" : "eq"
+    },
+    "max_score" : 1.0,
+    "hits" : [
+      {
+        "_index" : "empty_list",
+        "_type" : "_doc",
+        "_id" : "1",
+        "_score" : 1.0,
+        "_source" : {
+          "status_code" : null,
+          "ts" : 12345678910,
+          "name" : "john"
+        }
+      },
+      {
+        "_index" : "empty_list",
+        "_type" : "_doc",
+        "_id" : "2",
+        "_score" : 1.0,
+        "_source" : {
+          "status_code" : [
+            123
+          ],
+          "ts" : null,
+          "name" : "johnny"
+        }
+      },
+      {
+        "_index" : "empty_list",
+        "_type" : "_doc",
+        "_id" : "3",
+        "_score" : 1.0,
+        "_source" : {
+          "status_code" : [ ],
+          "ts" : 12345678910,
+          "name" : null
+        }
+      }
+    ]
+  }
+}

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

@@ -2242,6 +2242,26 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals(2, df.count())
   }
 
+  @Test
+  def testArraysAndNulls() {
+    val index = wrapIndex("sparksql-test-arrays-and-nulls")
+    val typed = "data"
+    val (target, docPath) = makeTargets(index, typed)
+    RestUtils.touch(index)
+    val document1 = """{ "id": 1, "status_code" : [123]}""".stripMargin
+    val document2 = """{ "id" : 2, "status_code" : []}""".stripMargin
+    val document3 = """{ "id" : 3, "status_code" : null}""".stripMargin
+    sc.makeRDD(Seq(document1, document2, document3)).saveJsonToEs(target)
+    RestUtils.refresh(index)
+    val df = sqc.read.format("es").option("es.read.field.as.array.include","status_code").load(index)
+      .select("id", "status_code")
+    var result = df.where("id = 1").first().getList(1)
+    assertEquals(123, result.get(0))
+    result = df.where("id = 2").first().getList(1)
+    assertTrue(result.isEmpty)
+    assertTrue(df.where("id = 3").first().isNullAt(1))
+  }
+
   @Test
   def testReadFieldInclude(): Unit = {
     val data = Seq(

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

@@ -2324,6 +2324,26 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals(2, df.count())
   }
 
+  @Test
+  def testArraysAndNulls() {
+    val index = wrapIndex("sparksql-test-arrays-and-nulls")
+    val typed = "data"
+    val (target, docPath) = makeTargets(index, typed)
+    RestUtils.touch(index)
+    val document1 = """{ "id": 1, "status_code" : [123]}""".stripMargin
+    val document2 = """{ "id" : 2, "status_code" : []}""".stripMargin
+    val document3 = """{ "id" : 3, "status_code" : null}""".stripMargin
+    sc.makeRDD(Seq(document1, document2, document3)).saveJsonToEs(target)
+    RestUtils.refresh(index)
+    val df = sqc.read.format("es").option("es.read.field.as.array.include","status_code").load(index)
+      .select("id", "status_code")
+    var result = df.where("id = 1").first().getList(1)
+    assertEquals(123, result.get(0))
+    result = df.where("id = 2").first().getList(1)
+    assertTrue(result.isEmpty)
+    assertTrue(df.where("id = 3").first().isNullAt(1))
+  }
+
   @Test
   def testReadFieldInclude(): Unit = {
     val data = Seq(

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

@@ -2325,6 +2325,26 @@ class AbstractScalaEsScalaSparkSQL(prefix: String, readMetadata: jl.Boolean, pus
     assertEquals(2, df.count())
   }
 
+  @Test
+  def testArraysAndNulls() {
+    val index = wrapIndex("sparksql-test-arrays-and-nulls")
+    val typed = "data"
+    val (target, docPath) = makeTargets(index, typed)
+    RestUtils.touch(index)
+    val document1 = """{ "id": 1, "status_code" : [123]}""".stripMargin
+    val document2 = """{ "id" : 2, "status_code" : []}""".stripMargin
+    val document3 = """{ "id" : 3, "status_code" : null}""".stripMargin
+    sc.makeRDD(Seq(document1, document2, document3)).saveJsonToEs(target)
+    RestUtils.refresh(index)
+    val df = sqc.read.format("es").option("es.read.field.as.array.include","status_code").load(index)
+      .select("id", "status_code")
+    var result = df.where("id = 1").first().getList(1)
+    assertEquals(123, result.get(0))
+    result = df.where("id = 2").first().getList(1)
+    assertTrue(result.isEmpty)
+    assertTrue(df.where("id = 3").first().isNullAt(1))
+  }
+
   @Test
   def testReadFieldInclude(): Unit = {
     val data = Seq(