Avoid resolving metadata fields twice in FetchFieldsPhase (#107474)

We have moved fetching of metadata fields to FetchFieldsPhase with #106325 .
For backwards compatibility, metadata fields requested via `stored_fields` are returned
even when they match a wildcard expression, which differs from how patterns are resolved
via `fields`. Instead of calling getMatchingFieldNames and getFieldType twice per field,
this distinction is now included in FieldFetcher#create via a predicate

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

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.BiPredicate;
 
 /**
  * A fetch sub-phase for high-level field retrieval. Given a list of fields, it
@@ -58,31 +59,34 @@ public FetchSubPhaseProcessor getProcessor(FetchContext fetchContext) {
         final FieldFetcher fieldFetcher = fetchFieldsContext == null ? null
             : fetchFieldsContext.fields() == null ? null
             : fetchFieldsContext.fields().isEmpty() ? null
-            : FieldFetcher.create(searchExecutionContext, fetchFieldsContext.fields());
+            // we want to skip metadata fields if we have a wildcard pattern
+            : FieldFetcher.create(
+                searchExecutionContext,
+                fetchFieldsContext.fields(),
+                (field, ft) -> true,
+                (field, ft) -> searchExecutionContext.isMetadataField(field) == false
+            );
 
         final FieldFetcher metadataFieldFetcher;
         if (storedFieldsContext != null
             && storedFieldsContext.fieldNames() != null
             && storedFieldsContext.fieldNames().isEmpty() == false) {
             final Set<FieldAndFormat> metadataFields = new HashSet<>(DEFAULT_METADATA_FIELDS);
             for (final String storedField : storedFieldsContext.fieldNames()) {
-                final Set<String> matchingFieldNames = searchExecutionContext.getMatchingFieldNames(storedField);
-                for (final String matchingFieldName : matchingFieldNames) {
-                    if (SourceFieldMapper.NAME.equals(matchingFieldName) || IdFieldMapper.NAME.equals(matchingFieldName)) {
-                        continue;
-                    }
-                    final MappedFieldType fieldType = searchExecutionContext.getFieldType(matchingFieldName);
-                    // NOTE: checking if the field is stored is required for backward compatibility reasons and to make
-                    // sure we also handle here stored fields requested via `stored_fields`, which was previously a
-                    // responsibility of StoredFieldsPhase.
-                    if (searchExecutionContext.isMetadataField(matchingFieldName) && fieldType.isStored()) {
-                        metadataFields.add(new FieldAndFormat(matchingFieldName, null));
-                    }
-                }
+                metadataFields.add(new FieldAndFormat(storedField, null));
             }
-            metadataFieldFetcher = FieldFetcher.create(searchExecutionContext, metadataFields);
+            BiPredicate<String, MappedFieldType> includePredicate = (field, ft) -> field.equals(SourceFieldMapper.NAME) == false
+                && field.equals(IdFieldMapper.NAME) == false
+                && searchExecutionContext.isMetadataField(field)
+                && ft.isStored();
+            metadataFieldFetcher = FieldFetcher.create(searchExecutionContext, metadataFields, includePredicate, includePredicate);
         } else {
-            metadataFieldFetcher = FieldFetcher.create(searchExecutionContext, DEFAULT_METADATA_FIELDS);
+            metadataFieldFetcher = FieldFetcher.create(
+                searchExecutionContext,
+                DEFAULT_METADATA_FIELDS,
+                (ft, s) -> true,
+                (field, ft) -> true
+            );
         }
         return new FetchSubPhaseProcessor() {
             @Override

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

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.BiPredicate;
 
 /**
  * A helper class to {@link FetchFieldsPhase} that's initialized with a list of field patterns to fetch.
@@ -40,7 +41,12 @@ private record ResolvedField(String field, String matchingPattern, MappedFieldTy
     /**
      * Build a FieldFetcher for a given search context and collection of fields and formats
      */
-    public static FieldFetcher create(SearchExecutionContext context, Collection<FieldAndFormat> fieldAndFormats) {
+    public static FieldFetcher create(
+        SearchExecutionContext context,
+        Collection<FieldAndFormat> fieldAndFormats,
+        BiPredicate<String, MappedFieldType> explicitIncludePredicate,
+        BiPredicate<String, MappedFieldType> patternIncludePredicate
+    ) {
 
         List<String> unmappedFetchPattern = new ArrayList<>();
         List<ResolvedField> resolvedFields = new ArrayList<>();
@@ -54,11 +60,11 @@ public static FieldFetcher create(SearchExecutionContext context, Collection<Fie
 
             for (String field : context.getMatchingFieldNames(fieldPattern)) {
                 MappedFieldType ft = context.getFieldType(field);
-                // we want to skip metadata fields if we have a wildcard pattern
-                if (context.isMetadataField(field) && matchingPattern != null) {
-                    continue;
+                // provide the fields separately, for the case of aliases where field name and ft.name are different
+                if ((matchingPattern == null && explicitIncludePredicate.test(field, ft))
+                    || (matchingPattern != null && patternIncludePredicate.test(field, ft))) {
+                    resolvedFields.add(new ResolvedField(field, matchingPattern, ft, fieldAndFormat.format));
                 }
-                resolvedFields.add(new ResolvedField(field, matchingPattern, ft, fieldAndFormat.format));
             }
         }
 

server/src/test/java/org/elasticsearch/index/mapper/PlaceHolderFieldMapperTests.java

@@ -71,7 +71,9 @@ public void testFetchValue() throws Exception {
             SearchExecutionContext searchExecutionContext = createSearchExecutionContext(mapperService);
             FieldFetcher fieldFetcher = FieldFetcher.create(
                 searchExecutionContext,
-                Collections.singletonList(new FieldAndFormat("field", null))
+                Collections.singletonList(new FieldAndFormat("field", null)),
+                (s1, mappedFieldType) -> true,
+                (s1, mappedFieldType) -> true
             );
             IndexSearcher searcher = newSearcher(iw);
             LeafReaderContext context = searcher.getIndexReader().leaves().get(0);