Add the ability to define custom inner hit sub context builder (#24676)

This commit moves the handling of nested and parent/child inner hits to specialized classes that can be defined outside of ES core.
InnerHitBuilderContext is now used by the parent query (nested or hasChild, ...) to build the sub context from the InnerHitBuilder definition.
BWC is also ensured so that nodes in previous versions can still send/receive inner hits to/from this version.

Relates #20257

13 files changed, 393 insertions, 484 deletions

diff --git a/core/src/main/java/org/elasticsearch/index/query/AbstractQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/AbstractQueryBuilder.java
index 9d18e42138..72af541476 100644
--- a/core/src/main/java/org/elasticsearch/index/query/AbstractQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/AbstractQueryBuilder.java
@@ -283,7 +283,7 @@ public abstract class AbstractQueryBuilder<QB extends AbstractQueryBuilder<QB>>
      * Extracts the inner hits from the query tree.
      * While it extracts inner hits, child inner hits are inlined into the inner hit builder they belong to.
      */
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
     }
 
     // Like Objects.requireNotNull(...) but instead throws a IllegalArgumentException
diff --git a/core/src/main/java/org/elasticsearch/index/query/BoolQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/BoolQueryBuilder.java
index dcab99ce6a..c02ff4175c 100644
--- a/core/src/main/java/org/elasticsearch/index/query/BoolQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/BoolQueryBuilder.java
@@ -454,13 +454,13 @@ public class BoolQueryBuilder extends AbstractQueryBuilder<BoolQueryBuilder> {
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
         List<QueryBuilder> clauses = new ArrayList<>(filter());
         clauses.addAll(must());
         clauses.addAll(should());
         // no need to include must_not (since there will be no hits for it)
         for (QueryBuilder clause : clauses) {
-            InnerHitBuilder.extractInnerHits(clause, innerHits);
+            InnerHitContextBuilder.extractInnerHits(clause, innerHits);
         }
     }
 
diff --git a/core/src/main/java/org/elasticsearch/index/query/BoostingQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/BoostingQueryBuilder.java
index cb16faf9b3..11aaa90fba 100644
--- a/core/src/main/java/org/elasticsearch/index/query/BoostingQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/BoostingQueryBuilder.java
@@ -230,8 +230,8 @@ public class BoostingQueryBuilder extends AbstractQueryBuilder<BoostingQueryBuil
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
-        InnerHitBuilder.extractInnerHits(positiveQuery, innerHits);
-        InnerHitBuilder.extractInnerHits(negativeQuery, innerHits);
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
+        InnerHitContextBuilder.extractInnerHits(positiveQuery, innerHits);
+        InnerHitContextBuilder.extractInnerHits(negativeQuery, innerHits);
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/index/query/ConstantScoreQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/ConstantScoreQueryBuilder.java
index 96035b94b1..9308d3495d 100644
--- a/core/src/main/java/org/elasticsearch/index/query/ConstantScoreQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/ConstantScoreQueryBuilder.java
@@ -166,7 +166,7 @@ public class ConstantScoreQueryBuilder extends AbstractQueryBuilder<ConstantScor
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
-        InnerHitBuilder.extractInnerHits(filterBuilder, innerHits);
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
+        InnerHitContextBuilder.extractInnerHits(filterBuilder, innerHits);
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/index/query/DisMaxQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/DisMaxQueryBuilder.java
index bd8e903177..7bad1d0fe3 100644
--- a/core/src/main/java/org/elasticsearch/index/query/DisMaxQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/DisMaxQueryBuilder.java
@@ -209,9 +209,9 @@ public class DisMaxQueryBuilder extends AbstractQueryBuilder<DisMaxQueryBuilder>
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
         for (QueryBuilder query : queries) {
-            InnerHitBuilder.extractInnerHits(query, innerHits);
+            InnerHitContextBuilder.extractInnerHits(query, innerHits);
         }
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/index/query/InnerHitBuilder.java b/core/src/main/java/org/elasticsearch/index/query/InnerHitBuilder.java
index d21d3d6670..4419d6a93a 100644
--- a/core/src/main/java/org/elasticsearch/index/query/InnerHitBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/InnerHitBuilder.java
@@ -28,31 +28,21 @@ import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
-import org.elasticsearch.index.mapper.DocumentMapper;
-import org.elasticsearch.index.mapper.ObjectMapper;
 import org.elasticsearch.script.Script;
-import org.elasticsearch.script.ScriptContext;
-import org.elasticsearch.script.SearchScript;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder.ScriptField;
 import org.elasticsearch.search.fetch.StoredFieldsContext;
-import org.elasticsearch.search.fetch.subphase.DocValueFieldsContext;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
-import org.elasticsearch.search.fetch.subphase.InnerHitsContext;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
-import org.elasticsearch.search.internal.SearchContext;
-import org.elasticsearch.search.sort.SortAndFormats;
 import org.elasticsearch.search.sort.SortBuilder;
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 
 import static org.elasticsearch.common.xcontent.XContentParser.Token.END_OBJECT;
@@ -61,7 +51,6 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
 
     public static final ParseField NAME_FIELD = new ParseField("name");
     public static final ParseField IGNORE_UNMAPPED = new ParseField("ignore_unmapped");
-    public static final ParseField INNER_HITS_FIELD = new ParseField("inner_hits");
     public static final QueryBuilder DEFAULT_INNER_HIT_QUERY = new MatchAllQueryBuilder();
 
     private static final ObjectParser<InnerHitBuilder, QueryParseContext> PARSER = new ObjectParser<>("inner_hits", InnerHitBuilder::new);
@@ -104,35 +93,9 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
         }, SearchSourceBuilder._SOURCE_FIELD, ObjectParser.ValueType.OBJECT_ARRAY_BOOLEAN_OR_STRING);
         PARSER.declareObject(InnerHitBuilder::setHighlightBuilder, (p, c) -> HighlightBuilder.fromXContent(c),
                 SearchSourceBuilder.HIGHLIGHT_FIELD);
-        PARSER.declareObject(InnerHitBuilder::setChildInnerHits, (p, c) -> {
-            try {
-                Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-                String innerHitName = null;
-                for (XContentParser.Token token = p.nextToken(); token != XContentParser.Token.END_OBJECT; token = p.nextToken()) {
-                    switch (token) {
-                        case START_OBJECT:
-                            InnerHitBuilder innerHitBuilder = InnerHitBuilder.fromXContent(c);
-                            innerHitBuilder.setName(innerHitName);
-                            innerHitBuilders.put(innerHitName, innerHitBuilder);
-                            break;
-                        case FIELD_NAME:
-                            innerHitName = p.currentName();
-                            break;
-                        default:
-                            throw new ParsingException(p.getTokenLocation(), "Expected [" + XContentParser.Token.START_OBJECT + "] in ["
-                                    + p.currentName() + "] but found [" + token + "]", p.getTokenLocation());
-                    }
-                }
-                return innerHitBuilders;
-            } catch (IOException e) {
-                throw new ParsingException(p.getTokenLocation(), "Could not parse inner query definition", e);
-            }
-        }, INNER_HITS_FIELD);
     }
 
     private String name;
-    private String nestedPath;
-    private String parentChildType;
     private boolean ignoreUnmapped;
 
     private int from;
@@ -148,71 +111,25 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
     private Set<ScriptField> scriptFields;
     private HighlightBuilder highlightBuilder;
     private FetchSourceContext fetchSourceContext;
-    private Map<String, InnerHitBuilder> childInnerHits;
 
     public InnerHitBuilder() {
+        this.name = null;
     }
 
-    private InnerHitBuilder(InnerHitBuilder other) {
-        name = other.name;
-        this.ignoreUnmapped = other.ignoreUnmapped;
-        from = other.from;
-        size = other.size;
-        explain = other.explain;
-        version = other.version;
-        trackScores = other.trackScores;
-        if (other.storedFieldsContext != null) {
-            storedFieldsContext = new StoredFieldsContext(other.storedFieldsContext);
-        }
-        if (other.docValueFields != null) {
-            docValueFields = new ArrayList<> (other.docValueFields);
-        }
-        if (other.scriptFields != null) {
-            scriptFields = new HashSet<> (other.scriptFields);
-        }
-        if (other.fetchSourceContext != null) {
-            fetchSourceContext = new FetchSourceContext(
-                    other.fetchSourceContext.fetchSource(), other.fetchSourceContext.includes(), other.fetchSourceContext.excludes()
-            );
-        }
-        if (other.sorts != null) {
-            sorts = new ArrayList<>(other.sorts);
-        }
-        highlightBuilder = other.highlightBuilder;
-        if (other.childInnerHits != null) {
-            childInnerHits = new HashMap<>(other.childInnerHits);
-        }
-    }
-
-
-    InnerHitBuilder(InnerHitBuilder other, String nestedPath, QueryBuilder query, boolean ignoreUnmapped) {
-        this(other);
-        this.query = query;
-        this.nestedPath = nestedPath;
-        this.ignoreUnmapped = ignoreUnmapped;
-        if (name == null) {
-            this.name = nestedPath;
-        }
+    public InnerHitBuilder(String name) {
+        this.name = name;
     }
 
-    // NORELEASE Do not use this ctr, it is public for hasChild and hasParent query but this is temporary
-    public InnerHitBuilder(InnerHitBuilder other, QueryBuilder query, String parentChildType, boolean ignoreUnmapped) {
-        this(other);
-        this.query = query;
-        this.parentChildType = parentChildType;
-        this.ignoreUnmapped = ignoreUnmapped;
-        if (name == null) {
-            this.name = parentChildType;
-        }
-    }
 
     /**
      * Read from a stream.
      */
     public InnerHitBuilder(StreamInput in) throws IOException {
         name = in.readOptionalString();
-        nestedPath = in.readOptionalString();
-        parentChildType = in.readOptionalString();
+        if (in.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) {
+            in.readOptionalString();
+            in.readOptionalString();
+        }
         if (in.getVersion().onOrAfter(Version.V_5_2_0)) {
             ignoreUnmapped = in.readBoolean();
         }
@@ -239,21 +156,94 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
             }
         }
         highlightBuilder = in.readOptionalWriteable(HighlightBuilder::new);
-        query = in.readNamedWriteable(QueryBuilder.class);
-        if (in.readBoolean()) {
-            int size = in.readVInt();
-            childInnerHits = new HashMap<>(size);
-            for (int i = 0; i < size; i++) {
-                childInnerHits.put(in.readString(), new InnerHitBuilder(in));
-            }
+        if (in.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) {
+            /**
+             * this is needed for BWC with nodes pre 5.5
+             */
+            in.readNamedWriteable(QueryBuilder.class);
+            boolean hasChildren = in.readBoolean();
+            assert hasChildren == false;
         }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) {
+            throw new IOException("Invalid output version, must >= " + Version.V_6_0_0_alpha2_UNRELEASED.toString());
+        }
         out.writeOptionalString(name);
-        out.writeOptionalString(nestedPath);
-        out.writeOptionalString(parentChildType);
+        out.writeBoolean(ignoreUnmapped);
+        out.writeVInt(from);
+        out.writeVInt(size);
+        out.writeBoolean(explain);
+        out.writeBoolean(version);
+        out.writeBoolean(trackScores);
+        out.writeOptionalWriteable(storedFieldsContext);
+        out.writeGenericValue(docValueFields);
+        boolean hasScriptFields = scriptFields != null;
+        out.writeBoolean(hasScriptFields);
+        if (hasScriptFields) {
+            out.writeVInt(scriptFields.size());
+            Iterator<ScriptField> iterator = scriptFields.stream()
+                    .sorted(Comparator.comparing(ScriptField::fieldName)).iterator();
+            while (iterator.hasNext()) {
+                iterator.next().writeTo(out);
+            }
+        }
+        out.writeOptionalWriteable(fetchSourceContext);
+        boolean hasSorts = sorts != null;
+        out.writeBoolean(hasSorts);
+        if (hasSorts) {
+            out.writeVInt(sorts.size());
+            for (SortBuilder<?> sort : sorts) {
+                out.writeNamedWriteable(sort);
+            }
+        }
+        out.writeOptionalWriteable(highlightBuilder);
+    }
+
+    /**
+     * BWC serialization for nested {@link InnerHitBuilder}.
+     * Should only be used to send nested inner hits to nodes pre 5.5.
+     */
+    protected void writeToNestedBWC(StreamOutput out, QueryBuilder query, String nestedPath) throws IOException {
+        assert out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED) :
+            "invalid output version, must be < " + Version.V_6_0_0_alpha2_UNRELEASED.toString();
+        writeToBWC(out, query, nestedPath, null);
+    }
+
+    /**
+     * BWC serialization for collapsing {@link InnerHitBuilder}.
+     * Should only be used to send collapsing inner hits to nodes pre 5.5.
+     */
+    public void writeToCollapseBWC(StreamOutput out) throws IOException {
+        assert out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED) :
+            "invalid output version, must be < " + Version.V_6_0_0_alpha2_UNRELEASED.toString();
+        writeToBWC(out, new MatchAllQueryBuilder(), null, null);
+    }
+
+    /**
+     * BWC serialization for parent/child {@link InnerHitBuilder}.
+     * Should only be used to send hasParent or hasChild inner hits to nodes pre 5.5.
+     */
+    public void writeToParentChildBWC(StreamOutput out, QueryBuilder query, String parentChildPath) throws IOException {
+        assert(out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) :
+            "invalid output version, must be < " + Version.V_6_0_0_alpha2_UNRELEASED.toString();
+        writeToBWC(out, query, null, parentChildPath);
+    }
+
+    private void writeToBWC(StreamOutput out,
+                            QueryBuilder query,
+                            String nestedPath,
+                            String parentChildPath) throws IOException {
+        out.writeOptionalString(name);
+        if (nestedPath != null) {
+            out.writeOptionalString(nestedPath);
+            out.writeOptionalString(null);
+        } else {
+            out.writeOptionalString(null);
+            out.writeOptionalString(parentChildPath);
+        }
         if (out.getVersion().onOrAfter(Version.V_5_2_0)) {
             out.writeBoolean(ignoreUnmapped);
         }
@@ -269,7 +259,7 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
         if (hasScriptFields) {
             out.writeVInt(scriptFields.size());
             Iterator<ScriptField> iterator = scriptFields.stream()
-                    .sorted((a, b) -> a.fieldName().compareTo(b.fieldName())).iterator();
+                .sorted(Comparator.comparing(ScriptField::fieldName)).iterator();
             while (iterator.hasNext()) {
                 iterator.next().writeTo(out);
             }
@@ -285,18 +275,7 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
         }
         out.writeOptionalWriteable(highlightBuilder);
         out.writeNamedWriteable(query);
-        boolean hasChildInnerHits = childInnerHits != null;
-        out.writeBoolean(hasChildInnerHits);
-        if (hasChildInnerHits) {
-            out.writeVInt(childInnerHits.size());
-            Iterator<Map.Entry<String, InnerHitBuilder>> iterator = childInnerHits.entrySet().stream()
-                    .sorted((a, b) -> a.getKey().compareTo(b.getKey())).iterator();
-            while (iterator.hasNext()) {
-                Map.Entry<String, InnerHitBuilder> entry = iterator.next();
-                out.writeString(entry.getKey());
-                entry.getValue().writeTo(out);
-            }
-        }
+        out.writeBoolean(false);
     }
 
     public String getName() {
@@ -308,6 +287,11 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
         return this;
     }
 
+    public InnerHitBuilder setIgnoreUnmapped(boolean value) {
+        this.ignoreUnmapped = value;
+        return this;
+    }
+
     /**
      * Whether to include inner hits in the search response hits if required mappings is missing
      */
@@ -525,136 +509,6 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
         return query;
     }
 
-    void setChildInnerHits(Map<String, InnerHitBuilder> childInnerHits) {
-        this.childInnerHits = childInnerHits;
-    }
-
-    String getParentChildType() {
-        return parentChildType;
-    }
-
-    String getNestedPath() {
-        return nestedPath;
-    }
-
-    void addChildInnerHit(InnerHitBuilder innerHitBuilder) {
-        if (childInnerHits == null) {
-            childInnerHits = new HashMap<>();
-        }
-        this.childInnerHits.put(innerHitBuilder.getName(), innerHitBuilder);
-    }
-
-    public InnerHitsContext.BaseInnerHits build(SearchContext parentSearchContext,
-                                                InnerHitsContext innerHitsContext) throws IOException {
-        QueryShardContext queryShardContext = parentSearchContext.getQueryShardContext();
-        if (nestedPath != null) {
-            ObjectMapper nestedObjectMapper = queryShardContext.getObjectMapper(nestedPath);
-            if (nestedObjectMapper == null) {
-                if (ignoreUnmapped == false) {
-                    throw new IllegalStateException("[" + query.getName() + "] no mapping found for type [" + nestedPath + "]");
-                } else {
-                    return null;
-                }
-            }
-
-            ObjectMapper parentObjectMapper = queryShardContext.nestedScope().nextLevel(nestedObjectMapper);
-            InnerHitsContext.NestedInnerHits nestedInnerHits = new InnerHitsContext.NestedInnerHits(
-                    name, parentSearchContext, parentObjectMapper, nestedObjectMapper
-            );
-            setupInnerHitsContext(queryShardContext, nestedInnerHits);
-            if (childInnerHits != null) {
-                buildChildInnerHits(parentSearchContext, nestedInnerHits);
-            }
-            queryShardContext.nestedScope().previousLevel();
-            innerHitsContext.addInnerHitDefinition(nestedInnerHits);
-            return nestedInnerHits;
-        } else if (parentChildType != null) {
-            DocumentMapper documentMapper = queryShardContext.documentMapper(parentChildType);
-            if (documentMapper == null) {
-                if (ignoreUnmapped == false) {
-                    throw new IllegalStateException("[" + query.getName() + "] no mapping found for type [" + parentChildType + "]");
-                } else {
-                    return null;
-                }
-            }
-
-            InnerHitsContext.ParentChildInnerHits parentChildInnerHits = new InnerHitsContext.ParentChildInnerHits(
-                    name, parentSearchContext, queryShardContext.getMapperService(), documentMapper
-            );
-            setupInnerHitsContext(queryShardContext, parentChildInnerHits);
-            if (childInnerHits != null) {
-                buildChildInnerHits(parentSearchContext, parentChildInnerHits);
-            }
-            innerHitsContext.addInnerHitDefinition( parentChildInnerHits);
-            return parentChildInnerHits;
-        } else {
-            throw new IllegalStateException("Neither a nested or parent/child inner hit");
-        }
-    }
-
-    private void buildChildInnerHits(SearchContext parentSearchContext, InnerHitsContext.BaseInnerHits innerHits) throws IOException {
-        Map<String, InnerHitsContext.BaseInnerHits> childInnerHits = new HashMap<>();
-        for (Map.Entry<String, InnerHitBuilder> entry : this.childInnerHits.entrySet()) {
-            InnerHitsContext.BaseInnerHits childInnerHit = entry.getValue().build(
-                    parentSearchContext, new InnerHitsContext()
-            );
-            if (childInnerHit != null) {
-                childInnerHits.put(entry.getKey(), childInnerHit);
-            }
-        }
-        innerHits.setChildInnerHits(childInnerHits);
-    }
-
-    private void setupInnerHitsContext(QueryShardContext context, InnerHitsContext.BaseInnerHits innerHitsContext) throws IOException {
-        innerHitsContext.from(from);
-        innerHitsContext.size(size);
-        innerHitsContext.explain(explain);
-        innerHitsContext.version(version);
-        innerHitsContext.trackScores(trackScores);
-        if (storedFieldsContext != null) {
-            innerHitsContext.storedFieldsContext(storedFieldsContext);
-        }
-        if (docValueFields != null) {
-            innerHitsContext.docValueFieldsContext(new DocValueFieldsContext(docValueFields));
-        }
-        if (scriptFields != null) {
-            for (ScriptField field : scriptFields) {
-                SearchScript searchScript = innerHitsContext.getQueryShardContext().getSearchScript(field.script(),
-                    ScriptContext.SEARCH);
-                innerHitsContext.scriptFields().add(new org.elasticsearch.search.fetch.subphase.ScriptFieldsContext.ScriptField(
-                        field.fieldName(), searchScript, field.ignoreFailure()));
-            }
-        }
-        if (fetchSourceContext != null) {
-            innerHitsContext.fetchSourceContext(fetchSourceContext);
-        }
-        if (sorts != null) {
-            Optional<SortAndFormats> optionalSort = SortBuilder.buildSort(sorts, context);
-            if (optionalSort.isPresent()) {
-                innerHitsContext.sort(optionalSort.get());
-            }
-        }
-        if (highlightBuilder != null) {
-            innerHitsContext.highlight(highlightBuilder.build(context));
-        }
-        ParsedQuery parsedQuery = new ParsedQuery(query.toQuery(context), context.copyNamedQueries());
-        innerHitsContext.parsedQuery(parsedQuery);
-    }
-
-    public void inlineInnerHits(Map<String, InnerHitBuilder> innerHits) {
-        InnerHitBuilder copy = new InnerHitBuilder(this);
-        copy.parentChildType = this.parentChildType;
-        copy.nestedPath = this.nestedPath;
-        copy.query = this.query;
-        innerHits.put(copy.getName(), copy);
-
-        Map<String, InnerHitBuilder> childInnerHits = new HashMap<>();
-        extractInnerHits(query, childInnerHits);
-        if (childInnerHits.size() > 0) {
-            copy.setChildInnerHits(childInnerHits);
-        }
-    }
-
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -697,13 +551,6 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
         if (highlightBuilder != null) {
             builder.field(SearchSourceBuilder.HIGHLIGHT_FIELD.getPreferredName(), highlightBuilder, params);
         }
-        if (childInnerHits != null) {
-            builder.startObject(INNER_HITS_FIELD.getPreferredName());
-            for (Map.Entry<String, InnerHitBuilder> entry : childInnerHits.entrySet()) {
-                builder.field(entry.getKey(), entry.getValue(), params);
-            }
-            builder.endObject();
-        }
         builder.endObject();
         return builder;
     }
@@ -715,8 +562,6 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
 
         InnerHitBuilder that = (InnerHitBuilder) o;
         return Objects.equals(name, that.name) &&
-                Objects.equals(nestedPath, that.nestedPath) &&
-                Objects.equals(parentChildType, that.parentChildType) &&
                 Objects.equals(ignoreUnmapped, that.ignoreUnmapped) &&
                 Objects.equals(from, that.from) &&
                 Objects.equals(size, that.size) &&
@@ -728,41 +573,16 @@ public final class InnerHitBuilder extends ToXContentToBytes implements Writeabl
                 Objects.equals(scriptFields, that.scriptFields) &&
                 Objects.equals(fetchSourceContext, that.fetchSourceContext) &&
                 Objects.equals(sorts, that.sorts) &&
-                Objects.equals(highlightBuilder, that.highlightBuilder) &&
-                Objects.equals(query, that.query) &&
-                Objects.equals(childInnerHits, that.childInnerHits);
+                Objects.equals(highlightBuilder, that.highlightBuilder);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, nestedPath, parentChildType, ignoreUnmapped, from, size, explain, version, trackScores,
-                storedFieldsContext, docValueFields, scriptFields, fetchSourceContext, sorts, highlightBuilder, query, childInnerHits);
+        return Objects.hash(name, ignoreUnmapped, from, size, explain, version, trackScores,
+                storedFieldsContext, docValueFields, scriptFields, fetchSourceContext, sorts, highlightBuilder);
     }
 
     public static InnerHitBuilder fromXContent(QueryParseContext context) throws IOException {
         return PARSER.parse(context.parser(), new InnerHitBuilder(), context);
     }
-
-    public static void extractInnerHits(QueryBuilder query, Map<String, InnerHitBuilder> innerHitBuilders) {
-        if (query instanceof AbstractQueryBuilder) {
-            ((AbstractQueryBuilder) query).extractInnerHitBuilders(innerHitBuilders);
-        } else {
-            throw new IllegalStateException("provided query builder [" + query.getClass() +
-                    "] class should inherit from AbstractQueryBuilder, but it doesn't");
-        }
-    }
-
-    // TODO public for hasParent and hasChild query
-    public static InnerHitBuilder rewrite(InnerHitBuilder original, QueryBuilder rewrittenQuery) {
-        if (original == null) {
-            return null;
-        }
-
-        InnerHitBuilder copy = new InnerHitBuilder(original);
-        copy.query = rewrittenQuery;
-        copy.parentChildType = original.parentChildType;
-        copy.nestedPath = original.nestedPath;
-        return copy;
-    }
-
 }
diff --git a/core/src/main/java/org/elasticsearch/index/query/InnerHitContextBuilder.java b/core/src/main/java/org/elasticsearch/index/query/InnerHitContextBuilder.java
new file mode 100644
index 0000000000..5863d02f90
--- /dev/null
+++ b/core/src/main/java/org/elasticsearch/index/query/InnerHitContextBuilder.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.query;
+
+import org.elasticsearch.script.ScriptContext;
+import org.elasticsearch.script.SearchScript;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.fetch.subphase.DocValueFieldsContext;
+import org.elasticsearch.search.fetch.subphase.InnerHitsContext;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.search.sort.SortAndFormats;
+import org.elasticsearch.search.sort.SortBuilder;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * A builder for {@link InnerHitsContext.InnerHitSubContext}
+ */
+public abstract class InnerHitContextBuilder {
+    protected final QueryBuilder query;
+    protected final InnerHitBuilder innerHitBuilder;
+    protected final Map<String, InnerHitContextBuilder> children;
+
+    protected InnerHitContextBuilder(QueryBuilder query, InnerHitBuilder innerHitBuilder, Map<String, InnerHitContextBuilder> children) {
+        this.innerHitBuilder = innerHitBuilder;
+        this.children = children;
+        this.query = query;
+    }
+
+    public abstract void build(SearchContext parentSearchContext,
+                                InnerHitsContext innerHitsContext) throws IOException;
+
+    public static void extractInnerHits(QueryBuilder query, Map<String, InnerHitContextBuilder> innerHitBuilders) {
+        if (query instanceof AbstractQueryBuilder) {
+            ((AbstractQueryBuilder) query).extractInnerHitBuilders(innerHitBuilders);
+        } else {
+            throw new IllegalStateException("provided query builder [" + query.getClass() +
+                "] class should inherit from AbstractQueryBuilder, but it doesn't");
+        }
+    }
+
+    protected void setupInnerHitsContext(QueryShardContext queryShardContext,
+                                         InnerHitsContext.InnerHitSubContext innerHitsContext) throws IOException {
+        innerHitsContext.from(innerHitBuilder.getFrom());
+        innerHitsContext.size(innerHitBuilder.getSize());
+        innerHitsContext.explain(innerHitBuilder.isExplain());
+        innerHitsContext.version(innerHitBuilder.isVersion());
+        innerHitsContext.trackScores(innerHitBuilder.isTrackScores());
+        if (innerHitBuilder.getStoredFieldsContext() != null) {
+            innerHitsContext.storedFieldsContext(innerHitBuilder.getStoredFieldsContext());
+        }
+        if (innerHitBuilder.getDocValueFields() != null) {
+            innerHitsContext.docValueFieldsContext(new DocValueFieldsContext(innerHitBuilder.getDocValueFields()));
+        }
+        if (innerHitBuilder.getScriptFields() != null) {
+            for (SearchSourceBuilder.ScriptField field : innerHitBuilder.getScriptFields()) {
+                SearchScript searchScript = innerHitsContext.getQueryShardContext().getSearchScript(field.script(),
+                    ScriptContext.SEARCH);
+                innerHitsContext.scriptFields().add(new org.elasticsearch.search.fetch.subphase.ScriptFieldsContext.ScriptField(
+                    field.fieldName(), searchScript, field.ignoreFailure()));
+            }
+        }
+        if (innerHitBuilder.getFetchSourceContext() != null) {
+            innerHitsContext.fetchSourceContext(innerHitBuilder.getFetchSourceContext() );
+        }
+        if (innerHitBuilder.getSorts() != null) {
+            Optional<SortAndFormats> optionalSort = SortBuilder.buildSort(innerHitBuilder.getSorts(), queryShardContext);
+            if (optionalSort.isPresent()) {
+                innerHitsContext.sort(optionalSort.get());
+            }
+        }
+        if (innerHitBuilder.getHighlightBuilder() != null) {
+            innerHitsContext.highlight(innerHitBuilder.getHighlightBuilder().build(queryShardContext));
+        }
+        ParsedQuery parsedQuery = new ParsedQuery(query.toQuery(queryShardContext), queryShardContext.copyNamedQueries());
+        innerHitsContext.parsedQuery(parsedQuery);
+        Map<String, InnerHitsContext.InnerHitSubContext> baseChildren =
+            buildChildInnerHits(innerHitsContext.parentSearchContext(), children);
+        innerHitsContext.setChildInnerHits(baseChildren);
+    }
+
+    private static Map<String, InnerHitsContext.InnerHitSubContext> buildChildInnerHits(SearchContext parentSearchContext,
+                                Map<String, InnerHitContextBuilder> children) throws IOException {
+
+        Map<String, InnerHitsContext.InnerHitSubContext> childrenInnerHits = new HashMap<>();
+        for (Map.Entry<String, InnerHitContextBuilder> entry : children.entrySet()) {
+            InnerHitsContext childInnerHitsContext = new InnerHitsContext();
+            entry.getValue().build(
+                parentSearchContext, childInnerHitsContext);
+            if (childInnerHitsContext.getInnerHits() != null) {
+                childrenInnerHits.putAll(childInnerHitsContext.getInnerHits());
+            }
+        }
+        return childrenInnerHits;
+    }
+}
diff --git a/core/src/main/java/org/elasticsearch/index/query/NestedQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/NestedQueryBuilder.java
index 2d23f256f0..5ebb074b79 100644
--- a/core/src/main/java/org/elasticsearch/index/query/NestedQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/NestedQueryBuilder.java
@@ -19,26 +19,43 @@
 
 package org.elasticsearch.index.query;
 
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.ReaderUtil;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TopDocs;
+import org.apache.lucene.search.TopDocsCollector;
+import org.apache.lucene.search.TopFieldCollector;
+import org.apache.lucene.search.TopScoreDocCollector;
+import org.apache.lucene.search.TotalHitCountCollector;
+import org.apache.lucene.search.Weight;
 import org.apache.lucene.search.join.BitSetProducer;
+import org.apache.lucene.search.join.ParentChildrenBlockJoinQuery;
 import org.apache.lucene.search.join.ScoreMode;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.lucene.search.Queries;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.mapper.ObjectMapper;
 import org.elasticsearch.index.search.ESToParentBlockJoinQuery;
 import org.elasticsearch.index.search.NestedHelper;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.fetch.subphase.InnerHitsContext;
+import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 
+import static org.elasticsearch.search.fetch.subphase.InnerHitsContext.intersect;
+
 public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder> {
     public static final String NAME = "nested";
     /**
@@ -86,7 +103,15 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
         out.writeString(path);
         out.writeVInt(scoreMode.ordinal());
         out.writeNamedWriteable(query);
-        out.writeOptionalWriteable(innerHitBuilder);
+        if (out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) {
+            final boolean hasInnerHit = innerHitBuilder != null;
+            out.writeBoolean(hasInnerHit);
+            if (hasInnerHit) {
+                innerHitBuilder.writeToNestedBWC(out, query, path);
+            }
+        } else {
+            out.writeOptionalWriteable(innerHitBuilder);
+        }
         out.writeBoolean(ignoreUnmapped);
     }
 
@@ -105,8 +130,8 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
         return innerHitBuilder;
     }
 
-    public NestedQueryBuilder innerHit(InnerHitBuilder innerHit, boolean ignoreUnmapped) {
-        this.innerHitBuilder = new InnerHitBuilder(innerHit, path, query, ignoreUnmapped);
+    public NestedQueryBuilder innerHit(InnerHitBuilder innerHitBuilder) {
+        this.innerHitBuilder = innerHitBuilder;
         return this;
     }
 
@@ -191,13 +216,10 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
                 }
             }
         }
-        NestedQueryBuilder queryBuilder =  new NestedQueryBuilder(path, query, scoreMode)
-                .ignoreUnmapped(ignoreUnmapped)
-                .queryName(queryName)
-                .boost(boost);
-        if (innerHitBuilder != null) {
-            queryBuilder.innerHit(innerHitBuilder, ignoreUnmapped);
-        }
+        NestedQueryBuilder queryBuilder =  new NestedQueryBuilder(path, query, scoreMode, innerHitBuilder)
+            .ignoreUnmapped(ignoreUnmapped)
+            .queryName(queryName)
+            .boost(boost);
         return queryBuilder;
     }
 
@@ -287,8 +309,7 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
     protected QueryBuilder doRewrite(QueryRewriteContext queryRewriteContext) throws IOException {
         QueryBuilder rewrittenQuery = query.rewrite(queryRewriteContext);
         if (rewrittenQuery != query) {
-            InnerHitBuilder rewrittenInnerHit = InnerHitBuilder.rewrite(innerHitBuilder, rewrittenQuery);
-            NestedQueryBuilder nestedQuery = new NestedQueryBuilder(path, rewrittenQuery, scoreMode, rewrittenInnerHit);
+            NestedQueryBuilder nestedQuery = new NestedQueryBuilder(path, rewrittenQuery, scoreMode, innerHitBuilder);
             nestedQuery.ignoreUnmapped(ignoreUnmapped);
             return nestedQuery;
         }
@@ -296,9 +317,102 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
+    public void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
         if (innerHitBuilder != null) {
-            innerHitBuilder.inlineInnerHits(innerHits);
+            Map<String, InnerHitContextBuilder> children = new HashMap<>();
+            InnerHitContextBuilder.extractInnerHits(query, children);
+            InnerHitContextBuilder innerHitContextBuilder = new NestedInnerHitContextBuilder(path, query, innerHitBuilder, children);
+            String name = innerHitBuilder.getName() != null ? innerHitBuilder.getName() : path;
+            innerHits.put(name, innerHitContextBuilder);
+        }
+    }
+
+    static class NestedInnerHitContextBuilder extends InnerHitContextBuilder {
+        private final String path;
+
+        NestedInnerHitContextBuilder(String path, QueryBuilder query, InnerHitBuilder innerHitBuilder,
+                                     Map<String, InnerHitContextBuilder> children) {
+            super(query, innerHitBuilder, children);
+            this.path = path;
+        }
+
+        @Override
+        public void build(SearchContext parentSearchContext,
+                          InnerHitsContext innerHitsContext) throws IOException {
+            QueryShardContext queryShardContext = parentSearchContext.getQueryShardContext();
+            ObjectMapper nestedObjectMapper = queryShardContext.getObjectMapper(path);
+            if (nestedObjectMapper == null) {
+                if (innerHitBuilder.isIgnoreUnmapped() == false) {
+                    throw new IllegalStateException("[" + query.getName() + "] no mapping found for type [" + path + "]");
+                } else {
+                    return;
+                }
+            }
+            String name =  innerHitBuilder.getName() != null ? innerHitBuilder.getName() : nestedObjectMapper.fullPath();
+            ObjectMapper parentObjectMapper = queryShardContext.nestedScope().nextLevel(nestedObjectMapper);
+            NestedInnerHitSubContext nestedInnerHits = new NestedInnerHitSubContext(
+                name, parentSearchContext, parentObjectMapper, nestedObjectMapper
+            );
+            setupInnerHitsContext(queryShardContext, nestedInnerHits);
+            queryShardContext.nestedScope().previousLevel();
+            innerHitsContext.addInnerHitDefinition(nestedInnerHits);
+        }
+    }
+
+    static final class NestedInnerHitSubContext extends InnerHitsContext.InnerHitSubContext {
+
+        private final ObjectMapper parentObjectMapper;
+        private final ObjectMapper childObjectMapper;
+
+        NestedInnerHitSubContext(String name, SearchContext context, ObjectMapper parentObjectMapper, ObjectMapper childObjectMapper) {
+            super(name, context);
+            this.parentObjectMapper = parentObjectMapper;
+            this.childObjectMapper = childObjectMapper;
+        }
+
+        @Override
+        public TopDocs[] topDocs(SearchHit[] hits) throws IOException {
+            Weight innerHitQueryWeight = createInnerHitQueryWeight();
+            TopDocs[] result = new TopDocs[hits.length];
+            for (int i = 0; i < hits.length; i++) {
+                SearchHit hit = hits[i];
+                Query rawParentFilter;
+                if (parentObjectMapper == null) {
+                    rawParentFilter = Queries.newNonNestedFilter();
+                } else {
+                    rawParentFilter = parentObjectMapper.nestedTypeFilter();
+                }
+
+                int parentDocId = hit.docId();
+                final int readerIndex = ReaderUtil.subIndex(parentDocId, searcher().getIndexReader().leaves());
+                // With nested inner hits the nested docs are always in the same segement, so need to use the other segments
+                LeafReaderContext ctx = searcher().getIndexReader().leaves().get(readerIndex);
+
+                Query childFilter = childObjectMapper.nestedTypeFilter();
+                BitSetProducer parentFilter = context.bitsetFilterCache().getBitSetProducer(rawParentFilter);
+                Query q = new ParentChildrenBlockJoinQuery(parentFilter, childFilter, parentDocId);
+                Weight weight = context.searcher().createNormalizedWeight(q, false);
+                if (size() == 0) {
+                    TotalHitCountCollector totalHitCountCollector = new TotalHitCountCollector();
+                    intersect(weight, innerHitQueryWeight, totalHitCountCollector, ctx);
+                    result[i] = new TopDocs(totalHitCountCollector.getTotalHits(), Lucene.EMPTY_SCORE_DOCS, 0);
+                } else {
+                    int topN = Math.min(from() + size(), context.searcher().getIndexReader().maxDoc());
+                    TopDocsCollector<?> topDocsCollector;
+                    if (sort() != null) {
+                        topDocsCollector = TopFieldCollector.create(sort().sort, topN, true, trackScores(), trackScores());
+                    } else {
+                        topDocsCollector = TopScoreDocCollector.create(topN);
+                    }
+                    try {
+                        intersect(weight, innerHitQueryWeight, topDocsCollector, ctx);
+                    } finally {
+                        clearReleasables(Lifetime.COLLECTION);
+                    }
+                    result[i] = topDocsCollector.topDocs(from(), size());
+                }
+            }
+            return result;
         }
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/index/query/functionscore/FunctionScoreQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/functionscore/FunctionScoreQueryBuilder.java
index fcb4ab4ddc..a84aa678e5 100644
--- a/core/src/main/java/org/elasticsearch/index/query/functionscore/FunctionScoreQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/functionscore/FunctionScoreQueryBuilder.java
@@ -36,7 +36,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentLocation;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
-import org.elasticsearch.index.query.InnerHitBuilder;
+import org.elasticsearch.index.query.InnerHitContextBuilder;
 import org.elasticsearch.index.query.MatchAllQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryParseContext;
@@ -431,8 +431,8 @@ public class FunctionScoreQueryBuilder extends AbstractQueryBuilder<FunctionScor
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
-        InnerHitBuilder.extractInnerHits(query(), innerHits);
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
+        InnerHitContextBuilder.extractInnerHits(query(), innerHits);
     }
 
     public static FunctionScoreQueryBuilder fromXContent(QueryParseContext parseContext) throws IOException {
diff --git a/core/src/main/java/org/elasticsearch/search/SearchService.java b/core/src/main/java/org/elasticsearch/search/SearchService.java
index ac7c33094e..a5ad038fd5 100644
--- a/core/src/main/java/org/elasticsearch/search/SearchService.java
+++ b/core/src/main/java/org/elasticsearch/search/SearchService.java
@@ -41,7 +41,7 @@ import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.Engine;
-import org.elasticsearch.index.query.InnerHitBuilder;
+import org.elasticsearch.index.query.InnerHitContextBuilder;
 import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.index.shard.IndexEventListener;
 import org.elasticsearch.index.shard.IndexShard;
@@ -603,17 +603,17 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
         QueryShardContext queryShardContext = context.getQueryShardContext();
         context.from(source.from());
         context.size(source.size());
-        Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
+        Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
         if (source.query() != null) {
-            InnerHitBuilder.extractInnerHits(source.query(), innerHitBuilders);
+            InnerHitContextBuilder.extractInnerHits(source.query(), innerHitBuilders);
             context.parsedQuery(queryShardContext.toQuery(source.query()));
         }
         if (source.postFilter() != null) {
-            InnerHitBuilder.extractInnerHits(source.postFilter(), innerHitBuilders);
+            InnerHitContextBuilder.extractInnerHits(source.postFilter(), innerHitBuilders);
             context.parsedPostFilter(queryShardContext.toQuery(source.postFilter()));
         }
         if (innerHitBuilders.size() > 0) {
-            for (Map.Entry<String, InnerHitBuilder> entry : innerHitBuilders.entrySet()) {
+            for (Map.Entry<String, InnerHitContextBuilder> entry : innerHitBuilders.entrySet()) {
                 try {
                     entry.getValue().build(context, context.innerHits());
                 } catch (IOException e) {
diff --git a/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java b/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java
index 542ae2c3ab..d89e50c8a4 100644
--- a/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java
+++ b/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java
@@ -19,6 +19,7 @@
 package org.elasticsearch.search.collapse;
 
 import org.apache.lucene.index.IndexOptions;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.support.ToXContentToBytes;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
@@ -81,7 +82,15 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(field);
         out.writeVInt(maxConcurrentGroupRequests);
-        out.writeOptionalWriteable(innerHit);
+        if (out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) {
+            final boolean hasInnerHit = innerHit != null;
+            out.writeBoolean(hasInnerHit);
+            if (hasInnerHit) {
+                innerHit.writeToCollapseBWC(out);
+            }
+        } else {
+            out.writeOptionalWriteable(innerHit);
+        }
     }
 
     public static CollapseBuilder fromXContent(QueryParseContext context) throws IOException {
diff --git a/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsContext.java b/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsContext.java
index c8eee3e91e..39f80ef498 100644
--- a/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsContext.java
+++ b/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsContext.java
@@ -20,38 +20,18 @@
 package org.elasticsearch.search.fetch.subphase;
 
 import org.apache.lucene.index.LeafReaderContext;
-import org.apache.lucene.index.ReaderUtil;
-import org.apache.lucene.index.Term;
-import org.apache.lucene.search.BooleanClause.Occur;
-import org.apache.lucene.search.BooleanQuery;
 import org.apache.lucene.search.CollectionTerminatedException;
 import org.apache.lucene.search.Collector;
 import org.apache.lucene.search.ConjunctionDISI;
 import org.apache.lucene.search.DocIdSetIterator;
-import org.apache.lucene.search.DocValuesTermsQuery;
 import org.apache.lucene.search.LeafCollector;
-import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.Scorer;
 import org.apache.lucene.search.ScorerSupplier;
-import org.apache.lucene.search.TermQuery;
 import org.apache.lucene.search.TopDocs;
-import org.apache.lucene.search.TopDocsCollector;
-import org.apache.lucene.search.TopFieldCollector;
-import org.apache.lucene.search.TopScoreDocCollector;
-import org.apache.lucene.search.TotalHitCountCollector;
 import org.apache.lucene.search.Weight;
-import org.apache.lucene.search.join.BitSetProducer;
-import org.apache.lucene.search.join.ParentChildrenBlockJoinQuery;
 import org.apache.lucene.util.Bits;
-import org.elasticsearch.common.lucene.Lucene;
-import org.elasticsearch.common.lucene.search.Queries;
-import org.elasticsearch.index.mapper.DocumentMapper;
-import org.elasticsearch.index.mapper.MapperService;
-import org.elasticsearch.index.mapper.ObjectMapper;
-import org.elasticsearch.index.mapper.ParentFieldMapper;
 import org.elasticsearch.search.SearchHit;
-import org.elasticsearch.search.SearchHitField;
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.search.internal.SubSearchContext;
 
@@ -61,23 +41,25 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 
+/**
+ * Context used for inner hits retrieval
+ */
 public final class InnerHitsContext {
-
-    private final Map<String, BaseInnerHits> innerHits;
+    private final Map<String, InnerHitSubContext> innerHits;
 
     public InnerHitsContext() {
         this.innerHits = new HashMap<>();
     }
 
-    InnerHitsContext(Map<String, BaseInnerHits> innerHits) {
+    InnerHitsContext(Map<String, InnerHitSubContext> innerHits) {
         this.innerHits = Objects.requireNonNull(innerHits);
     }
 
-    public Map<String, BaseInnerHits> getInnerHits() {
+    public Map<String, InnerHitSubContext> getInnerHits() {
         return innerHits;
     }
 
-    public void addInnerHitDefinition(BaseInnerHits innerHit) {
+    public void addInnerHitDefinition(InnerHitSubContext innerHit) {
         if (innerHits.containsKey(innerHit.getName())) {
             throw new IllegalArgumentException("inner_hit definition with the name [" + innerHit.getName() +
                     "] already exists. Use a different inner_hit name or define one explicitly");
@@ -86,13 +68,17 @@ public final class InnerHitsContext {
         innerHits.put(innerHit.getName(), innerHit);
     }
 
-    public abstract static class BaseInnerHits extends SubSearchContext {
+    /**
+     * A {@link SubSearchContext} that associates {@link TopDocs} to each {@link SearchHit}
+     * in the parent search context
+     */
+    public abstract static class InnerHitSubContext extends SubSearchContext {
 
         private final String name;
-        final SearchContext context;
+        protected final SearchContext context;
         private InnerHitsContext childInnerHits;
 
-        BaseInnerHits(String name, SearchContext context) {
+        protected InnerHitSubContext(String name, SearchContext context) {
             super(context);
             this.name = name;
             this.context = context;
@@ -109,158 +95,22 @@ public final class InnerHitsContext {
             return childInnerHits;
         }
 
-        public void setChildInnerHits(Map<String, InnerHitsContext.BaseInnerHits> childInnerHits) {
+        public void setChildInnerHits(Map<String, InnerHitSubContext> childInnerHits) {
             this.childInnerHits = new InnerHitsContext(childInnerHits);
         }
 
-        Weight createInnerHitQueryWeight() throws IOException {
+        protected Weight createInnerHitQueryWeight() throws IOException {
             final boolean needsScores = size() != 0 && (sort() == null || sort().sort.needsScores());
             return context.searcher().createNormalizedWeight(query(), needsScores);
         }
 
-    }
-
-    public static final class NestedInnerHits extends BaseInnerHits {
-
-        private final ObjectMapper parentObjectMapper;
-        private final ObjectMapper childObjectMapper;
-
-        public NestedInnerHits(String name, SearchContext context, ObjectMapper parentObjectMapper, ObjectMapper childObjectMapper) {
-            super(name != null ? name : childObjectMapper.fullPath(), context);
-            this.parentObjectMapper = parentObjectMapper;
-            this.childObjectMapper = childObjectMapper;
+        public SearchContext parentSearchContext() {
+            return context;
         }
 
-        @Override
-        public TopDocs[] topDocs(SearchHit[] hits) throws IOException {
-            Weight innerHitQueryWeight = createInnerHitQueryWeight();
-            TopDocs[] result = new TopDocs[hits.length];
-            for (int i = 0; i < hits.length; i++) {
-                SearchHit hit = hits[i];
-                Query rawParentFilter;
-                if (parentObjectMapper == null) {
-                    rawParentFilter = Queries.newNonNestedFilter();
-                } else {
-                    rawParentFilter = parentObjectMapper.nestedTypeFilter();
-                }
-
-                int parentDocId = hit.docId();
-                final int readerIndex = ReaderUtil.subIndex(parentDocId, searcher().getIndexReader().leaves());
-                // With nested inner hits the nested docs are always in the same segement, so need to use the other segments
-                LeafReaderContext ctx = searcher().getIndexReader().leaves().get(readerIndex);
-
-                Query childFilter = childObjectMapper.nestedTypeFilter();
-                BitSetProducer parentFilter = context.bitsetFilterCache().getBitSetProducer(rawParentFilter);
-                Query q = new ParentChildrenBlockJoinQuery(parentFilter, childFilter, parentDocId);
-                Weight weight = context.searcher().createNormalizedWeight(q, false);
-                if (size() == 0) {
-                    TotalHitCountCollector totalHitCountCollector = new TotalHitCountCollector();
-                    intersect(weight, innerHitQueryWeight, totalHitCountCollector, ctx);
-                    result[i] = new TopDocs(totalHitCountCollector.getTotalHits(), Lucene.EMPTY_SCORE_DOCS, 0);
-                } else {
-                    int topN = Math.min(from() + size(), context.searcher().getIndexReader().maxDoc());
-                    TopDocsCollector<?> topDocsCollector;
-                    if (sort() != null) {
-                        topDocsCollector = TopFieldCollector.create(sort().sort, topN, true, trackScores(), trackScores());
-                    } else {
-                        topDocsCollector = TopScoreDocCollector.create(topN);
-                    }
-                    try {
-                        intersect(weight, innerHitQueryWeight, topDocsCollector, ctx);
-                    } finally {
-                        clearReleasables(Lifetime.COLLECTION);
-                    }
-                    result[i] = topDocsCollector.topDocs(from(), size());
-                }
-            }
-            return result;
-        }
-    }
-
-    public static final class ParentChildInnerHits extends BaseInnerHits {
-
-        private final MapperService mapperService;
-        private final DocumentMapper documentMapper;
-
-        public ParentChildInnerHits(String name, SearchContext context, MapperService mapperService, DocumentMapper documentMapper) {
-            super(name != null ? name : documentMapper.type(), context);
-            this.mapperService = mapperService;
-            this.documentMapper = documentMapper;
-        }
-
-        @Override
-        public TopDocs[] topDocs(SearchHit[] hits) throws IOException {
-            Weight innerHitQueryWeight = createInnerHitQueryWeight();
-            TopDocs[] result = new TopDocs[hits.length];
-            for (int i = 0; i < hits.length; i++) {
-                SearchHit hit = hits[i];
-                final Query hitQuery;
-                if (isParentHit(hit)) {
-                    String field = ParentFieldMapper.joinField(hit.getType());
-                    hitQuery = new DocValuesTermsQuery(field, hit.getId());
-                } else if (isChildHit(hit)) {
-                    DocumentMapper hitDocumentMapper = mapperService.documentMapper(hit.getType());
-                    final String parentType = hitDocumentMapper.parentFieldMapper().type();
-                    SearchHitField parentField = hit.field(ParentFieldMapper.NAME);
-                    if (parentField == null) {
-                        throw new IllegalStateException("All children must have a _parent");
-                    }
-                    Term uidTerm = context.mapperService().createUidTerm(parentType, parentField.getValue());
-                    if (uidTerm == null) {
-                        hitQuery = new MatchNoDocsQuery("Missing type: " + parentType);
-                    } else {
-                        hitQuery = new TermQuery(uidTerm);
-                    }
-                } else {
-                    result[i] = Lucene.EMPTY_TOP_DOCS;
-                    continue;
-                }
-
-                BooleanQuery q = new BooleanQuery.Builder()
-                    // Only include docs that have the current hit as parent
-                    .add(hitQuery, Occur.FILTER)
-                    // Only include docs that have this inner hits type
-                    .add(documentMapper.typeFilter(context.getQueryShardContext()), Occur.FILTER)
-                    .build();
-                Weight weight = context.searcher().createNormalizedWeight(q, false);
-                if (size() == 0) {
-                    TotalHitCountCollector totalHitCountCollector = new TotalHitCountCollector();
-                    for (LeafReaderContext ctx : context.searcher().getIndexReader().leaves()) {
-                        intersect(weight, innerHitQueryWeight, totalHitCountCollector, ctx);
-                    }
-                    result[i] = new TopDocs(totalHitCountCollector.getTotalHits(), Lucene.EMPTY_SCORE_DOCS, 0);
-                } else {
-                    int topN = Math.min(from() + size(), context.searcher().getIndexReader().maxDoc());
-                    TopDocsCollector topDocsCollector;
-                    if (sort() != null) {
-                        topDocsCollector = TopFieldCollector.create(sort().sort, topN, true, trackScores(), trackScores());
-                    } else {
-                        topDocsCollector = TopScoreDocCollector.create(topN);
-                    }
-                    try {
-                        for (LeafReaderContext ctx : context.searcher().getIndexReader().leaves()) {
-                            intersect(weight, innerHitQueryWeight, topDocsCollector, ctx);
-                        }
-                    } finally {
-                        clearReleasables(Lifetime.COLLECTION);
-                    }
-                    result[i] = topDocsCollector.topDocs(from(), size());
-                }
-            }
-            return result;
-        }
-
-        private boolean isParentHit(SearchHit hit) {
-            return hit.getType().equals(documentMapper.parentFieldMapper().type());
-        }
-
-        private boolean isChildHit(SearchHit hit) {
-            DocumentMapper hitDocumentMapper = mapperService.documentMapper(hit.getType());
-            return documentMapper.type().equals(hitDocumentMapper.parentFieldMapper().type());
-        }
     }
 
-    static void intersect(Weight weight, Weight innerHitQueryWeight, Collector collector, LeafReaderContext ctx) throws IOException {
+    public static void intersect(Weight weight, Weight innerHitQueryWeight, Collector collector, LeafReaderContext ctx) throws IOException {
         ScorerSupplier scorerSupplier = weight.scorerSupplier(ctx);
         if (scorerSupplier == null) {
             return;
diff --git a/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsFetchSubPhase.java b/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsFetchSubPhase.java
index 90e1b5cf82..d1fd74eaa7 100644
--- a/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsFetchSubPhase.java
+++ b/core/src/main/java/org/elasticsearch/search/fetch/subphase/InnerHitsFetchSubPhase.java
@@ -47,8 +47,8 @@ public final class InnerHitsFetchSubPhase implements FetchSubPhase {
             return;
         }
 
-        for (Map.Entry<String, InnerHitsContext.BaseInnerHits> entry : context.innerHits().getInnerHits().entrySet()) {
-            InnerHitsContext.BaseInnerHits innerHits = entry.getValue();
+        for (Map.Entry<String, InnerHitsContext.InnerHitSubContext> entry : context.innerHits().getInnerHits().entrySet()) {
+            InnerHitsContext.InnerHitSubContext innerHits = entry.getValue();
             TopDocs[] topDocs = innerHits.topDocs(hits);
             for (int i = 0; i < hits.length; i++) {
                 SearchHit hit = hits[i];