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

25 files changed, 795 insertions, 690 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];
diff --git a/core/src/test/java/org/elasticsearch/index/query/InnerHitBuilderTests.java b/core/src/test/java/org/elasticsearch/index/query/InnerHitBuilderTests.java
index 9e199f71ce..20b0fc1e16 100644
--- a/core/src/test/java/org/elasticsearch/index/query/InnerHitBuilderTests.java
+++ b/core/src/test/java/org/elasticsearch/index/query/InnerHitBuilderTests.java
@@ -18,7 +18,6 @@
  */
 package org.elasticsearch.index.query;
 
-import org.apache.lucene.search.join.ScoreMode;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@@ -29,15 +28,12 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
-import org.elasticsearch.index.query.functionscore.FunctionScoreQueryBuilder;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptType;
 import org.elasticsearch.search.SearchModule;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
-import org.elasticsearch.search.fetch.subphase.InnerHitsContext;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilderTests;
-import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.search.internal.ShardSearchLocalRequest;
 import org.elasticsearch.search.sort.SortBuilder;
 import org.elasticsearch.search.sort.SortBuilders;
@@ -59,10 +55,7 @@ import static java.util.Collections.emptyList;
 import static org.elasticsearch.test.EqualsHashCodeTestUtils.checkEqualsAndHashCode;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
-import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.sameInstance;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 public class InnerHitBuilderTests extends ESTestCase {
 
@@ -124,7 +117,7 @@ public class InnerHitBuilderTests extends ESTestCase {
 
     public void testFromAndToXContent() throws Exception {
         for (int runs = 0; runs < NUMBER_OF_TESTBUILDERS; runs++) {
-            InnerHitBuilder innerHit = randomInnerHits(true, false);
+            InnerHitBuilder innerHit = randomInnerHits();
             XContentBuilder builder = XContentFactory.contentBuilder(randomFrom(XContentType.values()));
             innerHit.toXContent(builder, ToXContent.EMPTY_PARAMS);
             //fields is printed out as an object but parsed into a List where order matters, we disable shuffling
@@ -144,82 +137,7 @@ public class InnerHitBuilderTests extends ESTestCase {
         }
     }
 
-    public void testInlineLeafInnerHitsNestedQuery() throws Exception {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
-        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
-        nestedQueryBuilder.innerHit(leafInnerHits, false);
-        Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-        nestedQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
-        assertThat(innerHitBuilders.get(leafInnerHits.getName()), notNullValue());
-    }
-
-    public void testInlineLeafInnerHitsNestedQueryViaBoolQuery() {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
-        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
-                .innerHit(leafInnerHits, false);
-        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder().should(nestedQueryBuilder);
-        Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-        boolQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
-        assertThat(innerHitBuilders.get(leafInnerHits.getName()), notNullValue());
-    }
-
-    public void testInlineLeafInnerHitsNestedQueryViaConstantScoreQuery() {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
-        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
-                .innerHit(leafInnerHits, false);
-        ConstantScoreQueryBuilder constantScoreQueryBuilder = new ConstantScoreQueryBuilder(nestedQueryBuilder);
-        Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-        constantScoreQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
-        assertThat(innerHitBuilders.get(leafInnerHits.getName()), notNullValue());
-    }
-
-    public void testInlineLeafInnerHitsNestedQueryViaBoostingQuery() {
-        InnerHitBuilder leafInnerHits1 = randomInnerHits();
-        NestedQueryBuilder nestedQueryBuilder1 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
-                .innerHit(leafInnerHits1, false);
-        InnerHitBuilder leafInnerHits2 = randomInnerHits();
-        NestedQueryBuilder nestedQueryBuilder2 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
-                .innerHit(leafInnerHits2, false);
-        BoostingQueryBuilder constantScoreQueryBuilder = new BoostingQueryBuilder(nestedQueryBuilder1, nestedQueryBuilder2);
-        Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-        constantScoreQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
-        assertThat(innerHitBuilders.get(leafInnerHits1.getName()), notNullValue());
-        assertThat(innerHitBuilders.get(leafInnerHits2.getName()), notNullValue());
-    }
-
-    public void testInlineLeafInnerHitsNestedQueryViaFunctionScoreQuery() {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
-        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
-                .innerHit(leafInnerHits, false);
-        FunctionScoreQueryBuilder functionScoreQueryBuilder = new FunctionScoreQueryBuilder(nestedQueryBuilder);
-        Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-        ((AbstractQueryBuilder<?>) functionScoreQueryBuilder).extractInnerHitBuilders(innerHitBuilders);
-        assertThat(innerHitBuilders.get(leafInnerHits.getName()), notNullValue());
-    }
-
-    public void testBuildIgnoreUnmappedNestQuery() throws Exception {
-        QueryShardContext queryShardContext = mock(QueryShardContext.class);
-        when(queryShardContext.getObjectMapper("path")).thenReturn(null);
-        SearchContext searchContext = mock(SearchContext.class);
-        when(searchContext.getQueryShardContext()).thenReturn(queryShardContext);
-
-        InnerHitBuilder leafInnerHits = randomInnerHits();
-        NestedQueryBuilder query1 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
-        query1.innerHit(leafInnerHits, false);
-        expectThrows(IllegalStateException.class, () -> query1.innerHit().build(searchContext, new InnerHitsContext()));
-
-        NestedQueryBuilder query2 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
-        query2.innerHit(leafInnerHits, true);
-        InnerHitsContext innerHitsContext = new InnerHitsContext();
-        query2.innerHit().build(searchContext, innerHitsContext);
-        assertThat(innerHitsContext.getInnerHits().size(), equalTo(0));
-    }
-
     public static InnerHitBuilder randomInnerHits() {
-        return randomInnerHits(true, true);
-    }
-
-    public static InnerHitBuilder randomInnerHits(boolean recursive, boolean includeQueryTypeOrPath) {
         InnerHitBuilder innerHits = new InnerHitBuilder();
         innerHits.setName(randomAlphaOfLengthBetween(1, 16));
         innerHits.setFrom(randomIntBetween(0, 128));
@@ -256,33 +174,7 @@ public class InnerHitBuilderTests extends ESTestCase {
             );
         }
         innerHits.setHighlightBuilder(HighlightBuilderTests.randomHighlighterBuilder());
-        if (recursive && randomBoolean()) {
-            int size = randomIntBetween(1, 16);
-            for (int i = 0; i < size; i++) {
-                innerHits.addChildInnerHit(randomInnerHits(false, includeQueryTypeOrPath));
-            }
-        }
-
-        if (includeQueryTypeOrPath) {
-            QueryBuilder query = new MatchQueryBuilder(randomAlphaOfLengthBetween(1, 16), randomAlphaOfLengthBetween(1, 16));
-            if (randomBoolean()) {
-                return new InnerHitBuilder(innerHits, randomAlphaOfLength(8), query, randomBoolean());
-            } else {
-                return new InnerHitBuilder(innerHits, query, randomAlphaOfLength(8), randomBoolean());
-            }
-        } else {
-            return innerHits;
-        }
-    }
-
-    public void testCopyConstructor() throws Exception {
-        InnerHitBuilder original = randomInnerHits();
-        InnerHitBuilder copy = original.getNestedPath() != null ?
-                new InnerHitBuilder(original, original.getNestedPath(), original.getQuery(), original.isIgnoreUnmapped()) :
-                new InnerHitBuilder(original, original.getQuery(), original.getParentChildType(), original.isIgnoreUnmapped());
-        assertThat(copy, equalTo(original));
-        copy = mutate(copy);
-        assertThat(copy, not(equalTo(original)));
+        return innerHits;
     }
 
     static InnerHitBuilder mutate(InnerHitBuilder original) throws IOException {
diff --git a/core/src/test/java/org/elasticsearch/index/query/NestedQueryBuilderTests.java b/core/src/test/java/org/elasticsearch/index/query/NestedQueryBuilderTests.java
index ed4fbcd53c..a9e1b6ce23 100644
--- a/core/src/test/java/org/elasticsearch/index/query/NestedQueryBuilderTests.java
+++ b/core/src/test/java/org/elasticsearch/index/query/NestedQueryBuilderTests.java
@@ -23,24 +23,31 @@ import com.carrotsearch.randomizedtesting.generators.RandomPicks;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.join.ScoreMode;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.query.functionscore.FunctionScoreQueryBuilder;
 import org.elasticsearch.index.search.ESToParentBlockJoinQuery;
 import org.elasticsearch.search.fetch.subphase.InnerHitsContext;
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.AbstractQueryTestCase;
+import org.elasticsearch.test.VersionUtils;
+import org.hamcrest.Matchers;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.elasticsearch.index.query.InnerHitBuilderTests.randomInnerHits;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBuilder> {
 
@@ -74,19 +81,17 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
                 RandomPicks.randomFrom(random(), ScoreMode.values()));
         nqb.ignoreUnmapped(randomBoolean());
         if (randomBoolean()) {
-            nqb.innerHit(new InnerHitBuilder()
-                    .setName(randomAlphaOfLengthBetween(1, 10))
+            nqb.innerHit(new InnerHitBuilder(randomAlphaOfLengthBetween(1, 10))
                     .setSize(randomIntBetween(0, 100))
-                    .addSort(new FieldSortBuilder(INT_FIELD_NAME).order(SortOrder.ASC)), nqb.ignoreUnmapped());
+                    .addSort(new FieldSortBuilder(INT_FIELD_NAME).order(SortOrder.ASC))
+                    .setIgnoreUnmapped(nqb.ignoreUnmapped()));
         }
         return nqb;
     }
 
     @Override
     protected void doAssertLuceneQuery(NestedQueryBuilder queryBuilder, Query query, SearchContext searchContext) throws IOException {
-        QueryBuilder innerQueryBuilder = queryBuilder.query();
         assertThat(query, instanceOf(ESToParentBlockJoinQuery.class));
-        ESToParentBlockJoinQuery parentBlockJoinQuery = (ESToParentBlockJoinQuery) query;
         // TODO how to assert this?
         if (queryBuilder.innerHit() != null) {
             // have to rewrite again because the provided queryBuilder hasn't been rewritten (directly returned from
@@ -94,21 +99,35 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
             queryBuilder = (NestedQueryBuilder) queryBuilder.rewrite(searchContext.getQueryShardContext());
 
             assertNotNull(searchContext);
-            Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-            InnerHitBuilder.extractInnerHits(queryBuilder, innerHitBuilders);
-            for (InnerHitBuilder builder : innerHitBuilders.values()) {
+            Map<String, InnerHitContextBuilder> innerHitInternals = new HashMap<>();
+            InnerHitContextBuilder.extractInnerHits(queryBuilder, innerHitInternals);
+            for (InnerHitContextBuilder builder : innerHitInternals.values()) {
                 builder.build(searchContext, searchContext.innerHits());
             }
             assertNotNull(searchContext.innerHits());
             assertEquals(1, searchContext.innerHits().getInnerHits().size());
             assertTrue(searchContext.innerHits().getInnerHits().containsKey(queryBuilder.innerHit().getName()));
-            InnerHitsContext.BaseInnerHits innerHits = searchContext.innerHits().getInnerHits().get(queryBuilder.innerHit().getName());
+            InnerHitsContext.InnerHitSubContext innerHits = searchContext.innerHits().getInnerHits().get(queryBuilder.innerHit().getName());
             assertEquals(innerHits.size(), queryBuilder.innerHit().getSize());
             assertEquals(innerHits.sort().sort.getSort().length, 1);
             assertEquals(innerHits.sort().sort.getSort()[0].getField(), INT_FIELD_NAME);
         }
     }
 
+    /**
+     * Test (de)serialization on all previous released versions
+     */
+    public void testSerializationBWC() throws IOException {
+        for (Version version : VersionUtils.allReleasedVersions()) {
+            NestedQueryBuilder testQuery = createTestQueryBuilder();
+            if (version.before(Version.V_5_2_0) && testQuery.innerHit() != null) {
+                // ignore unmapped for inner_hits has been added on 5.2
+                testQuery.innerHit().setIgnoreUnmapped(false);
+            }
+            assertSerialization(testQuery, version);
+        }
+    }
+
     public void testValidate() {
         QueryBuilder innerQuery = RandomQueryBuilder.createQuery(random());
         IllegalArgumentException e =
@@ -245,4 +264,87 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
             () -> NestedQueryBuilder.parseScoreMode("unrecognized value"));
         assertEquals("No score mode for child query [unrecognized value] found", e.getMessage());
     }
+
+    public void testInlineLeafInnerHitsNestedQuery() throws Exception {
+        InnerHitBuilder leafInnerHits = randomInnerHits();
+        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
+        nestedQueryBuilder.innerHit(leafInnerHits);
+        Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+        nestedQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
+        assertThat(innerHitBuilders.get(leafInnerHits.getName()), Matchers.notNullValue());
+    }
+
+    public void testInlineLeafInnerHitsNestedQueryViaBoolQuery() {
+        InnerHitBuilder leafInnerHits = randomInnerHits();
+        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
+            .innerHit(leafInnerHits);
+        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder().should(nestedQueryBuilder);
+        Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+        boolQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
+        assertThat(innerHitBuilders.get(leafInnerHits.getName()), Matchers.notNullValue());
+    }
+
+    public void testInlineLeafInnerHitsNestedQueryViaConstantScoreQuery() {
+        InnerHitBuilder leafInnerHits = randomInnerHits();
+        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
+            .innerHit(leafInnerHits);
+        ConstantScoreQueryBuilder constantScoreQueryBuilder = new ConstantScoreQueryBuilder(nestedQueryBuilder);
+        Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+        constantScoreQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
+        assertThat(innerHitBuilders.get(leafInnerHits.getName()), Matchers.notNullValue());
+    }
+
+    public void testInlineLeafInnerHitsNestedQueryViaBoostingQuery() {
+        InnerHitBuilder leafInnerHits1 = randomInnerHits();
+        NestedQueryBuilder nestedQueryBuilder1 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
+            .innerHit(leafInnerHits1);
+        InnerHitBuilder leafInnerHits2 = randomInnerHits();
+        NestedQueryBuilder nestedQueryBuilder2 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
+            .innerHit(leafInnerHits2);
+        BoostingQueryBuilder constantScoreQueryBuilder = new BoostingQueryBuilder(nestedQueryBuilder1, nestedQueryBuilder2);
+        Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+        constantScoreQueryBuilder.extractInnerHitBuilders(innerHitBuilders);
+        assertThat(innerHitBuilders.get(leafInnerHits1.getName()), Matchers.notNullValue());
+        assertThat(innerHitBuilders.get(leafInnerHits2.getName()), Matchers.notNullValue());
+    }
+
+    public void testInlineLeafInnerHitsNestedQueryViaFunctionScoreQuery() {
+        InnerHitBuilder leafInnerHits = randomInnerHits();
+        NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
+            .innerHit(leafInnerHits);
+        FunctionScoreQueryBuilder functionScoreQueryBuilder = new FunctionScoreQueryBuilder(nestedQueryBuilder);
+        Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+        ((AbstractQueryBuilder<?>) functionScoreQueryBuilder).extractInnerHitBuilders(innerHitBuilders);
+        assertThat(innerHitBuilders.get(leafInnerHits.getName()), Matchers.notNullValue());
+    }
+
+    public void testBuildIgnoreUnmappedNestQuery() throws Exception {
+        QueryShardContext queryShardContext = mock(QueryShardContext.class);
+        when(queryShardContext.getObjectMapper("path")).thenReturn(null);
+        SearchContext searchContext = mock(SearchContext.class);
+        when(searchContext.getQueryShardContext()).thenReturn(queryShardContext);
+
+        InnerHitBuilder leafInnerHits = randomInnerHits();
+        NestedQueryBuilder query1 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
+        query1.innerHit(leafInnerHits);
+        final Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+        final InnerHitsContext innerHitsContext = new InnerHitsContext();
+        expectThrows(IllegalStateException.class, () -> {
+            query1.extractInnerHitBuilders(innerHitBuilders);
+            assertThat(innerHitBuilders.size(), Matchers.equalTo(1));
+            assertTrue(innerHitBuilders.containsKey(leafInnerHits.getName()));
+            innerHitBuilders.get(leafInnerHits.getName()).build(searchContext, innerHitsContext);
+        });
+        innerHitBuilders.clear();
+        NestedQueryBuilder query2 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
+        query2.innerHit(leafInnerHits.setIgnoreUnmapped(true));
+        query2.extractInnerHitBuilders(innerHitBuilders);
+        assertThat(innerHitBuilders.size(), Matchers.equalTo(1));
+        assertTrue(innerHitBuilders.containsKey(leafInnerHits.getName()));
+        assertThat(innerHitBuilders.get(leafInnerHits.getName()), instanceOf(NestedQueryBuilder.NestedInnerHitContextBuilder.class));
+        NestedQueryBuilder.NestedInnerHitContextBuilder nestedContextBuilder =
+            (NestedQueryBuilder.NestedInnerHitContextBuilder) innerHitBuilders.get(leafInnerHits.getName());
+        nestedContextBuilder.build(searchContext, innerHitsContext);
+        assertThat(innerHitsContext.getInnerHits().size(), Matchers.equalTo(0));
+    }
 }
diff --git a/core/src/test/java/org/elasticsearch/search/collapse/CollapseBuilderTests.java b/core/src/test/java/org/elasticsearch/search/collapse/CollapseBuilderTests.java
index e8d8af2502..a0533a396a 100644
--- a/core/src/test/java/org/elasticsearch/search/collapse/CollapseBuilderTests.java
+++ b/core/src/test/java/org/elasticsearch/search/collapse/CollapseBuilderTests.java
@@ -70,7 +70,7 @@ public class CollapseBuilderTests extends AbstractWireSerializingTestCase {
         CollapseBuilder builder = new CollapseBuilder(randomAlphaOfLength(10));
         builder.setMaxConcurrentGroupRequests(randomIntBetween(1, 48));
         if (randomBoolean()) {
-            InnerHitBuilder innerHit = InnerHitBuilderTests.randomInnerHits(false, false);
+            InnerHitBuilder innerHit = InnerHitBuilderTests.randomInnerHits();
             builder.setInnerHits(innerHit);
         }
         return builder;
diff --git a/core/src/test/java/org/elasticsearch/search/fetch/subphase/InnerHitsIT.java b/core/src/test/java/org/elasticsearch/search/fetch/subphase/InnerHitsIT.java
index 8eca505564..617aa28a0b 100644
--- a/core/src/test/java/org/elasticsearch/search/fetch/subphase/InnerHitsIT.java
+++ b/core/src/test/java/org/elasticsearch/search/fetch/subphase/InnerHitsIT.java
@@ -123,7 +123,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments", matchQuery("comments.message", "fox"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder().setName("comment"), false)
+                        .innerHit(new InnerHitBuilder("comment"))
                 ).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -141,7 +141,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments", matchQuery("comments.message", "elephant"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder().setName("comment"), false)
+                        .innerHit(new InnerHitBuilder("comment"))
                 ).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -168,8 +168,7 @@ public class InnerHitsIT extends ESIntegTestCase {
                                 .addDocValueField("comments.message")
                                 .addScriptField("script",
                                         new Script(ScriptType.INLINE, MockScriptEngine.NAME, "5", Collections.emptyMap()))
-                                .setSize(1),
-                        false)).get();
+                                .setSize(1))).get();
         assertNoFailures(response);
         innerHits = response.getHits().getAt(0).getInnerHits().get("comments");
         assertThat(innerHits.getTotalHits(), equalTo(2L));
@@ -208,10 +207,10 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         int size = randomIntBetween(0, numDocs);
         BoolQueryBuilder boolQuery = new BoolQueryBuilder();
-        boolQuery.should(nestedQuery("field1", matchAllQuery(), ScoreMode.Avg).innerHit(new InnerHitBuilder().setName("a").setSize(size)
-                .addSort(new FieldSortBuilder("_doc").order(SortOrder.DESC)), false));
-        boolQuery.should(nestedQuery("field2", matchAllQuery(), ScoreMode.Avg).innerHit(new InnerHitBuilder().setName("b")
-                .addSort(new FieldSortBuilder("_doc").order(SortOrder.DESC)).setSize(size), false));
+        boolQuery.should(nestedQuery("field1", matchAllQuery(), ScoreMode.Avg).innerHit(new InnerHitBuilder("a").setSize(size)
+                .addSort(new FieldSortBuilder("_doc").order(SortOrder.DESC))));
+        boolQuery.should(nestedQuery("field2", matchAllQuery(), ScoreMode.Avg).innerHit(new InnerHitBuilder("b")
+                .addSort(new FieldSortBuilder("_doc").order(SortOrder.DESC)).setSize(size)));
         SearchResponse searchResponse = client().prepareSearch("idx")
                 .setQuery(boolQuery)
                 .setSize(numDocs)
@@ -291,8 +290,8 @@ public class InnerHitsIT extends ESIntegTestCase {
                 .setQuery(
                         nestedQuery("comments",
                                 nestedQuery("comments.remarks", matchQuery("comments.remarks.message", "good"), ScoreMode.Avg)
-                                        .innerHit(new InnerHitBuilder().setName("remark"), false),
-                                ScoreMode.Avg).innerHit(new InnerHitBuilder(), false)
+                                        .innerHit(new InnerHitBuilder("remark")),
+                                ScoreMode.Avg).innerHit(new InnerHitBuilder())
                 ).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -316,7 +315,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         // Directly refer to the second level:
         response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments.remarks", matchQuery("comments.remarks.message", "bad"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder(), false)).get();
+                        .innerHit(new InnerHitBuilder())).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         assertSearchHit(response, 1, hasId("2"));
@@ -334,8 +333,8 @@ public class InnerHitsIT extends ESIntegTestCase {
                 .setQuery(
                         nestedQuery("comments",
                                 nestedQuery("comments.remarks", matchQuery("comments.remarks.message", "bad"), ScoreMode.Avg)
-                                        .innerHit(new InnerHitBuilder().setName("remark"), false),
-                                ScoreMode.Avg).innerHit(new InnerHitBuilder(), false)
+                                        .innerHit(new InnerHitBuilder("remark")),
+                                ScoreMode.Avg).innerHit(new InnerHitBuilder())
                 ).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -370,7 +369,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments", matchQuery("comments.message", "fox"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder(), false))
+                        .innerHit(new InnerHitBuilder()))
                 .get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -412,7 +411,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments.messages", matchQuery("comments.messages.message", "fox"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder(), false)).get();
+                        .innerHit(new InnerHitBuilder())).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         SearchHit hit = response.getHits().getAt(0);
@@ -426,7 +425,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         response = client().prepareSearch("articles")
                 .setQuery(nestedQuery("comments.messages", matchQuery("comments.messages.message", "bear"), ScoreMode.Avg)
-                        .innerHit(new InnerHitBuilder(), false)).get();
+                        .innerHit(new InnerHitBuilder())).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         hit = response.getHits().getAt(0);
@@ -447,7 +446,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(), false)).get();
+                        .innerHit(new InnerHitBuilder())).get();
         assertNoFailures(response);
         assertHitCount(response, 1);
         hit = response.getHits().getAt(0);;
@@ -534,7 +533,7 @@ public class InnerHitsIT extends ESIntegTestCase {
                 .should(termQuery("nested1.n_field1", "n_value1_3").queryName("test2"))
                 .should(termQuery("nested1.n_field2", "n_value2_2").queryName("test3"));
         query = nestedQuery("nested1", query, ScoreMode.Avg).innerHit(
-                new InnerHitBuilder().addSort(new FieldSortBuilder("nested1.n_field1").order(SortOrder.ASC)), false);
+                new InnerHitBuilder().addSort(new FieldSortBuilder("nested1.n_field1").order(SortOrder.ASC)));
         SearchResponse searchResponse = client().prepareSearch("test")
                 .setQuery(query)
                 .setSize(numDocs)
@@ -581,7 +580,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         SearchResponse response = client().prepareSearch()
                 .setQuery(nestedQuery("comments", matchQuery("comments.message", "fox"), ScoreMode.None)
                 .innerHit(new InnerHitBuilder().setFetchSourceContext(new FetchSourceContext(true,
-                    new String[]{"comments.message"}, null)), false))
+                    new String[]{"comments.message"}, null))))
                 .get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -608,7 +607,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         SearchResponse response = client().prepareSearch("index1", "index2")
             .setQuery(boolQuery()
                 .should(nestedQuery("nested_type", matchAllQuery(), ScoreMode.None).ignoreUnmapped(true)
-                        .innerHit(new InnerHitBuilder(), true))
+                        .innerHit(new InnerHitBuilder().setIgnoreUnmapped(true)))
                 .should(termQuery("key", "value"))
             )
             .get();
diff --git a/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasChildQueryBuilder.java b/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasChildQueryBuilder.java
index 95d000e3cc..566edf1cb8 100644
--- a/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasChildQueryBuilder.java
+++ b/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasChildQueryBuilder.java
@@ -27,6 +27,7 @@ import org.apache.lucene.search.Query;
 import org.apache.lucene.search.join.JoinUtil;
 import org.apache.lucene.search.join.ScoreMode;
 import org.apache.lucene.search.similarities.Similarity;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -41,6 +42,7 @@ import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.ParentFieldMapper;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
 import org.elasticsearch.index.query.InnerHitBuilder;
+import org.elasticsearch.index.query.InnerHitContextBuilder;
 import org.elasticsearch.index.query.NestedQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryParseContext;
@@ -49,6 +51,7 @@ import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.index.query.QueryShardException;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 
@@ -123,7 +126,15 @@ public class HasChildQueryBuilder extends AbstractQueryBuilder<HasChildQueryBuil
         out.writeInt(maxChildren);
         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.writeToParentChildBWC(out, query, type);
+            }
+        } else {
+            out.writeOptionalWriteable(innerHitBuilder);
+        }
         out.writeBoolean(ignoreUnmapped);
     }
 
@@ -153,8 +164,8 @@ public class HasChildQueryBuilder extends AbstractQueryBuilder<HasChildQueryBuil
         return innerHitBuilder;
     }
 
-    public HasChildQueryBuilder innerHit(InnerHitBuilder innerHit, boolean ignoreUnmapped) {
-        this.innerHitBuilder = new InnerHitBuilder(Objects.requireNonNull(innerHit), query, type, ignoreUnmapped);
+    public HasChildQueryBuilder innerHit(InnerHitBuilder innerHit) {
+        this.innerHitBuilder = innerHit;
         return this;
     }
 
@@ -281,7 +292,8 @@ public class HasChildQueryBuilder extends AbstractQueryBuilder<HasChildQueryBuil
         hasChildQueryBuilder.boost(boost);
         hasChildQueryBuilder.ignoreUnmapped(ignoreUnmapped);
         if (innerHitBuilder != null) {
-            hasChildQueryBuilder.innerHit(innerHitBuilder, ignoreUnmapped);
+            hasChildQueryBuilder.innerHit(innerHitBuilder);
+            hasChildQueryBuilder.ignoreUnmapped(ignoreUnmapped);
         }
         return hasChildQueryBuilder;
     }
@@ -454,12 +466,11 @@ public class HasChildQueryBuilder extends AbstractQueryBuilder<HasChildQueryBuil
     }
 
     @Override
-    protected QueryBuilder doRewrite(QueryRewriteContext queryRewriteContext) throws IOException {
-        QueryBuilder rewrittenQuery = query.rewrite(queryRewriteContext);
+    protected QueryBuilder doRewrite(QueryRewriteContext queryShardContext) throws IOException {
+        QueryBuilder rewrittenQuery = query.rewrite(queryShardContext);
         if (rewrittenQuery != query) {
-            InnerHitBuilder rewrittenInnerHit = InnerHitBuilder.rewrite(innerHitBuilder, rewrittenQuery);
             HasChildQueryBuilder hasChildQueryBuilder =
-                new HasChildQueryBuilder(type, rewrittenQuery, minChildren, maxChildren, scoreMode, rewrittenInnerHit);
+                new HasChildQueryBuilder(type, rewrittenQuery, minChildren, maxChildren, scoreMode, innerHitBuilder);
             hasChildQueryBuilder.ignoreUnmapped(ignoreUnmapped);
             return hasChildQueryBuilder;
         }
@@ -467,9 +478,14 @@ public class HasChildQueryBuilder extends AbstractQueryBuilder<HasChildQueryBuil
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
         if (innerHitBuilder != null) {
-            innerHitBuilder.inlineInnerHits(innerHits);
+            Map<String, InnerHitContextBuilder> children = new HashMap<>();
+            InnerHitContextBuilder.extractInnerHits(query, children);
+            String name = innerHitBuilder.getName() != null ? innerHitBuilder.getName() : type;
+            InnerHitContextBuilder innerHitContextBuilder =
+                new HasParentQueryBuilder.ParentChildInnerHitContextBuilder(type, query, innerHitBuilder, children);
+            innerHits.put(name, innerHitContextBuilder);
         }
     }
 }
diff --git a/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasParentQueryBuilder.java b/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasParentQueryBuilder.java
index b216e886a5..30918233fa 100644
--- a/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasParentQueryBuilder.java
+++ b/modules/parent-join/src/main/java/org/elasticsearch/join/query/HasParentQueryBuilder.java
@@ -18,36 +18,57 @@
  */
 package org.elasticsearch.join.query;
 
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.Term;
 import org.apache.lucene.search.BooleanClause;
 import org.apache.lucene.search.BooleanQuery;
+import org.apache.lucene.search.DocValuesTermsQuery;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
+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.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.fielddata.plain.SortedSetDVOrdinalsIndexFieldData;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ParentFieldMapper;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
 import org.elasticsearch.index.query.InnerHitBuilder;
+import org.elasticsearch.index.query.InnerHitContextBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.index.query.QueryRewriteContext;
 import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.index.query.QueryShardException;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHitField;
+import org.elasticsearch.search.fetch.subphase.InnerHitsContext;
+import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
+import static org.elasticsearch.search.fetch.subphase.InnerHitsContext.intersect;
+
 /**
  * Builder for the 'has_parent' query.
  */
@@ -69,18 +90,18 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
     private final QueryBuilder query;
     private final String type;
     private final boolean score;
-    private InnerHitBuilder innerHit;
+    private InnerHitBuilder innerHitBuilder;
     private boolean ignoreUnmapped = false;
 
     public HasParentQueryBuilder(String type, QueryBuilder query, boolean score) {
         this(type, query, score, null);
     }
 
-    private HasParentQueryBuilder(String type, QueryBuilder query, boolean score, InnerHitBuilder innerHit) {
+    private HasParentQueryBuilder(String type, QueryBuilder query, boolean score, InnerHitBuilder innerHitBuilder) {
         this.type = requireValue(type, "[" + NAME + "] requires 'type' field");
         this.query = requireValue(query, "[" + NAME + "] requires 'query' field");
         this.score = score;
-        this.innerHit = innerHit;
+        this.innerHitBuilder = innerHitBuilder;
     }
 
     /**
@@ -91,7 +112,7 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
         type = in.readString();
         score = in.readBoolean();
         query = in.readNamedWriteable(QueryBuilder.class);
-        innerHit = in.readOptionalWriteable(InnerHitBuilder::new);
+        innerHitBuilder = in.readOptionalWriteable(InnerHitBuilder::new);
         ignoreUnmapped = in.readBoolean();
     }
 
@@ -100,7 +121,15 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
         out.writeString(type);
         out.writeBoolean(score);
         out.writeNamedWriteable(query);
-        out.writeOptionalWriteable(innerHit);
+        if (out.getVersion().before(Version.V_6_0_0_alpha2_UNRELEASED)) {
+            final boolean hasInnerHit = innerHitBuilder != null;
+            out.writeBoolean(hasInnerHit);
+            if (hasInnerHit) {
+                innerHitBuilder.writeToParentChildBWC(out, query, type);
+            }
+        } else {
+            out.writeOptionalWriteable(innerHitBuilder);
+        }
         out.writeBoolean(ignoreUnmapped);
     }
 
@@ -129,11 +158,11 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
      *  Returns inner hit definition in the scope of this query and reusing the defined type and query.
      */
     public InnerHitBuilder innerHit() {
-        return innerHit;
+        return innerHitBuilder;
     }
 
-    public HasParentQueryBuilder innerHit(InnerHitBuilder innerHit, boolean ignoreUnmapped) {
-        this.innerHit = new InnerHitBuilder(innerHit, query, type, ignoreUnmapped);
+    public HasParentQueryBuilder innerHit(InnerHitBuilder innerHit) {
+        this.innerHitBuilder = innerHit;
         return this;
     }
 
@@ -172,7 +201,8 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
             if (ignoreUnmapped) {
                 return new MatchNoDocsQuery();
             } else {
-                throw new QueryShardException(context, "[" + NAME + "] query configured 'parent_type' [" + type + "] is not a valid type");
+                throw new QueryShardException(context,
+                    "[" + NAME + "] query configured 'parent_type' [" + type + "] is not a valid type");
             }
         }
 
@@ -224,8 +254,8 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
         builder.field(SCORE_FIELD.getPreferredName(), score);
         builder.field(IGNORE_UNMAPPED_FIELD.getPreferredName(), ignoreUnmapped);
         printBoostAndQueryName(builder);
-        if (innerHit != null) {
-            builder.field(INNER_HITS_FIELD.getPreferredName(), innerHit, params);
+        if (innerHitBuilder != null) {
+            builder.field(INNER_HITS_FIELD.getPreferredName(), innerHitBuilder, params);
         }
         builder.endObject();
     }
@@ -251,7 +281,8 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
                 } else if (INNER_HITS_FIELD.match(currentFieldName)) {
                     innerHits = InnerHitBuilder.fromXContent(parseContext);
                 } else {
-                    throw new ParsingException(parser.getTokenLocation(), "[has_parent] query does not support [" + currentFieldName + "]");
+                    throw new ParsingException(parser.getTokenLocation(),
+                        "[has_parent] query does not support [" + currentFieldName + "]");
                 }
             } else if (token.isValue()) {
                 if (TYPE_FIELD.match(currentFieldName)) {
@@ -275,7 +306,8 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
                 } else if (AbstractQueryBuilder.NAME_FIELD.match(currentFieldName)) {
                     queryName = parser.text();
                 } else {
-                    throw new ParsingException(parser.getTokenLocation(), "[has_parent] query does not support [" + currentFieldName + "]");
+                    throw new ParsingException(parser.getTokenLocation(),
+                        "[has_parent] query does not support [" + currentFieldName + "]");
                 }
             }
         }
@@ -284,7 +316,7 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
                 .queryName(queryName)
                 .boost(boost);
         if (innerHits != null) {
-            queryBuilder.innerHit(innerHits, ignoreUnmapped);
+            queryBuilder.innerHit(innerHits);
         }
         return queryBuilder;
     }
@@ -299,21 +331,20 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
         return Objects.equals(query, that.query)
                 && Objects.equals(type, that.type)
                 && Objects.equals(score, that.score)
-                && Objects.equals(innerHit, that.innerHit)
+                && Objects.equals(innerHitBuilder, that.innerHitBuilder)
                 && Objects.equals(ignoreUnmapped, that.ignoreUnmapped);
     }
 
     @Override
     protected int doHashCode() {
-        return Objects.hash(query, type, score, innerHit, ignoreUnmapped);
+        return Objects.hash(query, type, score, innerHitBuilder, ignoreUnmapped);
     }
 
     @Override
     protected QueryBuilder doRewrite(QueryRewriteContext queryShardContext) throws IOException {
         QueryBuilder rewrittenQuery = query.rewrite(queryShardContext);
         if (rewrittenQuery != query) {
-            InnerHitBuilder rewrittenInnerHit = InnerHitBuilder.rewrite(innerHit, rewrittenQuery);
-            HasParentQueryBuilder hasParentQueryBuilder = new HasParentQueryBuilder(type, rewrittenQuery, score, rewrittenInnerHit);
+            HasParentQueryBuilder hasParentQueryBuilder = new HasParentQueryBuilder(type, rewrittenQuery, score, innerHitBuilder);
             hasParentQueryBuilder.ignoreUnmapped(ignoreUnmapped);
             return hasParentQueryBuilder;
         }
@@ -321,9 +352,125 @@ public class HasParentQueryBuilder extends AbstractQueryBuilder<HasParentQueryBu
     }
 
     @Override
-    protected void extractInnerHitBuilders(Map<String, InnerHitBuilder> innerHits) {
-        if (innerHit!= null) {
-            innerHit.inlineInnerHits(innerHits);
+    protected void extractInnerHitBuilders(Map<String, InnerHitContextBuilder> innerHits) {
+        if (innerHitBuilder != null) {
+            Map<String, InnerHitContextBuilder> children = new HashMap<>();
+            InnerHitContextBuilder.extractInnerHits(query, children);
+            String name = innerHitBuilder.getName() != null ? innerHitBuilder.getName() : type;
+            InnerHitContextBuilder innerHitContextBuilder =
+                new ParentChildInnerHitContextBuilder(type, query, innerHitBuilder, children);
+            innerHits.put(name, innerHitContextBuilder);
+        }
+    }
+
+    static class ParentChildInnerHitContextBuilder extends InnerHitContextBuilder {
+        private final String typeName;
+
+        ParentChildInnerHitContextBuilder(String typeName, QueryBuilder query, InnerHitBuilder innerHitBuilder,
+                                          Map<String, InnerHitContextBuilder> children) {
+            super(query, innerHitBuilder, children);
+            this.typeName = typeName;
+        }
+
+        @Override
+        public void build(SearchContext parentSearchContext, InnerHitsContext innerHitsContext) throws IOException {
+            QueryShardContext queryShardContext = parentSearchContext.getQueryShardContext();
+            DocumentMapper documentMapper = queryShardContext.documentMapper(typeName);
+            if (documentMapper == null) {
+                if (innerHitBuilder.isIgnoreUnmapped() == false) {
+                    throw new IllegalStateException("[" + query.getName() + "] no mapping found for type [" + typeName + "]");
+                } else {
+                    return;
+                }
+            }
+            String name = innerHitBuilder.getName() != null ? innerHitBuilder.getName() : documentMapper.type();
+            ParentChildInnerHitSubContext parentChildInnerHits = new ParentChildInnerHitSubContext(
+                name, parentSearchContext, queryShardContext.getMapperService(), documentMapper
+            );
+            setupInnerHitsContext(queryShardContext, parentChildInnerHits);
+            innerHitsContext.addInnerHitDefinition(parentChildInnerHits);
+        }
+    }
+
+    static final class ParentChildInnerHitSubContext extends InnerHitsContext.InnerHitSubContext {
+        private final MapperService mapperService;
+        private final DocumentMapper documentMapper;
+
+        ParentChildInnerHitSubContext(String name, SearchContext context, MapperService mapperService, DocumentMapper documentMapper) {
+            super(name, 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, BooleanClause.Occur.FILTER)
+                    // Only include docs that have this inner hits type
+                    .add(documentMapper.typeFilter(context.getQueryShardContext()), BooleanClause.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());
         }
     }
 }
diff --git a/modules/parent-join/src/test/java/org/elasticsearch/join/aggregations/ChildrenIT.java b/modules/parent-join/src/test/java/org/elasticsearch/join/aggregations/ChildrenIT.java
index 8da6dbcdf6..1b67b2a22b 100644
--- a/modules/parent-join/src/test/java/org/elasticsearch/join/aggregations/ChildrenIT.java
+++ b/modules/parent-join/src/test/java/org/elasticsearch/join/aggregations/ChildrenIT.java
@@ -356,7 +356,7 @@ public class ChildrenIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch(indexName).setTypes(masterType)
                 .setQuery(hasChildQuery(childType, termQuery("color", "orange"), ScoreMode.None))
-.addAggregation(children("my-refinements", childType)
+                        .addAggregation(children("my-refinements", childType)
                                 .subAggregation(terms("my-colors").field("color"))
                                 .subAggregation(terms("my-sizes").field("size"))
                 ).get();
diff --git a/modules/parent-join/src/test/java/org/elasticsearch/join/query/ChildQuerySearchIT.java b/modules/parent-join/src/test/java/org/elasticsearch/join/query/ChildQuerySearchIT.java
index 87efdf0a5c..2c828bb3cc 100644
--- a/modules/parent-join/src/test/java/org/elasticsearch/join/query/ChildQuerySearchIT.java
+++ b/modules/parent-join/src/test/java/org/elasticsearch/join/query/ChildQuerySearchIT.java
@@ -872,7 +872,7 @@ public class ChildQuerySearchIT extends ESIntegTestCase {
                 hasChildQuery("child", matchQuery("c_field", "foo"), ScoreMode.None)
                     .innerHit(new InnerHitBuilder().setHighlightBuilder(
                         new HighlightBuilder().field(new Field("c_field")
-                                .highlightQuery(QueryBuilders.matchQuery("c_field", "bar")))), false))
+                                .highlightQuery(QueryBuilders.matchQuery("c_field", "bar"))))))
                 .get();
         assertNoFailures(searchResponse);
         assertThat(searchResponse.getHits().getTotalHits(), equalTo(1L));
diff --git a/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasChildQueryBuilderTests.java b/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasChildQueryBuilderTests.java
index 8f4fc9d0c3..e95f3a7713 100644
--- a/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasChildQueryBuilderTests.java
+++ b/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasChildQueryBuilderTests.java
@@ -42,6 +42,7 @@ import org.elasticsearch.index.mapper.Uid;
 import org.elasticsearch.index.mapper.UidFieldMapper;
 import org.elasticsearch.index.query.IdsQueryBuilder;
 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.QueryShardContext;
@@ -56,6 +57,7 @@ import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.AbstractQueryTestCase;
+import org.elasticsearch.test.VersionUtils;
 
 import java.io.IOException;
 import java.util.Collection;
@@ -128,7 +130,8 @@ public class HasChildQueryBuilderTests extends AbstractQueryTestCase<HasChildQue
             hqb.innerHit(new InnerHitBuilder()
                     .setName(randomAlphaOfLengthBetween(1, 10))
                     .setSize(randomIntBetween(0, 100))
-                    .addSort(new FieldSortBuilder(STRING_FIELD_NAME_2).order(SortOrder.ASC)), hqb.ignoreUnmapped());
+                    .addSort(new FieldSortBuilder(STRING_FIELD_NAME_2).order(SortOrder.ASC))
+                    .setIgnoreUnmapped(hqb.ignoreUnmapped()));
         }
         return hqb;
     }
@@ -144,15 +147,15 @@ public class HasChildQueryBuilderTests extends AbstractQueryTestCase<HasChildQue
             // have to rewrite again because the provided queryBuilder hasn't been rewritten (directly returned from
             // doCreateTestQueryBuilder)
             queryBuilder = (HasChildQueryBuilder) queryBuilder.rewrite(searchContext.getQueryShardContext());
-            Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-            InnerHitBuilder.extractInnerHits(queryBuilder, innerHitBuilders);
-            for (InnerHitBuilder builder : innerHitBuilders.values()) {
+            Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+            InnerHitContextBuilder.extractInnerHits(queryBuilder, innerHitBuilders);
+            for (InnerHitContextBuilder builder : innerHitBuilders.values()) {
                 builder.build(searchContext, searchContext.innerHits());
             }
             assertNotNull(searchContext.innerHits());
             assertEquals(1, searchContext.innerHits().getInnerHits().size());
             assertTrue(searchContext.innerHits().getInnerHits().containsKey(queryBuilder.innerHit().getName()));
-            InnerHitsContext.BaseInnerHits innerHits =
+            InnerHitsContext.InnerHitSubContext innerHits =
                     searchContext.innerHits().getInnerHits().get(queryBuilder.innerHit().getName());
             assertEquals(innerHits.size(), queryBuilder.innerHit().getSize());
             assertEquals(innerHits.sort().sort.getSort().length, 1);
@@ -160,6 +163,20 @@ public class HasChildQueryBuilderTests extends AbstractQueryTestCase<HasChildQue
         }
     }
 
+    /**
+     * Test (de)serialization on all previous released versions
+     */
+    public void testSerializationBWC() throws IOException {
+        for (Version version : VersionUtils.allReleasedVersions()) {
+            HasChildQueryBuilder testQuery = createTestQueryBuilder();
+            if (version.before(Version.V_5_2_0) && testQuery.innerHit() != null) {
+                // ignore unmapped for inner_hits has been added on 5.2
+                testQuery.innerHit().setIgnoreUnmapped(false);
+            }
+            assertSerialization(testQuery, version);
+        }
+    }
+
     public void testIllegalValues() {
         QueryBuilder query = new MatchAllQueryBuilder();
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
@@ -231,7 +248,7 @@ public class HasChildQueryBuilderTests extends AbstractQueryTestCase<HasChildQue
         assertEquals(query, queryBuilder.childType(), "child");
         assertEquals(query, queryBuilder.scoreMode(), ScoreMode.Avg);
         assertNotNull(query, queryBuilder.innerHit());
-        InnerHitBuilder expected = new InnerHitBuilder(new InnerHitBuilder(), queryBuilder.query(), "child", false)
+        InnerHitBuilder expected = new InnerHitBuilder("child")
                 .setName("inner_hits_name")
                 .setSize(100)
                 .addSort(new FieldSortBuilder("mapped_string").order(SortOrder.ASC));
diff --git a/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasParentQueryBuilderTests.java b/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasParentQueryBuilderTests.java
index 825dfede61..62ab92ac5e 100644
--- a/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasParentQueryBuilderTests.java
+++ b/modules/parent-join/src/test/java/org/elasticsearch/join/query/HasParentQueryBuilderTests.java
@@ -22,6 +22,7 @@ package org.elasticsearch.join.query;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.join.ScoreMode;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.xcontent.ToXContent;
@@ -30,6 +31,7 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.query.IdsQueryBuilder;
 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.QueryShardContext;
@@ -43,6 +45,7 @@ import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.AbstractQueryTestCase;
+import org.elasticsearch.test.VersionUtils;
 
 import java.io.IOException;
 import java.util.Collection;
@@ -108,7 +111,8 @@ public class HasParentQueryBuilderTests extends AbstractQueryTestCase<HasParentQ
             hqb.innerHit(new InnerHitBuilder()
                     .setName(randomAlphaOfLengthBetween(1, 10))
                     .setSize(randomIntBetween(0, 100))
-                    .addSort(new FieldSortBuilder(STRING_FIELD_NAME_2).order(SortOrder.ASC)), hqb.ignoreUnmapped());
+                    .addSort(new FieldSortBuilder(STRING_FIELD_NAME_2).order(SortOrder.ASC))
+                    .setIgnoreUnmapped(hqb.ignoreUnmapped()));
         }
         return hqb;
     }
@@ -125,15 +129,15 @@ public class HasParentQueryBuilderTests extends AbstractQueryTestCase<HasParentQ
             queryBuilder = (HasParentQueryBuilder) queryBuilder.rewrite(searchContext.getQueryShardContext());
 
             assertNotNull(searchContext);
-            Map<String, InnerHitBuilder> innerHitBuilders = new HashMap<>();
-            InnerHitBuilder.extractInnerHits(queryBuilder, innerHitBuilders);
-            for (InnerHitBuilder builder : innerHitBuilders.values()) {
+            Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
+            InnerHitContextBuilder.extractInnerHits(queryBuilder, innerHitBuilders);
+            for (InnerHitContextBuilder builder : innerHitBuilders.values()) {
                 builder.build(searchContext, searchContext.innerHits());
             }
             assertNotNull(searchContext.innerHits());
             assertEquals(1, searchContext.innerHits().getInnerHits().size());
             assertTrue(searchContext.innerHits().getInnerHits().containsKey(queryBuilder.innerHit().getName()));
-            InnerHitsContext.BaseInnerHits innerHits = searchContext.innerHits()
+            InnerHitsContext.InnerHitSubContext innerHits = searchContext.innerHits()
                     .getInnerHits().get(queryBuilder.innerHit().getName());
             assertEquals(innerHits.size(), queryBuilder.innerHit().getSize());
             assertEquals(innerHits.sort().sort.getSort().length, 1);
@@ -141,6 +145,20 @@ public class HasParentQueryBuilderTests extends AbstractQueryTestCase<HasParentQ
         }
     }
 
+    /**
+     * Test (de)serialization on all previous released versions
+     */
+    public void testSerializationBWC() throws IOException {
+        for (Version version : VersionUtils.allReleasedVersions()) {
+            HasParentQueryBuilder testQuery = createTestQueryBuilder();
+            if (version.before(Version.V_5_2_0) && testQuery.innerHit() != null) {
+                // ignore unmapped for inner_hits has been added on 5.2
+                testQuery.innerHit().setIgnoreUnmapped(false);
+            }
+            assertSerialization(testQuery, version);
+        }
+    }
+
     public void testIllegalValues() throws IOException {
         QueryBuilder query = new MatchAllQueryBuilder();
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
diff --git a/modules/parent-join/src/test/java/org/elasticsearch/join/query/InnerHitsIT.java b/modules/parent-join/src/test/java/org/elasticsearch/join/query/InnerHitsIT.java
index ad8e49e9f5..1d2e1c29e9 100644
--- a/modules/parent-join/src/test/java/org/elasticsearch/join/query/InnerHitsIT.java
+++ b/modules/parent-join/src/test/java/org/elasticsearch/join/query/InnerHitsIT.java
@@ -117,7 +117,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch("articles")
             .setQuery(hasChildQuery("comment", matchQuery("message", "fox"), ScoreMode.None)
-                .innerHit(new InnerHitBuilder(), false))
+                .innerHit(new InnerHitBuilder()))
             .get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -135,7 +135,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         response = client().prepareSearch("articles")
             .setQuery(hasChildQuery("comment", matchQuery("message", "elephant"), ScoreMode.None)
-                .innerHit(new InnerHitBuilder(), false))
+                .innerHit(new InnerHitBuilder()))
             .get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -160,8 +160,7 @@ public class InnerHitsIT extends ESIntegTestCase {
                         .setHighlightBuilder(new HighlightBuilder().field("message"))
                         .setExplain(true).setSize(1)
                         .addScriptField("script", new Script(ScriptType.INLINE, MockScriptEngine.NAME, "5",
-                            Collections.emptyMap())),
-                    false)
+                            Collections.emptyMap())))
             ).get();
         assertNoFailures(response);
         innerHits = response.getHits().getAt(0).getInnerHits().get("comment");
@@ -209,10 +208,10 @@ public class InnerHitsIT extends ESIntegTestCase {
         BoolQueryBuilder boolQuery = new BoolQueryBuilder();
         boolQuery.should(constantScoreQuery(hasChildQuery("child1", matchAllQuery(), ScoreMode.None)
             .innerHit(new InnerHitBuilder().setName("a")
-                .addSort(new FieldSortBuilder("_uid").order(SortOrder.ASC)).setSize(size), false)));
+                .addSort(new FieldSortBuilder("_uid").order(SortOrder.ASC)).setSize(size))));
         boolQuery.should(constantScoreQuery(hasChildQuery("child2", matchAllQuery(), ScoreMode.None)
             .innerHit(new InnerHitBuilder().setName("b")
-                .addSort(new FieldSortBuilder("_uid").order(SortOrder.ASC)).setSize(size), false)));
+                .addSort(new FieldSortBuilder("_uid").order(SortOrder.ASC)).setSize(size))));
         SearchResponse searchResponse = client().prepareSearch("idx")
             .setSize(numDocs)
             .setTypes("parent")
@@ -279,7 +278,7 @@ public class InnerHitsIT extends ESIntegTestCase {
             .setQuery(
                 boolQuery()
                     .must(matchQuery("body", "fail2ban"))
-                    .must(hasParentQuery("question", matchAllQuery(), false).innerHit(new InnerHitBuilder(), false))
+                    .must(hasParentQuery("question", matchAllQuery(), false).innerHit(new InnerHitBuilder()))
             ).get();
         assertNoFailures(response);
         assertHitCount(response, 2);
@@ -318,8 +317,8 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch("articles")
             .setQuery(hasChildQuery("comment",
-                hasChildQuery("remark", matchQuery("message", "good"), ScoreMode.None).innerHit(new InnerHitBuilder(), false),
-                ScoreMode.None).innerHit(new InnerHitBuilder(), false))
+                hasChildQuery("remark", matchQuery("message", "good"), ScoreMode.None).innerHit(new InnerHitBuilder()),
+                ScoreMode.None).innerHit(new InnerHitBuilder()))
             .get();
 
         assertNoFailures(response);
@@ -339,8 +338,8 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         response = client().prepareSearch("articles")
             .setQuery(hasChildQuery("comment",
-                hasChildQuery("remark", matchQuery("message", "bad"), ScoreMode.None).innerHit(new InnerHitBuilder(), false),
-                ScoreMode.None).innerHit(new InnerHitBuilder(), false))
+                hasChildQuery("remark", matchQuery("message", "bad"), ScoreMode.None).innerHit(new InnerHitBuilder()),
+                ScoreMode.None).innerHit(new InnerHitBuilder()))
             .get();
 
         assertNoFailures(response);
@@ -397,16 +396,16 @@ public class InnerHitsIT extends ESIntegTestCase {
             .setTypes("duke")
             .setQuery(boolQuery()
                 .filter(hasParentQuery("prince",
-                    hasParentQuery("king", matchAllQuery(), false).innerHit(new InnerHitBuilder().setName("kings"), false),
-                    false).innerHit(new InnerHitBuilder().setName("princes"), false)
+                    hasParentQuery("king", matchAllQuery(), false).innerHit(new InnerHitBuilder().setName("kings")),
+                    false).innerHit(new InnerHitBuilder().setName("princes"))
                 )
                 .filter(hasChildQuery("earl",
                     hasChildQuery("baron", matchAllQuery(), ScoreMode.None)
-                        .innerHit(new InnerHitBuilder().setName("barons"), false),
+                        .innerHit(new InnerHitBuilder().setName("barons")),
                     ScoreMode.None).innerHit(new InnerHitBuilder()
                     .addSort(SortBuilders.fieldSort("_uid").order(SortOrder.ASC))
                     .setName("earls")
-                    .setSize(4), false)
+                    .setSize(4))
                 )
             )
             .get();
@@ -459,7 +458,7 @@ public class InnerHitsIT extends ESIntegTestCase {
 
         SearchResponse response = client().prepareSearch("index")
             .setQuery(hasChildQuery("child", matchQuery("field", "value1").queryName("_name1"), ScoreMode.None)
-                .innerHit(new InnerHitBuilder(), false))
+                .innerHit(new InnerHitBuilder()))
             .addSort("_uid", SortOrder.ASC)
             .get();
         assertHitCount(response, 2);
@@ -474,7 +473,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         assertThat(response.getHits().getAt(1).getInnerHits().get("child").getAt(0).getMatchedQueries()[0], equalTo("_name1"));
 
         QueryBuilder query = hasChildQuery("child", matchQuery("field", "value2").queryName("_name2"), ScoreMode.None)
-            .innerHit(new InnerHitBuilder(), false);
+            .innerHit(new InnerHitBuilder());
         response = client().prepareSearch("index")
             .setQuery(query)
             .addSort("_uid", SortOrder.ASC)
@@ -496,7 +495,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         indexRandom(true, requests);
 
         QueryBuilder query = hasChildQuery("child", matchQuery("field", "value1"), ScoreMode.None)
-            .innerHit(new InnerHitBuilder().setSize(ArrayUtil.MAX_ARRAY_LENGTH - 1), false);
+            .innerHit(new InnerHitBuilder().setSize(ArrayUtil.MAX_ARRAY_LENGTH - 1));
         SearchResponse response = client().prepareSearch("index1")
             .setQuery(query)
             .get();
@@ -515,7 +514,7 @@ public class InnerHitsIT extends ESIntegTestCase {
             .get();
 
         query = nestedQuery("nested", matchQuery("nested.field", "value1"), ScoreMode.Avg)
-            .innerHit(new InnerHitBuilder().setSize(ArrayUtil.MAX_ARRAY_LENGTH - 1), false);
+            .innerHit(new InnerHitBuilder().setSize(ArrayUtil.MAX_ARRAY_LENGTH - 1));
         response = client().prepareSearch("index2")
             .setQuery(query)
             .get();
@@ -534,7 +533,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         SearchResponse response = client().prepareSearch("test")
             .setQuery(boolQuery().must(matchQuery("key", "value"))
                 .should(hasChildQuery("child_type", nestedQuery("nested_type", matchAllQuery(), ScoreMode.None)
-                    .innerHit(new InnerHitBuilder(), false), ScoreMode.None).innerHit(new InnerHitBuilder(), false)))
+                    .innerHit(new InnerHitBuilder()), ScoreMode.None).innerHit(new InnerHitBuilder())))
             .get();
         assertHitCount(response, 1);
         SearchHit hit = response.getHits().getAt(0);
@@ -557,7 +556,7 @@ public class InnerHitsIT extends ESIntegTestCase {
         SearchResponse response = client().prepareSearch("index1", "index2")
             .setQuery(boolQuery()
                 .should(hasChildQuery("child_type", matchAllQuery(), ScoreMode.None).ignoreUnmapped(true)
-                    .innerHit(new InnerHitBuilder(), true))
+                    .innerHit(new InnerHitBuilder().setIgnoreUnmapped(true)))
                 .should(termQuery("key", "value"))
             )
             .get();
diff --git a/test/framework/src/main/java/org/elasticsearch/test/AbstractQueryTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/AbstractQueryTestCase.java
index 56275ac828..3f44c24494 100644
--- a/test/framework/src/main/java/org/elasticsearch/test/AbstractQueryTestCase.java
+++ b/test/framework/src/main/java/org/elasticsearch/test/AbstractQueryTestCase.java
@@ -706,13 +706,19 @@ public abstract class AbstractQueryTestCase<QB extends AbstractQueryBuilder<QB>>
         }
     }
 
+    protected static QueryBuilder assertSerialization(QueryBuilder testQuery) throws IOException {
+        return assertSerialization(testQuery, Version.CURRENT);
+    }
+
     /**
      * Serialize the given query builder and asserts that both are equal
      */
-    protected static QueryBuilder assertSerialization(QueryBuilder testQuery) throws IOException {
+    protected static QueryBuilder assertSerialization(QueryBuilder testQuery, Version version) throws IOException {
         try (BytesStreamOutput output = new BytesStreamOutput()) {
+            output.setVersion(version);
             output.writeNamedWriteable(testQuery);
             try (StreamInput in = new NamedWriteableAwareStreamInput(output.bytes().streamInput(), serviceHolder.namedWriteableRegistry)) {
+                in.setVersion(version);
                 QueryBuilder deserializedQuery = in.readNamedWriteable(QueryBuilder.class);
                 assertEquals(testQuery, deserializedQuery);
                 assertEquals(testQuery.hashCode(), deserializedQuery.hashCode());