Handle nested arrays in field retrieval. (#60981)

We accept _source values with multiple levels of arrays, such as
`"field": [[[1, 2]]]`. This PR ensures that field retrieval can handle nested
arrays by unwrapping the arrays before parsing the values.

Author: jtibshirani

server/src/main/java/org/elasticsearch/index/mapper/FieldMapper.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.search.lookup.SourceLookup;
 
 import java.io.IOException;
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -46,6 +47,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Queue;
 import java.util.TreeMap;
 import java.util.stream.StreamSupport;
 
@@ -307,11 +309,17 @@ public List<?> lookupValues(SourceLookup lookup, @Nullable String format) {
         List<Object> values = new ArrayList<>();
         if (parsesArrayValue()) {
             return (List<?>) parseSourceValue(sourceValue, format);
-        } else {
-            List<?> sourceValues = sourceValue instanceof List
-                ? (List<?>) sourceValue
-                : org.elasticsearch.common.collect.List.of(sourceValue);
-            for (Object value : sourceValues) {
+        }
+
+        // We allow source values to contain multiple levels of arrays, such as `"field": [[1, 2]]`.
+        // So we need to unwrap these arrays before passing them on to be parsed.
+        Queue<Object> queue = new ArrayDeque<>();
+        queue.add(sourceValue);
+        while (queue.isEmpty() == false) {
+            Object value = queue.poll();
+            if (value instanceof List) {
+                queue.addAll((List<?>) value);
+            } else {
                 Object parsedValue = parseSourceValue(value, format);
                 if (parsedValue != null) {
                     values.add(parsedValue);

server/src/test/java/org/elasticsearch/search/fetch/subphase/FieldValueRetrieverTests.java

@@ -115,6 +115,35 @@ public void testRetrieveAllFields() throws IOException {
         assertThat(fields.size(), equalTo(2));
     }
 
+    public void testNestedArrays() throws IOException {
+        MapperService mapperService = createMapperService();
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .startArray("field")
+                .startArray().value("first").value("second").endArray()
+            .endArray()
+        .endObject();
+
+        Map<String, DocumentField> fields = retrieveFields(mapperService, source, "field");
+        DocumentField field = fields.get("field");
+        assertNotNull(field);
+        assertThat(field.getValues().size(), equalTo(2));
+        assertThat(field.getValues(), hasItems("first", "second"));
+
+        source = XContentFactory.jsonBuilder().startObject()
+            .startArray("object")
+                .startObject().array("field", "first", "second").endObject()
+                .startObject().array("field", "third").endObject()
+                .startObject().field("field", "fourth").endObject()
+            .endArray()
+            .endObject();
+
+        fields = retrieveFields(mapperService, source, "object.field");
+        field = fields.get("object.field");
+        assertNotNull(field);
+        assertThat(field.getValues().size(), equalTo(4));
+        assertThat(field.getValues(), hasItems("first", "second", "third", "fourth"));
+    }
+
     public void testArrayValueMappers() throws IOException {
         MapperService mapperService = createMapperService();