Do not deserialise the document when not needed in the fields fetch phase (#84184)

docs/changelog/84184.yaml

@@ -0,0 +1,5 @@
+pr: 84184
+summary: Do not deserialise the document when not needed in the fields fetch phase
+area: Search
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/search/fetch/subphase/FieldFetcher.java

@@ -174,7 +174,7 @@ public class FieldFetcher {
                 documentFields.put(field, new DocumentField(field, parsedValues, ignoredValues));
             }
         }
-        collectUnmapped(documentFields, sourceLookup.source(), "", 0);
+        collectUnmapped(documentFields, sourceLookup, "", 0);
         return documentFields;
     }
 

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

@@ -152,6 +152,13 @@ public class SourceLookup implements Map<String, Object> {
         return sourceAsBytes;
     }
 
+    /**
+     * Checks if the source has been deserialized as a {@link Map} of java objects.
+     */
+    public boolean hasSourceAsMap() {
+        return source != null;
+    }
+
     /**
      * Returns the values associated with the path. Those are "low" level values, and it can
      * handle path expression where an array/list is navigated within.

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

@@ -275,6 +275,30 @@ public class FieldFetcherTests extends MapperServiceTestCase {
         assertThat(fields.size(), equalTo(2));
     }
 
+    public void testEmptyFetch() throws IOException {
+        MapperService mapperService = createMapperService();
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject().field("field", "value").endObject();
+        SourceLookup sourceLookup = new SourceLookup();
+        sourceLookup.setSource(BytesReference.bytes(source));
+        {
+            // make sure that an empty fetch don't deserialize the document
+            FieldFetcher fieldFetcher = FieldFetcher.create(newSearchExecutionContext(mapperService), List.of());
+            Map<String, DocumentField> fields = fieldFetcher.fetch(sourceLookup);
+            assertThat(fields.size(), equalTo(0));
+            assertThat(sourceLookup.hasSourceAsMap(), equalTo(false));
+        }
+        {
+            // but a non-empty fetch deserialize the document
+            FieldFetcher fieldFetcher = FieldFetcher.create(
+                newSearchExecutionContext(mapperService),
+                fieldAndFormatList("field", null, false)
+            );
+            Map<String, DocumentField> fields = fieldFetcher.fetch(sourceLookup);
+            assertThat(fields.size(), equalTo(1));
+            assertThat(sourceLookup.hasSourceAsMap(), equalTo(true));
+        }
+    }
+
     public void testNestedArrays() throws IOException {
         MapperService mapperService = createMapperService();
         XContentBuilder source = XContentFactory.jsonBuilder()