inner hits: Do not allow inner hits that use _source and have a non nested object field as parent

Closes #25315

core/src/main/java/org/elasticsearch/index/cache/bitset/BitsetFilterCache.java

@@ -239,7 +239,7 @@ public final class BitsetFilterCache extends AbstractIndexComponent implements I
                     hasNested = true;
                     for (ObjectMapper objectMapper : docMapper.objectMappers().values()) {
                         if (objectMapper.nested().isNested()) {
-                            ObjectMapper parentObjectMapper = docMapper.findParentObjectMapper(objectMapper);
+                            ObjectMapper parentObjectMapper = objectMapper.getParentObjectMapper(mapperService);
                             if (parentObjectMapper != null && parentObjectMapper.nested().isNested()) {
                                 warmUp.add(parentObjectMapper.nestedTypeFilter());
                             }

core/src/main/java/org/elasticsearch/index/mapper/DocumentMapper.java

@@ -292,21 +292,6 @@ public class DocumentMapper implements ToXContentFragment {
         return nestedObjectMapper;
     }
 
-    /**
-     * Returns the parent {@link ObjectMapper} instance of the specified object mapper or <code>null</code> if there
-     * isn't any.
-     */
-    // TODO: We should add: ObjectMapper#getParentObjectMapper()
-    public ObjectMapper findParentObjectMapper(ObjectMapper objectMapper) {
-        int indexOfLastDot = objectMapper.fullPath().lastIndexOf('.');
-        if (indexOfLastDot != -1) {
-            String parentNestObjectPath = objectMapper.fullPath().substring(0, indexOfLastDot);
-            return objectMappers().get(parentNestObjectPath);
-        } else {
-            return null;
-        }
-    }
-
     public boolean isParent(String type) {
         return mapperService.getParentTypes().contains(type);
     }

core/src/main/java/org/elasticsearch/index/mapper/ObjectMapper.java

@@ -396,6 +396,35 @@ public class ObjectMapper extends Mapper implements Cloneable {
         return dynamic;
     }
 
+    /**
+     * Returns the parent {@link ObjectMapper} instance of the specified object mapper or <code>null</code> if there
+     * isn't any.
+     */
+    public ObjectMapper getParentObjectMapper(MapperService mapperService) {
+        int indexOfLastDot = fullPath().lastIndexOf('.');
+        if (indexOfLastDot != -1) {
+            String parentNestObjectPath = fullPath().substring(0, indexOfLastDot);
+            return mapperService.getObjectMapper(parentNestObjectPath);
+        } else {
+            return null;
+        }
+    }
+
+    /**
+     * Returns whether all parent objects fields are nested too.
+     */
+    public boolean parentObjectMapperAreNested(MapperService mapperService) {
+        for (ObjectMapper parent = getParentObjectMapper(mapperService);
+             parent != null;
+             parent = parent.getParentObjectMapper(mapperService)) {
+
+            if (parent.nested().isNested() == false) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     @Override
     public ObjectMapper merge(Mapper mergeWith, boolean updateAllTypes) {
         if (!(mergeWith instanceof ObjectMapper)) {

core/src/main/java/org/elasticsearch/index/query/NestedQueryBuilder.java

@@ -353,6 +353,11 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
                 name, parentSearchContext, parentObjectMapper, nestedObjectMapper
             );
             setupInnerHitsContext(queryShardContext, nestedInnerHits);
+            if ((nestedInnerHits.hasFetchSourceContext() == false || nestedInnerHits.sourceRequested()) &&
+                nestedObjectMapper.parentObjectMapperAreNested(parentSearchContext.mapperService()) == false) {
+                throw new IllegalArgumentException("Cannot execute inner hits. One or more parent object fields of nested field [" +
+                    nestedObjectMapper.name() + "] are not nested. All parent fields need to be nested fields too");
+            }
             queryShardContext.nestedScope().previousLevel();
             innerHitsContext.addInnerHitDefinition(nestedInnerHits);
         }

core/src/main/java/org/elasticsearch/search/fetch/FetchPhase.java

@@ -40,6 +40,7 @@ import org.elasticsearch.index.fieldvisitor.CustomFieldsVisitor;
 import org.elasticsearch.index.fieldvisitor.FieldsVisitor;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ObjectMapper;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.mapper.Uid;
@@ -246,7 +247,7 @@ public class FetchPhase implements SearchPhase {
         ObjectMapper nestedObjectMapper = documentMapper.findNestedObjectMapper(nestedSubDocId, context, subReaderContext);
         assert nestedObjectMapper != null;
         SearchHit.NestedIdentity nestedIdentity =
-                getInternalNestedIdentity(context, nestedSubDocId, subReaderContext, documentMapper, nestedObjectMapper);
+                getInternalNestedIdentity(context, nestedSubDocId, subReaderContext, context.mapperService(), nestedObjectMapper);
 
         if (source != null) {
             Tuple<XContentType, Map<String, Object>> tuple = XContentHelper.convertToMap(source, true);
@@ -311,9 +312,7 @@ public class FetchPhase implements SearchPhase {
         return searchFields;
     }
 
-    private SearchHit.NestedIdentity getInternalNestedIdentity(SearchContext context, int nestedSubDocId,
+    private SearchHit.NestedIdentity getInternalNestedIdentity(SearchContext context, int nestedSubDocId, LeafReaderContext subReaderContext, MapperService mapperService, ObjectMapper nestedObjectMapper) throws IOException {
-                                                               LeafReaderContext subReaderContext, DocumentMapper documentMapper,
-                                                               ObjectMapper nestedObjectMapper) throws IOException {
         int currentParent = nestedSubDocId;
         ObjectMapper nestedParentObjectMapper;
         ObjectMapper current = nestedObjectMapper;
@@ -321,7 +320,7 @@ public class FetchPhase implements SearchPhase {
         SearchHit.NestedIdentity nestedIdentity = null;
         do {
             Query parentFilter;
-            nestedParentObjectMapper = documentMapper.findParentObjectMapper(current);
+            nestedParentObjectMapper = current.getParentObjectMapper(mapperService);
             if (nestedParentObjectMapper != null) {
                 if (nestedParentObjectMapper.nested().isNested() == false) {
                     current = nestedParentObjectMapper;

core/src/test/java/org/elasticsearch/index/mapper/NestedObjectMapperTests.java

@@ -36,6 +36,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.function.Function;
 
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.nullValue;
@@ -428,4 +429,35 @@ public class NestedObjectMapperTests extends ESSingleNodeTestCase {
         createIndex("test5", Settings.builder().put(MapperService.INDEX_MAPPING_NESTED_FIELDS_LIMIT_SETTING.getKey(), 0).build())
             .mapperService().merge("type", new CompressedXContent(mapping.apply("type")), MergeReason.MAPPING_RECOVERY, false);
     }
+
+    public void testParentObjectMapperAreNested() throws Exception {
+        MapperService mapperService = createIndex("index1", Settings.EMPTY, "doc", jsonBuilder().startObject()
+                .startObject("properties")
+                    .startObject("comments")
+                        .field("type", "nested")
+                        .startObject("properties")
+                            .startObject("messages")
+                                .field("type", "nested").endObject()
+                            .endObject()
+                        .endObject()
+                    .endObject()
+                .endObject()).mapperService();
+        ObjectMapper objectMapper = mapperService.getObjectMapper("comments.messages");
+        assertTrue(objectMapper.parentObjectMapperAreNested(mapperService));
+
+        mapperService = createIndex("index2", Settings.EMPTY, "doc", jsonBuilder().startObject()
+            .startObject("properties")
+                .startObject("comments")
+                    .field("type", "object")
+                        .startObject("properties")
+                            .startObject("messages")
+                                .field("type", "nested").endObject()
+                            .endObject()
+                    .endObject()
+                .endObject()
+            .endObject()).mapperService();
+        objectMapper = mapperService.getObjectMapper("comments.messages");
+        assertFalse(objectMapper.parentObjectMapperAreNested(mapperService));
+    }
+
 }

core/src/test/java/org/elasticsearch/search/fetch/subphase/InnerHitsIT.java

@@ -411,9 +411,26 @@ public class InnerHitsIT extends ESIntegTestCase {
                 .endObject()));
         indexRandom(true, requests);
 
+        SearchPhaseExecutionException e = expectThrows(
+            SearchPhaseExecutionException.class,
+            () -> client().prepareSearch("articles").setQuery(nestedQuery("comments.messages",
+                matchQuery("comments.messages.message", "fox"), ScoreMode.Avg).innerHit(new InnerHitBuilder())).get()
+        );
+        assertEquals("Cannot execute inner hits. One or more parent object fields of nested field [comments.messages] are " +
+            "not nested. All parent fields need to be nested fields too", e.shardFailures()[0].getCause().getMessage());
+
+        e = expectThrows(
+            SearchPhaseExecutionException.class,
+            () -> client().prepareSearch("articles").setQuery(nestedQuery("comments.messages",
+                matchQuery("comments.messages.message", "fox"), ScoreMode.Avg).innerHit(new InnerHitBuilder()
+                .setFetchSourceContext(new FetchSourceContext(true)))).get()
+        );
+        assertEquals("Cannot execute inner hits. One or more parent object fields of nested field [comments.messages] are " +
+            "not nested. All parent fields need to be nested fields too", e.shardFailures()[0].getCause().getMessage());
+
         SearchResponse response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments.messages", matchQuery("comments.messages.message", "fox"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder())).get();
+                        .innerHit(new InnerHitBuilder().setFetchSourceContext(new FetchSourceContext(false)))).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         SearchHit hit = response.getHits().getAt(0);
@@ -427,7 +444,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments.messages", matchQuery("comments.messages.message", "bear"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder())).get();
+                        .innerHit(new InnerHitBuilder().setFetchSourceContext(new FetchSourceContext(false)))).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         hit = response.getHits().getAt(0);
@@ -448,7 +465,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         indexRandom(true, requests);
         response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments.messages", matchQuery("comments.messages.message", "fox"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder())).get();
+                        .innerHit(new InnerHitBuilder().setFetchSourceContext(new FetchSourceContext(false)))).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         hit = response.getHits().getAt(0);;