Fix an NPE when requesting inner hits and _source is disabled. (#44836)

This PR makes two changes to FetchSourceSubPhase when _source is disabled and
we're in a nested context:
* If no source filters are provided, return early to avoid an NPE.
* If there are source filters, make sure to throw an exception.

The behavior was chosen to match what currently happens in a non-nested context.

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

@@ -42,17 +42,23 @@ public void hitExecute(SearchContext context, HitContext hitContext) {
         SourceLookup source = context.lookup().source();
         FetchSourceContext fetchSourceContext = context.fetchSourceContext();
         assert fetchSourceContext.fetchSource();
-        if (nestedHit == false) {
-            if (fetchSourceContext.includes().length == 0 && fetchSourceContext.excludes().length == 0) {
-                hitContext.hit().sourceRef(source.internalSourceRef());
-                return;
-            }
-            if (source.internalSourceRef() == null) {
+
+        // If source is disabled in the mapping, then attempt to return early.
+        if (source.source() == null && source.internalSourceRef() == null) {
+            if (containsFilters(fetchSourceContext)) {
                 throw new IllegalArgumentException("unable to fetch fields from _source field: _source is disabled in the mappings " +
-                        "for index [" + context.indexShard().shardId().getIndexName() + "]");
+                    "for index [" + context.indexShard().shardId().getIndexName() + "]");
             }
+            return;
+        }
+
+        // If this is a parent document and there are no source filters, then add the source as-is.
+        if (nestedHit == false && containsFilters(fetchSourceContext) == false) {
+            hitContext.hit().sourceRef(source.internalSourceRef());
+            return;
         }
 
+        // Otherwise, filter the source and add it to the hit.
         Object value = source.filter(fetchSourceContext);
         if (nestedHit) {
             value = getNestedSource((Map<String, Object>) value, hitContext);
@@ -79,6 +85,10 @@ public void hitExecute(SearchContext context, HitContext hitContext) {
         }
     }
 
+    private static boolean containsFilters(FetchSourceContext context) {
+        return context.includes().length != 0 || context.excludes().length != 0;
+    }
+
     private Map<String, Object> getNestedSource(Map<String, Object> sourceAsMap, HitContext hitContext) {
         for (SearchHit.NestedIdentity o = hitContext.hit().getNestedIdentity(); o != null; o = o.getChild()) {
             sourceAsMap = (Map<String, Object>) sourceAsMap.get(o.getField().string());

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

@@ -119,6 +119,17 @@ public void testSourceDisabled() throws IOException {
                 "for index [index]", exception.getMessage());
     }
 
+    public void testNestedSourceWithSourceDisabled() {
+        FetchSubPhase.HitContext hitContext = hitExecute(null, true, null, null,
+            new SearchHit.NestedIdentity("nested1", 0, null));
+        assertNull(hitContext.hit().getSourceAsMap());
+
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
+            () -> hitExecute(null, true, "field1", null, new SearchHit.NestedIdentity("nested1", 0, null)));
+        assertEquals("unable to fetch fields from _source field: _source is disabled in the mappings " +
+            "for index [index]", e.getMessage());
+    }
+
     private FetchSubPhase.HitContext hitExecute(XContentBuilder source, boolean fetchSource, String include, String exclude) {
         return hitExecute(source, fetchSource, include, exclude, null);
     }