Extend support of allowedFields to getMatchingFieldNames and getAllFi…

…elds

The SearchExecutionContext supports the notion of allowed fields, provided via a specific setter method.
Fields are though only filtered for the getFieldType method. There needs to be consistency between getMatchingFieldNames and getFieldType.
In fact there are places in the code where getMatchingFieldNames is called to resolve field name patterns, and later getFieldType is called
on each of the resolved fields. If the former resolves to one field that we can't retrieve a field type for, that is unexpected and to be considered a bug.

In addition, this commit adds consistency for getAllFields: this is only called by field caps, hence a different codepath that does not seem to set allowed fields
for now, but it's important for the context to provide consistency around fields access, especially for methods that are as broad as getAllFields,
despite their currently very specific usage.

This surfaced as we are trying to move fetching of the `_ignored` field to use value fetchers, which use a search execution context and resolve the field type,
whereas until now they are retrieved directly via StoredFieldsPhase and completely bypass such check.

This commit also adds a test that was missing around verifying that SearchExecutionContext applies the allowedFields predicate when provided.

server/src/main/java/org/elasticsearch/index/query/QueryRewriteContext.java

@@ -37,6 +37,7 @@
 import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 /**
  * Context object used to rewrite {@link QueryBuilder} instances into simplified version.
@@ -318,35 +319,45 @@ public boolean indexMatches(String pattern) {
      * @param pattern the field name pattern
      */
     public Set<String> getMatchingFieldNames(String pattern) {
+        Set<String> matches;
         if (runtimeMappings.isEmpty()) {
-            return mappingLookup.getMatchingFieldNames(pattern);
-        }
-        Set<String> matches = new HashSet<>(mappingLookup.getMatchingFieldNames(pattern));
-        if ("*".equals(pattern)) {
-            matches.addAll(runtimeMappings.keySet());
-        } else if (Regex.isSimpleMatchPattern(pattern) == false) {
-            // no wildcard
-            if (runtimeMappings.containsKey(pattern)) {
-                matches.add(pattern);
-            }
+            matches = mappingLookup.getMatchingFieldNames(pattern);
         } else {
-            for (String name : runtimeMappings.keySet()) {
-                if (Regex.simpleMatch(pattern, name)) {
-                    matches.add(name);
+            matches = new HashSet<>(mappingLookup.getMatchingFieldNames(pattern));
+            if ("*".equals(pattern)) {
+                matches.addAll(runtimeMappings.keySet());
+            } else if (Regex.isSimpleMatchPattern(pattern) == false) {
+                // no wildcard
+                if (runtimeMappings.containsKey(pattern)) {
+                    matches.add(pattern);
+                }
+            } else {
+                for (String name : runtimeMappings.keySet()) {
+                    if (Regex.simpleMatch(pattern, name)) {
+                        matches.add(name);
+                    }
                 }
             }
         }
-        return matches;
+        // If the field is not allowed, behave as if it is not mapped
+        return allowedFields == null ? matches : matches.stream().filter(allowedFields).collect(Collectors.toSet());
     }
 
     /**
      * @return An {@link Iterable} with key the field name and value the MappedFieldType
      */
     public Iterable<Map.Entry<String, MappedFieldType>> getAllFields() {
-        var allFromMapping = mappingLookup.getFullNameToFieldType();
-        // runtime mappings and non-runtime fields don't overlap, so we can simply concatenate the iterables here
-        return runtimeMappings.isEmpty()
+        Map<String, MappedFieldType> allFromMapping = mappingLookup.getFullNameToFieldType();
+        Set<Map.Entry<String, MappedFieldType>> allEntrySet = allowedFields == null
             ? allFromMapping.entrySet()
-            : () -> Iterators.concat(allFromMapping.entrySet().iterator(), runtimeMappings.entrySet().iterator());
+            : allFromMapping.entrySet().stream().filter(entry -> allowedFields.test(entry.getKey())).collect(Collectors.toSet());
+        if (runtimeMappings.isEmpty()) {
+            return allEntrySet;
+        }
+        Set<Map.Entry<String, MappedFieldType>> runtimeEntrySet = allowedFields == null
+            ? runtimeMappings.entrySet()
+            : runtimeMappings.entrySet().stream().filter(entry -> allowedFields.test(entry.getKey())).collect(Collectors.toSet());
+        // runtime mappings and non-runtime fields don't overlap, so we can simply concatenate the iterables here
+        return () -> Iterators.concat(allEntrySet.iterator(), runtimeEntrySet.iterator());
     }
 }

server/src/test/java/org/elasticsearch/index/query/SearchExecutionContextTests.java

@@ -96,6 +96,7 @@
 import java.util.stream.Collectors;
 
 import static java.util.Collections.singletonMap;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.notNullValue;
@@ -401,6 +402,84 @@ public void testSyntheticSourceSearchLookup() throws IOException {
         assertEquals("meow", source.source().get("cat"));
     }
 
+    public void testAllowedFields() {
+        Map<String, Object> runtimeMappings = Map.ofEntries(
+            Map.entry("runtimecat", Map.of("type", "keyword")),
+            Map.entry("runtimedog", Map.of("type", "long"))
+        );
+        SearchExecutionContext context = createSearchExecutionContext(
+            "uuid",
+            null,
+            SearchExecutionContextTests.createMappingLookup(
+                List.of(new MockFieldMapper.FakeFieldType("pig"), new MockFieldMapper.FakeFieldType("cat")),
+                List.of(new TestRuntimeField("runtime", "long"))
+            ),
+            runtimeMappings
+        );
+
+        assertNotNull(context.getFieldType("pig"));
+        assertNotNull(context.getFieldType("cat"));
+        assertNotNull(context.getFieldType("runtimecat"));
+        assertNotNull(context.getFieldType("runtimedog"));
+        assertNotNull(context.getFieldType("runtime"));
+        assertEquals(3, context.getMatchingFieldNames("runtime*").size());
+        assertEquals(2, context.getMatchingFieldNames("*cat").size());
+        assertThat(getFieldNames(context.getAllFields()), containsInAnyOrder("pig", "cat", "runtimecat", "runtimedog", "runtime"));
+
+        context.setAllowedFields(s -> true);
+        assertNotNull(context.getFieldType("pig"));
+        assertTrue(context.isFieldMapped("pig"));
+        assertNotNull(context.getFieldType("cat"));
+        assertTrue(context.isFieldMapped("cat"));
+        assertNotNull(context.getFieldType("runtimecat"));
+        assertTrue(context.isFieldMapped("runtimecat"));
+        assertNotNull(context.getFieldType("runtimedog"));
+        assertTrue(context.isFieldMapped("runtimedog"));
+        assertNotNull(context.getFieldType("runtime"));
+        assertTrue(context.isFieldMapped("runtime"));
+        assertEquals(3, context.getMatchingFieldNames("runtime*").size());
+        assertEquals(2, context.getMatchingFieldNames("*cat").size());
+        assertThat(getFieldNames(context.getAllFields()), containsInAnyOrder("pig", "cat", "runtimecat", "runtimedog", "runtime"));
+
+        context.setAllowedFields(s -> s.equals("cat"));
+        assertNull(context.getFieldType("pig"));
+        assertFalse(context.isFieldMapped("pig"));
+        assertNotNull(context.getFieldType("cat"));
+        assertTrue(context.isFieldMapped("cat"));
+        assertNull(context.getFieldType("runtimecat"));
+        assertFalse(context.isFieldMapped("runtimecat"));
+        assertNull(context.getFieldType("runtimedog"));
+        assertFalse(context.isFieldMapped("runtimedog"));
+        assertNull(context.getFieldType("runtime"));
+        assertFalse(context.isFieldMapped("runtime"));
+        assertEquals(0, context.getMatchingFieldNames("runtime*").size());
+        assertEquals(1, context.getMatchingFieldNames("*cat").size());
+        assertThat(getFieldNames(context.getAllFields()), containsInAnyOrder("cat"));
+
+        context.setAllowedFields(s -> s.contains("dog") == false);
+        assertNotNull(context.getFieldType("pig"));
+        assertTrue(context.isFieldMapped("pig"));
+        assertNotNull(context.getFieldType("cat"));
+        assertTrue(context.isFieldMapped("cat"));
+        assertNotNull(context.getFieldType("runtimecat"));
+        assertTrue(context.isFieldMapped("runtimecat"));
+        assertNull(context.getFieldType("runtimedog"));
+        assertFalse(context.isFieldMapped("runtimedog"));
+        assertNotNull(context.getFieldType("runtime"));
+        assertTrue(context.isFieldMapped("runtime"));
+        assertEquals(2, context.getMatchingFieldNames("runtime*").size());
+        assertEquals(2, context.getMatchingFieldNames("*cat").size());
+        assertThat(getFieldNames(context.getAllFields()), containsInAnyOrder("pig", "cat", "runtimecat", "runtime"));
+    }
+
+    private static List<String> getFieldNames(Iterable<Map.Entry<String, MappedFieldType>> fields) {
+        List<String> fieldNames = new ArrayList<>();
+        for (Map.Entry<String, MappedFieldType> field : fields) {
+            fieldNames.add(field.getKey());
+        }
+        return fieldNames;
+    }
+
     public static SearchExecutionContext createSearchExecutionContext(String indexUuid, String clusterAlias) {
         return createSearchExecutionContext(indexUuid, clusterAlias, MappingLookup.EMPTY, Map.of());
     }