Unable to retrieve multiple stored field values (elastic#106575)

The issue happens when we try to use multiple stored fields through
the FetchFieldsPhase, which we do when using `_fields` since
we have a single shared instance of SingleFieldsVisitor per field
and document and use a shared `currentValues` array.

docs/changelog/106575.yaml

@@ -0,0 +1,5 @@
+pr: 106575
+summary: Unable to retrieve multiple stored field values
+area: "Search"
+type: bug
+issues: []

modules/runtime-fields-common/src/yamlRestTest/resources/rest-api-spec/test/runtime_fields/10_keyword.yml

@@ -108,6 +108,76 @@ setup:
   - match: {hits.hits.0.fields.day_of_week_letters: [T, a, d, h, r, s, u, y] }
   - match: {hits.hits.0.fields.prefixed_node: [node_c] }
 
+---
+"fetch multiple stored fields":
+  - skip:
+      version: " - 8.13.99"
+      reason: "bug fixed in 8.14"
+
+  - do:
+      indices.create:
+        index: sensor-test
+        body:
+          settings:
+            number_of_shards: 1
+            number_of_replicas: 0
+          mappings:
+            runtime:
+              prefixed_node:
+                type: keyword
+                script:
+                  source: |
+                    for (String node : params._fields.node.values) {
+                      emit(params.prefix + node);
+                    }
+                  params:
+                    prefix: node_
+              prefixed_region:
+                type: keyword
+                script:
+                  source: |
+                    for (String region : params._fields.region.values) {
+                      emit(params.prefix + region)
+                    }
+                  params:
+                    prefix: us-
+            properties:
+              timestamp:
+                type: date
+              node:
+                type: keyword
+                store: true
+              region:
+                type: keyword
+                store: true
+
+  - do:
+      bulk:
+        index: sensor-test
+        refresh: true
+        body: |
+          {"index":{}}
+          {"timestamp": 1516729294000, "node": "a", "region": "west-1" }
+          {"index":{}}
+          {"timestamp": 1516642894000, "node": "b", "region": "west-2" }
+          {"index":{}}
+          {"timestamp": 1516556494000, "node": "a", "region": "west-1"}
+          {"index":{}}
+          {"timestamp": 1516470094000, "node": "b", "region": "west-2"}
+          {"index":{}}
+          {"timestamp": 1516383694000, "node": "c", "region": "west-2"}
+          {"index":{}}
+          {"timestamp": 1516297294000, "node": "c", "region": "west-2"}
+  - do:
+      search:
+        index: sensor-test
+        body:
+          sort: timestamp
+          fields: [prefixed_node, prefixed_region]
+  - match: {hits.total.value: 6}
+  - match: {hits.hits.0.fields.prefixed_node: [node_c] }
+  - match: {hits.hits.0.fields.prefixed_region: [us-west-2]}
+
 ---
 "docvalue_fields":
   - do:

server/src/main/java/org/elasticsearch/search/lookup/LeafFieldLookupProvider.java

@@ -34,27 +34,20 @@ static Function<LeafReaderContext, LeafFieldLookupProvider> fromStoredFields() {
         return ctx -> new LeafFieldLookupProvider() {
 
             StoredFields storedFields;
-            int currentDoc = -1;
-            final List<Object> currentValues = new ArrayList<>(2);
 
             @Override
             public void populateFieldLookup(FieldLookup fieldLookup, int doc) throws IOException {
                 if (storedFields == null) {
                     storedFields = ctx.reader().storedFields();
                 }
-                if (doc == currentDoc) {
-                    fieldLookup.setValues(currentValues);
-                } else {
-                    currentDoc = doc;
-                    currentValues.clear();
-                    // TODO can we remember which fields have been loaded here and get them eagerly next time?
-                    // likelihood is if a script is loading several fields on one doc they will load the same
-                    // set of fields next time round
-                    SingleFieldsVisitor visitor = new SingleFieldsVisitor(fieldLookup.fieldType(), currentValues);
-                    storedFields.document(doc, visitor);
-                    fieldLookup.setValues(currentValues);
-                }
+                // TODO can we remember which fields have been loaded here and get them eagerly next time?
+                // likelihood is if a script is loading several fields on one doc they will load the same
+                // set of fields next time round
+                final List<Object> currentValues = new ArrayList<>(2);
+                storedFields.document(doc, new SingleFieldsVisitor(fieldLookup.fieldType(), currentValues));
+                fieldLookup.setValues(currentValues);
             }
+
         };
     }
 

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

@@ -0,0 +1,77 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.search.fetch.subphase;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.StoredField;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.StoredFields;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.mapper.MapperServiceTestCase;
+import org.elasticsearch.search.lookup.FieldLookup;
+import org.elasticsearch.search.lookup.LeafFieldLookupProvider;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentFactory;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.util.List;
+
+public class PopulateFieldLookupTests extends MapperServiceTestCase {
+    public void testPopulateFieldLookup() throws IOException {
+        final XContentBuilder mapping = createMapping();
+        final MapperService mapperService = createMapperService(mapping);
+        withLuceneIndex(mapperService, iw -> {
+            final Document doc = new Document();
+            doc.add(new StoredField("integer", 101));
+            doc.add(new StoredField("keyword", new BytesRef("foobar")));
+            iw.addDocument(doc);
+        }, reader -> {
+            final StoredFields storedFields = reader.storedFields();
+            final Document document = storedFields.document(0);
+            final List<String> documentFields = document.getFields().stream().map(IndexableField::name).toList();
+            assertThat(documentFields, Matchers.containsInAnyOrder("integer", "keyword"));
+
+            final IndexSearcher searcher = newSearcher(reader);
+            final LeafReaderContext readerContext = searcher.getIndexReader().leaves().get(0);
+            final LeafFieldLookupProvider provider = LeafFieldLookupProvider.fromStoredFields().apply(readerContext);
+            final FieldLookup integerFieldLookup = new FieldLookup(mapperService.fieldType("integer"));
+            final FieldLookup keywordFieldLookup = new FieldLookup(mapperService.fieldType("keyword"));
+            provider.populateFieldLookup(integerFieldLookup, 0);
+            provider.populateFieldLookup(keywordFieldLookup, 0);
+            assertEquals(List.of(101), integerFieldLookup.getValues());
+            assertEquals(List.of("foobar"), keywordFieldLookup.getValues());
+        });
+    }
+
+    private static XContentBuilder createMapping() throws IOException {
+        final XContentBuilder mapping = XContentFactory.jsonBuilder().startObject().startObject("_doc");
+        {
+            mapping.startObject("properties");
+            {
+                mapping.startObject("integer");
+                {
+                    mapping.field("type", "integer").field("store", "true");
+                }
+                mapping.endObject();
+                mapping.startObject("keyword");
+                {
+                    mapping.field("type", "keyword").field("store", "true");
+                }
+                mapping.endObject();
+            }
+            mapping.endObject();
+
+        }
+        return mapping.endObject().endObject();
+    }
+}