Make array value parsing flag more robust. (#63354)

When constructing a value fetcher, the 'parsesArrayValue' flag must match
`FieldMapper#parsesArrayValue`. However there is nothing in code or tests to
help enforce this.

This PR reworks the value fetcher constructors so that `parsesArrayValue` is
'false' by default. Just as for `FieldMapper#parsesArrayValue`, field types must
explicitly set it to true and ensure the behavior is covered by tests.

Follow-up to #62974.

Author: jtibshirani

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/RankFeatureFieldMapper.java

@@ -118,7 +118,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected Float parseSourceValue(Object value) {
                     return objectToFloat(value);

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/RankFeaturesFieldMapper.java

@@ -93,7 +93,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected Object parseSourceValue(Object value) {
                     return value;

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/ScaledFloatFieldMapper.java

@@ -211,7 +211,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected Double parseSourceValue(Object value) {
                     double doubleValue;

modules/parent-join/src/main/java/org/elasticsearch/join/mapper/ParentJoinFieldMapper.java

@@ -227,7 +227,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected Object parseSourceValue(Object value) {
                     return value;

modules/percolator/src/main/java/org/elasticsearch/percolator/PercolatorFieldMapper.java

@@ -222,7 +222,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected Object parseSourceValue(Object value) {
                     return value;

plugins/analysis-icu/src/main/java/org/elasticsearch/index/mapper/ICUCollationKeywordFieldMapper.java

@@ -104,7 +104,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
 
-            return new SourceValueFetcher(name(), mapperService, false, nullValue) {
+            return new SourceValueFetcher(name(), mapperService, nullValue) {
                 @Override
                 protected String parseSourceValue(Object value) {
                     String keywordValue = value.toString();

plugins/mapper-murmur3/src/main/java/org/elasticsearch/index/mapper/murmur3/Murmur3FieldMapper.java

@@ -110,7 +110,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected String parseSourceValue(Object value) {
                     return value.toString();

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

@@ -256,13 +256,21 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             String geoFormat = format != null ? format : GeoJsonGeometryFormat.NAME;
 
             Function<Object, Object> valueParser = value -> geometryParser.parseAndFormatObject(value, geoFormat);
-
-            return new SourceValueFetcher(name(), mapperService, parsesArrayValue) {
-                @Override
-                protected Object parseSourceValue(Object value) {
-                    return valueParser.apply(value);
-                }
-            };
+            if (parsesArrayValue) {
+                return new ArraySourceValueFetcher(name(), mapperService) {
+                    @Override
+                    protected Object parseSourceValue(Object value) {
+                        return valueParser.apply(value);
+                    }
+                };
+            } else {
+                return new SourceValueFetcher(name(), mapperService) {
+                    @Override
+                    protected Object parseSourceValue(Object value) {
+                        return valueParser.apply(value);
+                    }
+                };
+            }
         }
     }
 

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

@@ -0,0 +1,74 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.mapper;
+
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.search.lookup.SourceLookup;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * An implementation of {@link ValueFetcher} that knows how to extract values
+ * from the document source.
+ *
+ * This class differs from {@link SourceValueFetcher} in that it directly handles
+ * array values in parsing. Field types should use this class if their corresponding
+ * mapper returns true for {@link FieldMapper#parsesArrayValue()}.
+ */
+public abstract class ArraySourceValueFetcher implements ValueFetcher {
+    private final Set<String> sourcePaths;
+    private final @Nullable Object nullValue;
+
+    public ArraySourceValueFetcher(String fieldName, MapperService mapperService) {
+        this(fieldName, mapperService, null);
+    }
+
+    /**
+     * @param fieldName The name of the field.
+     * @param mapperService A mapper service.
+     * @param nullValue A optional substitute value if the _source value is 'null'.
+     */
+    public ArraySourceValueFetcher(String fieldName, MapperService mapperService, Object nullValue) {
+        this.sourcePaths = mapperService.sourcePath(fieldName);
+        this.nullValue = nullValue;
+    }
+
+    @Override
+    public List<Object> fetchValues(SourceLookup lookup) {
+        List<Object> values = new ArrayList<>();
+        for (String path : sourcePaths) {
+            Object sourceValue = lookup.extractValue(path, nullValue);
+            if (sourceValue == null) {
+                return List.of();
+            }
+            values.addAll((List<?>) parseSourceValue(sourceValue));
+        }
+        return values;
+    }
+
+    /**
+     * Given a value that has been extracted from a document's source, parse it into a standard
+     * format. This parsing logic should closely mirror the value parsing in
+     * {@link FieldMapper#parseCreateField} or {@link FieldMapper#parse}.
+     */
+    protected abstract Object parseSourceValue(Object value);
+}

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

@@ -102,7 +102,7 @@ public ValueFetcher valueFetcher(MapperService mapperService, SearchLookup searc
             if (format != null) {
                 throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
             }
-            return new SourceValueFetcher(name(), mapperService, false) {
+            return new SourceValueFetcher(name(), mapperService) {
                 @Override
                 protected Object parseSourceValue(Object value) {
                     return value;