Expose sequence number and primary terms in search responses (#37639)

Users may require the sequence number and primary terms to perform optimistic concurrency control operations. Currently, you can get the sequence number via the `docvalues_fields` API but the primary term is not accessible because it is maintained by the `SeqNoFieldMapper` and the infrastructure can't find it. 

This commit adds a dedicated sub fetch phase to return both numbers that is connected to a new `seq_no_primary_term` parameter.

docs/reference/aggregations/metrics/tophits-aggregation.asciidoc

@@ -25,6 +25,7 @@ The top_hits aggregation returns regular search hits, because of this many per h
 * <<search-request-script-fields,Script fields>>
 * <<search-request-docvalue-fields,Doc value fields>>
 * <<search-request-version,Include versions>>
+* <<search-request-seq-no-primary-term,Include Sequence Numbers and Primary Terms>>
 
 ==== Example
 

docs/reference/docs/concurrency-control.asciidoc

@@ -87,7 +87,7 @@ returns:
 
 
 Note: The <<search-search,Search API>> can return the `_seq_no` and `_primary_term`
-for each search hit by requesting the `_seq_no` and `_primary_term` <<search-request-docvalue-fields,Doc Value Fields>>.
+for each search hit by setting <<search-request-seq-no-primary-term,`seq_no_primary_term` parameter>>.
 
 The sequence number and the primary term uniquely identify a change. By noting down 
 the sequence number and primary term returned, you can make sure to only change the

docs/reference/search/request-body.asciidoc

@@ -213,7 +213,7 @@ include::request/preference.asciidoc[]
 
 include::request/explain.asciidoc[]
 
-include::request/version.asciidoc[]
+include::request/version-and-seq-no.asciidoc[]
 
 include::request/index-boost.asciidoc[]
 

docs/reference/search/request/inner-hits.asciidoc

@@ -76,6 +76,7 @@ Inner hits also supports the following per document features:
 * <<search-request-script-fields,Script fields>>
 * <<search-request-docvalue-fields,Doc value fields>>
 * <<search-request-version,Include versions>>
+* <<search-request-seq-no-primary-term,Include Sequence Numbers and Primary Terms>>
 
 [[nested-inner-hits]]
 ==== Nested inner hits

docs/reference/search/request/version-and-seq-no.asciidoc

@@ -0,0 +1,34 @@
+[[search-request-seq-no-primary-term]]
+=== Sequence Numbers and Primary Term
+
+Returns the sequence number and primary term of the last modification to each search hit.
+See <<optimistic-concurrency-control>> for more details.
+
+[source,js]
+--------------------------------------------------
+GET /_search
+{
+    "seq_no_primary_term": true,
+    "query" : {
+        "term" : { "user" : "kimchy" }
+    }
+}
+--------------------------------------------------
+// CONSOLE
+
+[[search-request-version]]
+=== Version
+
+Returns a version for each search hit.
+
+[source,js]
+--------------------------------------------------
+GET /_search
+{
+    "version": true,
+    "query" : {
+        "term" : { "user" : "kimchy" }
+    }
+}
+--------------------------------------------------
+// CONSOLE

docs/reference/search/request/version.asciidoc

@@ -1,16 +0,0 @@
-[[search-request-version]]
-=== Version
-
-Returns a version for each search hit.
-
-[source,js]
---------------------------------------------------
-GET /_search
-{
-    "version": true,
-    "query" : {
-        "term" : { "user" : "kimchy" }
-    }
-}
---------------------------------------------------
-// CONSOLE

modules/parent-join/src/test/java/org/elasticsearch/join/query/HasChildQueryBuilderTests.java

@@ -252,6 +252,7 @@ public class HasChildQueryBuilderTests extends AbstractQueryTestCase<HasChildQue
                 "      \"from\" : 0,\n" +
                 "      \"size\" : 100,\n" +
                 "      \"version\" : false,\n" +
+                "      \"seq_no_primary_term\" : false,\n" +
                 "      \"explain\" : false,\n" +
                 "      \"track_scores\" : false,\n" +
                 "      \"sort\" : [ {\n" +

modules/parent-join/src/test/java/org/elasticsearch/join/query/InnerHitsIT.java

@@ -56,6 +56,8 @@ import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
 import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
 import static org.elasticsearch.index.query.QueryBuilders.nestedQuery;
 import static org.elasticsearch.index.query.QueryBuilders.termQuery;
+import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
+import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
 import static org.elasticsearch.join.query.JoinQueryBuilders.hasChildQuery;
 import static org.elasticsearch.join.query.JoinQueryBuilders.hasParentQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
@@ -66,6 +68,7 @@ import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSear
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.hasId;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 
@@ -133,9 +136,10 @@ public class InnerHitsIT extends ParentChildTestCase {
         assertThat(innerHits.getAt(1).getId(), equalTo("c2"));
         assertThat(innerHits.getAt(1).getType(), equalTo("doc"));
 
+        final boolean seqNoAndTerm = randomBoolean();
         response = client().prepareSearch("articles")
             .setQuery(hasChildQuery("comment", matchQuery("message", "elephant"), ScoreMode.None)
-                .innerHit(new InnerHitBuilder()))
+                .innerHit(new InnerHitBuilder().setSeqNoAndPrimaryTerm(seqNoAndTerm)))
             .get();
         assertNoFailures(response);
         assertHitCount(response, 1);
@@ -152,6 +156,22 @@ public class InnerHitsIT extends ParentChildTestCase {
         assertThat(innerHits.getAt(2).getId(), equalTo("c6"));
         assertThat(innerHits.getAt(2).getType(), equalTo("doc"));
 
+        if (seqNoAndTerm) {
+            assertThat(innerHits.getAt(0).getPrimaryTerm(), equalTo(1L));
+            assertThat(innerHits.getAt(1).getPrimaryTerm(), equalTo(1L));
+            assertThat(innerHits.getAt(2).getPrimaryTerm(), equalTo(1L));
+            assertThat(innerHits.getAt(0).getSeqNo(), greaterThanOrEqualTo(0L));
+            assertThat(innerHits.getAt(1).getSeqNo(), greaterThanOrEqualTo(0L));
+            assertThat(innerHits.getAt(2).getSeqNo(), greaterThanOrEqualTo(0L));
+        } else {
+            assertThat(innerHits.getAt(0).getPrimaryTerm(), equalTo(UNASSIGNED_PRIMARY_TERM));
+            assertThat(innerHits.getAt(1).getPrimaryTerm(), equalTo(UNASSIGNED_PRIMARY_TERM));
+            assertThat(innerHits.getAt(2).getPrimaryTerm(), equalTo(UNASSIGNED_PRIMARY_TERM));
+            assertThat(innerHits.getAt(0).getSeqNo(), equalTo(UNASSIGNED_SEQ_NO));
+            assertThat(innerHits.getAt(1).getSeqNo(), equalTo(UNASSIGNED_SEQ_NO));
+            assertThat(innerHits.getAt(2).getSeqNo(), equalTo(UNASSIGNED_SEQ_NO));
+        }
+
         response = client().prepareSearch("articles")
             .setQuery(
                 hasChildQuery("comment", matchQuery("message", "fox"), ScoreMode.None).innerHit(

modules/parent-join/src/test/resources/rest-api-spec/test/11_parent_child.yml

@@ -11,26 +11,26 @@ setup:
                     relations:
                       parent: child
 
----
-"Parent/child inner hits":
-    - do:
-        index:
-          index: test
-          type:  doc
-          id:    1
-          body:  {"foo": "bar", "join_field": {"name" : "parent"} }
+  - do:
+      index:
+        index: test
+        type:  doc
+        id:    1
+        body:  {"foo": "bar", "join_field": {"name" : "parent"} }
 
-    - do:
-        index:
-          index: test
-          type:  doc
-          id:    2
-          routing: 1
-          body:  {"bar": "baz", "join_field": { "name" : "child", "parent": "1"} }
+  - do:
+      index:
+        index: test
+        type:  doc
+        id:    2
+        routing: 1
+        body:  {"bar": "baz", "join_field": { "name" : "child", "parent": "1"} }
 
-    - do:
-        indices.refresh: {}
+  - do:
+      indices.refresh: {}
 
+---
+"Parent/child inner hits":
     - do:
         search:
           rest_total_hits_as_int: true
@@ -41,3 +41,24 @@ setup:
     - match: { hits.hits.0.inner_hits.child.hits.hits.0._index: "test"}
     - match: { hits.hits.0.inner_hits.child.hits.hits.0._id: "2" }
     - is_false: hits.hits.0.inner_hits.child.hits.hits.0._nested
+
+---
+"Parent/child inner hits with seq no":
+  - skip:
+      version: " - 6.99.99"
+      reason: support was added in 7.0
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        body: { "query" : { "has_child" :
+            { "type" : "child", "query" : { "match_all" : {} }, "inner_hits" : { "seq_no_primary_term": true} }
+        } }
+  - match: { hits.total: 1 }
+  - match: { hits.hits.0._index: "test" }
+  - match: { hits.hits.0._id: "1" }
+  - match: { hits.hits.0.inner_hits.child.hits.hits.0._index: "test"}
+  - match: { hits.hits.0.inner_hits.child.hits.hits.0._id: "2" }
+  - is_false: hits.hits.0.inner_hits.child.hits.hits.0._nested
+  - gte: { hits.hits.0.inner_hits.child.hits.hits.0._seq_no: 0 }
+  - gte: { hits.hits.0.inner_hits.child.hits.hits.0._primary_term: 1 }

modules/reindex/src/test/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuildersTests.java

@@ -179,7 +179,7 @@ public class RemoteRequestBuildersTests extends ESTestCase {
             fetchVersion = randomBoolean();
             searchRequest.source().version(fetchVersion);
         }
-
+        
         Map<String, String> params = initialSearch(searchRequest, query, remoteVersion).getParameters();
 
         if (scroll == null) {

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/multi_cluster/10_basic.yml

@@ -30,6 +30,7 @@
         rest_total_hits_as_int: true
         index: test_index,my_remote_cluster:test_index
         body:
+          seq_no_primary_term: true
           aggs:
             cluster:
               terms:
@@ -37,6 +38,8 @@
 
   - match: { _shards.total: 5 }
   - match: { hits.total: 11 }
+  - gte: { hits.hits.0._seq_no: 0 }
+  - gte: { hits.hits.0._primary_term: 1 }
   - length: { aggregations.cluster.buckets: 2 }
   - match: { aggregations.cluster.buckets.0.key: "remote_cluster" }
   - match: { aggregations.cluster.buckets.0.doc_count: 6 }

rest-api-spec/src/main/resources/rest-api-spec/api/search.json

@@ -164,6 +164,10 @@
           "type" : "boolean",
           "description" : "Specify whether to return document version as part of a hit"
         },
+        "seq_no_primary_term": {
+          "type" : "boolean",
+          "description" : "Specify whether to return sequence number and primary term of the last modification of each hit"
+        },
         "request_cache": {
           "type" : "boolean",
           "description" : "Specify if request cache should be used for this request or not, defaults to index level setting"

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/200_top_hits_metric.yml

@@ -1,8 +1,4 @@
----
-"top_hits aggregation with nested documents":
-  - skip:
-          version: " - 6.1.99"
-          reason: "<= 6.1 nodes don't always include index or id in nested top hits"
+setup:
   - do:
       indices.create:
         index: my-index
@@ -54,6 +50,12 @@
             ]
           }
 
+---
+"top_hits aggregation with nested documents":
+  - skip:
+          version: " - 6.1.99"
+          reason: "<= 6.1 nodes don't always include index or id in nested top hits"
+
   - do:
       search:
         rest_total_hits_as_int: true
@@ -81,3 +83,35 @@
   - match: { aggregations.to-users.users.hits.hits.2._index: my-index }
   - match: { aggregations.to-users.users.hits.hits.2._nested.field: users }
   - match: { aggregations.to-users.users.hits.hits.2._nested.offset: 1 }
+
+
+---
+"top_hits aggregation with sequence numbers":
+  - skip:
+      version: " - 6.99.99"
+      reason: support was added in 7.0
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        body:
+          aggs:
+            groups:
+              terms:
+                field: group.keyword
+              aggs:
+                users:
+                  top_hits:
+                    sort: "users.last.keyword"
+                    seq_no_primary_term: true
+
+  - match: { hits.total: 2 }
+  - length: { aggregations.groups.buckets.0.users.hits.hits: 2 }
+  - match: { aggregations.groups.buckets.0.users.hits.hits.0._id: "1" }
+  - match: { aggregations.groups.buckets.0.users.hits.hits.0._index: my-index }
+  - gte: { aggregations.groups.buckets.0.users.hits.hits.0._seq_no: 0 }
+  - gte: { aggregations.groups.buckets.0.users.hits.hits.0._primary_term: 1 }
+  - match: { aggregations.groups.buckets.0.users.hits.hits.1._id: "2" }
+  - match: { aggregations.groups.buckets.0.users.hits.hits.1._index: my-index }
+  - gte: { aggregations.groups.buckets.0.users.hits.hits.1._seq_no: 0 }
+  - gte: { aggregations.groups.buckets.0.users.hits.hits.1._primary_term: 1 }

rest-api-spec/src/main/resources/rest-api-spec/test/search/110_field_collapsing.yml

@@ -405,3 +405,57 @@ setup:
   - match: { hits.hits.1.inner_hits.sub_hits.hits.total: 3}
   - match: { hits.hits.2.fields.group_alias: [25] }
   - match: { hits.hits.2.inner_hits.sub_hits.hits.total: 2}
+
+---
+"field collapsing, inner_hits and seq_no":
+
+  - skip:
+      version: " - 6.99.0"
+      reason:  "sequence numbers introduced in 7.0.0"
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: test
+        body:
+          collapse: { field: numeric_group, inner_hits: {
+            name: sub_hits, seq_no_primary_term: true, size: 2, sort: [{ sort: asc }]
+          } }
+          sort: [{ sort: desc }]
+
+  - match: { hits.total: 6 }
+  - length: { hits.hits: 3 }
+  - match: { hits.hits.0._index: test }
+  - match: { hits.hits.0.fields.numeric_group: [3] }
+  - match: { hits.hits.0.sort: [36] }
+  - match: { hits.hits.0._id: "6" }
+  - match: { hits.hits.0.inner_hits.sub_hits.hits.total: 1 }
+  - length: { hits.hits.0.inner_hits.sub_hits.hits.hits: 1 }
+  - match: { hits.hits.0.inner_hits.sub_hits.hits.hits.0._id: "6" }
+  - gte: { hits.hits.0.inner_hits.sub_hits.hits.hits.0._seq_no: 0 }
+  - gte: { hits.hits.0.inner_hits.sub_hits.hits.hits.0._primary_term: 1 }
+  - match: { hits.hits.1._index: test }
+  - match: { hits.hits.1.fields.numeric_group: [1] }
+  - match: { hits.hits.1.sort: [24] }
+  - match: { hits.hits.1._id: "3" }
+  - match: { hits.hits.1.inner_hits.sub_hits.hits.total: 3 }
+  - length: { hits.hits.1.inner_hits.sub_hits.hits.hits: 2 }
+  - match: { hits.hits.1.inner_hits.sub_hits.hits.hits.0._id: "2" }
+  - gte: { hits.hits.1.inner_hits.sub_hits.hits.hits.0._seq_no: 0 }
+  - gte: { hits.hits.1.inner_hits.sub_hits.hits.hits.0._primary_term: 1 }
+  - match: { hits.hits.1.inner_hits.sub_hits.hits.hits.1._id: "1" }
+  - gte: { hits.hits.1.inner_hits.sub_hits.hits.hits.1._seq_no: 0 }
+  - gte: { hits.hits.1.inner_hits.sub_hits.hits.hits.1._primary_term: 1 }
+  - match: { hits.hits.2._index: test }
+  - match: { hits.hits.2._type: test }
+  - match: { hits.hits.2.fields.numeric_group: [25] }
+  - match: { hits.hits.2.sort: [10] }
+  - match: { hits.hits.2._id: "4" }
+  - match: { hits.hits.2.inner_hits.sub_hits.hits.total: 2 }
+  - length: { hits.hits.2.inner_hits.sub_hits.hits.hits: 2 }
+  - match: { hits.hits.2.inner_hits.sub_hits.hits.hits.0._id: "5" }
+  - gte: { hits.hits.2.inner_hits.sub_hits.hits.hits.0._seq_no: 0 }
+  - gte: { hits.hits.2.inner_hits.sub_hits.hits.hits.0._primary_term: 1 }
+  - match: { hits.hits.2.inner_hits.sub_hits.hits.hits.1._id: "4" }
+  - gte: { hits.hits.2.inner_hits.sub_hits.hits.hits.1._seq_no: 0 }
+  - gte: { hits.hits.2.inner_hits.sub_hits.hits.hits.1._primary_term: 1 }

rest-api-spec/src/main/resources/rest-api-spec/test/search/300_sequence_numbers.yml

@@ -0,0 +1,74 @@
+setup:
+  - do:
+      indices.create:
+          index:  test_1
+
+  - do:
+      index:
+          index:  test_1
+          type:   test
+          id:     1
+          body:   { foo: foo }
+
+## we index again in order to make the seq# 1 (so we can check for the field existence with is_false)
+  - do:
+      index:
+        index:  test_1
+        type:   test
+        id:     1
+        body:   { foo: bar }
+
+  - do:
+      indices.refresh:
+        index: [test_1]
+
+---
+"sequence numbers are returned if requested from body":
+  - skip:
+      version: " - 6.99.99"
+      reason: sequence numbers were added in 7.0.0
+
+  - do:
+      search:
+        index: _all
+        body:
+          query:
+            match:
+              foo: bar
+          seq_no_primary_term: true
+
+  - match: {hits.total.value: 1}
+  - match: {hits.hits.0._seq_no: 1}
+  - gte: {hits.hits.0._primary_term: 1}
+
+---
+"sequence numbers are returned if requested from url":
+  - skip:
+      version: " - 6.99.99"
+      reason: sequence numbers were added in 7.0.0
+
+  - do:
+      search:
+        index: _all
+        body:
+          query:
+            match:
+              foo: bar
+        seq_no_primary_term: true
+
+  - match: {hits.total.value: 1}
+  - match: {hits.hits.0._seq_no: 1}
+  - gte: {hits.hits.0._primary_term: 1}
+
+---
+"sequence numbers are not returned if not requested":
+  - do:
+      search:
+        index: _all
+        body:
+          query:
+            match:
+              foo: bar
+
+  - is_false: hits.hits.0._seq_no
+  - is_false: hits.hits.0._primary_term

server/src/main/java/org/elasticsearch/action/search/ExpandSearchPhase.java

@@ -153,6 +153,7 @@ final class ExpandSearchPhase extends SearchPhase {
         groupSource.explain(options.isExplain());
         groupSource.trackScores(options.isTrackScores());
         groupSource.version(options.isVersion());
+        groupSource.seqNoAndPrimaryTerm(options.isSeqNoAndPrimaryTerm());
         if (innerCollapseBuilder != null) {
             groupSource.collapse(innerCollapseBuilder);
         }

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

@@ -68,6 +68,7 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
         PARSER.declareInt(InnerHitBuilder::setSize, SearchSourceBuilder.SIZE_FIELD);
         PARSER.declareBoolean(InnerHitBuilder::setExplain, SearchSourceBuilder.EXPLAIN_FIELD);
         PARSER.declareBoolean(InnerHitBuilder::setVersion, SearchSourceBuilder.VERSION_FIELD);
+        PARSER.declareBoolean(InnerHitBuilder::setSeqNoAndPrimaryTerm, SearchSourceBuilder.SEQ_NO_PRIMARY_TERM_FIELD);
         PARSER.declareBoolean(InnerHitBuilder::setTrackScores, SearchSourceBuilder.TRACK_SCORES_FIELD);
         PARSER.declareStringArray(InnerHitBuilder::setStoredFieldNames, SearchSourceBuilder.STORED_FIELDS_FIELD);
         PARSER.declareObjectArray(InnerHitBuilder::setDocValueFields,
@@ -117,7 +118,6 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
 
         }, COLLAPSE_FIELD, ObjectParser.ValueType.OBJECT);
     }
-
     private String name;
     private boolean ignoreUnmapped;
 
@@ -125,6 +125,7 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
     private int size = 3;
     private boolean explain;
     private boolean version;
+    private boolean seqNoAndPrimaryTerm;
     private boolean trackScores;
 
     private StoredFieldsContext storedFieldsContext;
@@ -155,6 +156,11 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
         size = in.readVInt();
         explain = in.readBoolean();
         version = in.readBoolean();
+        if (in.getVersion().onOrAfter(Version.V_7_0_0)){
+            seqNoAndPrimaryTerm = in.readBoolean();
+        } else {
+            seqNoAndPrimaryTerm = false;
+        }
         trackScores = in.readBoolean();
         storedFieldsContext = in.readOptionalWriteable(StoredFieldsContext::new);
         if (in.getVersion().before(Version.V_6_4_0)) {
@@ -199,6 +205,9 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
         out.writeVInt(size);
         out.writeBoolean(explain);
         out.writeBoolean(version);
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeBoolean(seqNoAndPrimaryTerm);
+        }
         out.writeBoolean(trackScores);
         out.writeOptionalWriteable(storedFieldsContext);
         if (out.getVersion().before(Version.V_6_4_0)) {
@@ -299,6 +308,15 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
         return this;
     }
 
+    public boolean isSeqNoAndPrimaryTerm() {
+        return seqNoAndPrimaryTerm;
+    }
+
+    public InnerHitBuilder setSeqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+        this.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
+        return this;
+    }
+
     public boolean isTrackScores() {
         return trackScores;
     }
@@ -436,6 +454,7 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
         builder.field(SearchSourceBuilder.FROM_FIELD.getPreferredName(), from);
         builder.field(SearchSourceBuilder.SIZE_FIELD.getPreferredName(), size);
         builder.field(SearchSourceBuilder.VERSION_FIELD.getPreferredName(), version);
+        builder.field(SearchSourceBuilder.SEQ_NO_PRIMARY_TERM_FIELD.getPreferredName(), seqNoAndPrimaryTerm);
         builder.field(SearchSourceBuilder.EXPLAIN_FIELD.getPreferredName(), explain);
         builder.field(SearchSourceBuilder.TRACK_SCORES_FIELD.getPreferredName(), trackScores);
         if (fetchSourceContext != null) {
@@ -494,6 +513,7 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
                 Objects.equals(size, that.size) &&
                 Objects.equals(explain, that.explain) &&
                 Objects.equals(version, that.version) &&
+                Objects.equals(seqNoAndPrimaryTerm, that.seqNoAndPrimaryTerm) &&
                 Objects.equals(trackScores, that.trackScores) &&
                 Objects.equals(storedFieldsContext, that.storedFieldsContext) &&
                 Objects.equals(docValueFields, that.docValueFields) &&
@@ -506,7 +526,7 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, ignoreUnmapped, from, size, explain, version, trackScores,
+        return Objects.hash(name, ignoreUnmapped, from, size, explain, version, seqNoAndPrimaryTerm, trackScores,
                 storedFieldsContext, docValueFields, scriptFields, fetchSourceContext, sorts, highlightBuilder, innerCollapseBuilder);
     }
 

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

@@ -78,6 +78,7 @@ public abstract class InnerHitContextBuilder {
         innerHitsContext.size(innerHitBuilder.getSize());
         innerHitsContext.explain(innerHitBuilder.isExplain());
         innerHitsContext.version(innerHitBuilder.isVersion());
+        innerHitsContext.seqNoAndPrimaryTerm(innerHitBuilder.isSeqNoAndPrimaryTerm());
         innerHitsContext.trackScores(innerHitBuilder.isTrackScores());
         if (innerHitBuilder.getStoredFieldsContext() != null) {
             innerHitsContext.storedFieldsContext(innerHitBuilder.getStoredFieldsContext());

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

@@ -368,6 +368,14 @@ public class NestedQueryBuilder extends AbstractQueryBuilder<NestedQueryBuilder>
             this.childObjectMapper = childObjectMapper;
         }
 
+        @Override
+        public void seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+            assert seqNoAndPrimaryTerm() == false;
+            if (seqNoAndPrimaryTerm) {
+                throw new UnsupportedOperationException("nested documents are not assigned sequence numbers");
+            }
+        }
+
         @Override
         public TopDocsAndMaxScore[] topDocs(SearchHit[] hits) throws IOException {
             Weight innerHitQueryWeight = createInnerHitQueryWeight();

server/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java

@@ -201,6 +201,9 @@ public class RestSearchAction extends BaseRestHandler {
         if (request.hasParam("version")) {
             searchSourceBuilder.version(request.paramAsBoolean("version", null));
         }
+        if (request.hasParam("seq_no_primary_term")) {
+            searchSourceBuilder.seqNoAndPrimaryTerm(request.paramAsBoolean("seq_no_primary_term", null));
+        }
         if (request.hasParam("timeout")) {
             searchSourceBuilder.timeout(request.paramAsTime("timeout", null));
         }

server/src/main/java/org/elasticsearch/search/DefaultSearchContext.java

@@ -107,6 +107,7 @@ final class DefaultSearchContext extends SearchContext {
     private ScrollContext scrollContext;
     private boolean explain;
     private boolean version = false; // by default, we don't return versions
+    private boolean seqAndPrimaryTerm = false;
     private StoredFieldsContext storedFields;
     private ScriptFieldsContext scriptFields;
     private FetchSourceContext fetchSourceContext;
@@ -719,6 +720,16 @@ final class DefaultSearchContext extends SearchContext {
         this.version = version;
     }
 
+    @Override
+    public boolean seqNoAndPrimaryTerm() {
+        return seqAndPrimaryTerm;
+    }
+
+    @Override
+    public void seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+        this.seqAndPrimaryTerm = seqNoAndPrimaryTerm;
+    }
+
     @Override
     public int[] docIdsToLoad() {
         return docIdsToLoad;

server/src/main/java/org/elasticsearch/search/SearchHit.java

@@ -21,6 +21,7 @@ package org.elasticsearch.search;
 
 import org.apache.lucene.search.Explanation;
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
@@ -46,6 +47,7 @@ import org.elasticsearch.common.xcontent.XContentParser.Token;
 import org.elasticsearch.index.mapper.IgnoredFieldMapper;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
 import org.elasticsearch.search.lookup.SourceLookup;
@@ -91,6 +93,8 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
     private NestedIdentity nestedIdentity;
 
     private long version = -1;
+    private long seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
+    private long primaryTerm = SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
 
     private BytesReference source;
 
@@ -168,6 +172,30 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         return this.version;
     }
 
+
+    public void setSeqNo(long seqNo) {
+        this.seqNo = seqNo;
+    }
+
+    public void setPrimaryTerm(long primaryTerm) {
+        this.primaryTerm = primaryTerm;
+    }
+
+    /**
+     * returns the sequence number of the last modification to the document, or {@link SequenceNumbers#UNASSIGNED_SEQ_NO}
+     * if not requested.
+     **/
+    public long getSeqNo() {
+        return this.seqNo;
+    }
+
+    /**
+     * returns the primary term of the last modification to the document, or {@link SequenceNumbers#UNASSIGNED_PRIMARY_TERM}
+     * if not requested. */
+    public long getPrimaryTerm() {
+        return this.primaryTerm;
+    }
+
     /**
      * The index of the hit.
      */
@@ -393,6 +421,8 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         static final String _TYPE = "_type";
         static final String _ID = "_id";
         static final String _VERSION = "_version";
+        static final String _SEQ_NO = "_seq_no";
+        static final String _PRIMARY_TERM = "_primary_term";
         static final String _SCORE = "_score";
         static final String FIELDS = "fields";
         static final String HIGHLIGHT = "highlight";
@@ -453,6 +483,12 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         if (version != -1) {
             builder.field(Fields._VERSION, version);
         }
+
+        if (seqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {
+            builder.field(Fields._SEQ_NO, seqNo);
+            builder.field(Fields._PRIMARY_TERM, primaryTerm);
+        }
+
         if (Float.isNaN(score)) {
             builder.nullField(Fields._SCORE);
         } else {
@@ -537,6 +573,8 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         parser.declareField((map, value) -> map.put(Fields._SCORE, value), SearchHit::parseScore, new ParseField(Fields._SCORE),
                 ValueType.FLOAT_OR_NULL);
         parser.declareLong((map, value) -> map.put(Fields._VERSION, value), new ParseField(Fields._VERSION));
+        parser.declareLong((map, value) -> map.put(Fields._SEQ_NO, value), new ParseField(Fields._SEQ_NO));
+        parser.declareLong((map, value) -> map.put(Fields._PRIMARY_TERM, value), new ParseField(Fields._PRIMARY_TERM));
         parser.declareField((map, value) -> map.put(Fields._SHARD, value), (p, c) -> ShardId.fromString(p.text()),
                 new ParseField(Fields._SHARD), ValueType.STRING);
         parser.declareObject((map, value) -> map.put(SourceFieldMapper.NAME, value), (p, c) -> parseSourceBytes(p),
@@ -588,6 +626,8 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         }
         searchHit.score(get(Fields._SCORE, values, DEFAULT_SCORE));
         searchHit.version(get(Fields._VERSION, values, -1L));
+        searchHit.setSeqNo(get(Fields._SEQ_NO, values, SequenceNumbers.UNASSIGNED_SEQ_NO));
+        searchHit.setPrimaryTerm(get(Fields._PRIMARY_TERM, values, SequenceNumbers.UNASSIGNED_PRIMARY_TERM));
         searchHit.sortValues(get(Fields.SORT, values, SearchSortValues.EMPTY));
         searchHit.highlightFields(get(Fields.HIGHLIGHT, values, null));
         searchHit.sourceRef(get(SourceFieldMapper.NAME, values, null));
@@ -744,6 +784,10 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         type = in.readOptionalText();
         nestedIdentity = in.readOptionalWriteable(NestedIdentity::new);
         version = in.readLong();
+        if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
+            seqNo = in.readZLong();
+            primaryTerm = in.readVLong();
+        }
         source = in.readBytesReference();
         if (source.length() == 0) {
             source = null;
@@ -812,6 +856,10 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
         out.writeOptionalText(type);
         out.writeOptionalWriteable(nestedIdentity);
         out.writeLong(version);
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeZLong(seqNo);
+            out.writeVLong(primaryTerm);
+        }
         out.writeBytesReference(source);
         if (explanation == null) {
             out.writeBoolean(false);
@@ -867,6 +915,8 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
                 && Objects.equals(type, other.type)
                 && Objects.equals(nestedIdentity, other.nestedIdentity)
                 && Objects.equals(version, other.version)
+                && Objects.equals(seqNo, other.seqNo)
+                && Objects.equals(primaryTerm, other.primaryTerm)
                 && Objects.equals(source, other.source)
                 && Objects.equals(fields, other.fields)
                 && Objects.equals(getHighlightFields(), other.getHighlightFields())
@@ -880,8 +930,8 @@ public final class SearchHit implements Streamable, ToXContentObject, Iterable<D
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, type, nestedIdentity, version, source, fields, getHighlightFields(), Arrays.hashCode(matchedQueries),
-                explanation, shard, innerHits, index, clusterAlias);
+        return Objects.hash(id, type, nestedIdentity, version, seqNo, primaryTerm, source, fields, getHighlightFields(),
+            Arrays.hashCode(matchedQueries), explanation, shard, innerHits, index, clusterAlias);
     }
 
     /**

server/src/main/java/org/elasticsearch/search/SearchModule.java

@@ -240,6 +240,7 @@ import org.elasticsearch.search.fetch.subphase.FetchSourceSubPhase;
 import org.elasticsearch.search.fetch.subphase.MatchedQueriesFetchSubPhase;
 import org.elasticsearch.search.fetch.subphase.ScoreFetchSubPhase;
 import org.elasticsearch.search.fetch.subphase.ScriptFieldsFetchSubPhase;
+import org.elasticsearch.search.fetch.subphase.SeqNoPrimaryTermFetchSubPhase;
 import org.elasticsearch.search.fetch.subphase.VersionFetchSubPhase;
 import org.elasticsearch.search.fetch.subphase.highlight.FastVectorHighlighter;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightPhase;
@@ -727,6 +728,7 @@ public class SearchModule {
         registerFetchSubPhase(new ScriptFieldsFetchSubPhase());
         registerFetchSubPhase(new FetchSourceSubPhase());
         registerFetchSubPhase(new VersionFetchSubPhase());
+        registerFetchSubPhase(new SeqNoPrimaryTermFetchSubPhase());
         registerFetchSubPhase(new MatchedQueriesFetchSubPhase());
         registerFetchSubPhase(new HighlightPhase(highlighters));
         registerFetchSubPhase(new ScoreFetchSubPhase());

server/src/main/java/org/elasticsearch/search/SearchService.java

@@ -901,6 +901,11 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
         if (source.version() != null) {
             context.version(source.version());
         }
+
+        if (source.seqNoAndPrimaryTerm() != null) {
+            context.seqNoAndPrimaryTerm(source.seqNoAndPrimaryTerm());
+        }
+
         if (source.stats() != null) {
             context.groupStats(source.stats());
         }

server/src/main/java/org/elasticsearch/search/aggregations/metrics/TopHitsAggregationBuilder.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.search.aggregations.metrics;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.Strings;
@@ -66,6 +67,7 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
     private int size = 3;
     private boolean explain = false;
     private boolean version = false;
+    private boolean seqNoAndPrimaryTerm = false;
     private boolean trackScores = false;
     private List<SortBuilder<?>> sorts = null;
     private HighlightBuilder highlightBuilder;
@@ -85,6 +87,7 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
         this.size = clone.size;
         this.explain = clone.explain;
         this.version = clone.version;
+        this.seqNoAndPrimaryTerm = clone.seqNoAndPrimaryTerm;
         this.trackScores = clone.trackScores;
         this.sorts = clone.sorts == null ? null : new ArrayList<>(clone.sorts);
         this.highlightBuilder = clone.highlightBuilder == null ? null :
@@ -137,6 +140,9 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
         }
         trackScores = in.readBoolean();
         version = in.readBoolean();
+        if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
+            seqNoAndPrimaryTerm = in.readBoolean();
+        }
     }
 
     @Override
@@ -173,6 +179,9 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
         }
         out.writeBoolean(trackScores);
         out.writeBoolean(version);
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeBoolean(seqNoAndPrimaryTerm);
+        }
     }
 
     /**
@@ -526,6 +535,23 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
         return version;
     }
 
+    /**
+     * Should each {@link org.elasticsearch.search.SearchHit} be returned with the
+     * sequence number and primary term of the last modification of the document.
+     */
+    public TopHitsAggregationBuilder seqNoAndPrimaryTerm(Boolean seqNoAndPrimaryTerm) {
+        this.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
+        return this;
+    }
+
+    /**
+     * Indicates whether {@link org.elasticsearch.search.SearchHit}s should be returned with the
+     * sequence number and primary term of the last modification of the document.
+     */
+    public Boolean seqNoAndPrimaryTerm() {
+        return seqNoAndPrimaryTerm;
+    }
+
     /**
      * Applies when sorting, and controls if scores will be tracked as well.
      * Defaults to {@code false}.
@@ -579,8 +605,9 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
         } else {
             optionalSort = SortBuilder.buildSort(sorts, context.getQueryShardContext());
         }
-        return new TopHitsAggregatorFactory(name, from, size, explain, version, trackScores, optionalSort, highlightBuilder,
-                storedFieldsContext, docValueFields, fields, fetchSourceContext, context, parent, subfactoriesBuilder, metaData);
+        return new TopHitsAggregatorFactory(name, from, size, explain, version, seqNoAndPrimaryTerm, trackScores, optionalSort,
+            highlightBuilder, storedFieldsContext, docValueFields, fields, fetchSourceContext, context, parent, subfactoriesBuilder,
+            metaData);
     }
 
     @Override
@@ -589,6 +616,7 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
         builder.field(SearchSourceBuilder.FROM_FIELD.getPreferredName(), from);
         builder.field(SearchSourceBuilder.SIZE_FIELD.getPreferredName(), size);
         builder.field(SearchSourceBuilder.VERSION_FIELD.getPreferredName(), version);
+        builder.field(SearchSourceBuilder.SEQ_NO_PRIMARY_TERM_FIELD.getPreferredName(), seqNoAndPrimaryTerm);
         builder.field(SearchSourceBuilder.EXPLAIN_FIELD.getPreferredName(), explain);
         if (fetchSourceContext != null) {
             builder.field(SearchSourceBuilder._SOURCE_FIELD.getPreferredName(), fetchSourceContext);
@@ -646,6 +674,8 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
                     factory.size(parser.intValue());
                 } else if (SearchSourceBuilder.VERSION_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     factory.version(parser.booleanValue());
+                } else if (SearchSourceBuilder.SEQ_NO_PRIMARY_TERM_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
+                    factory.seqNoAndPrimaryTerm(parser.booleanValue());
                 } else if (SearchSourceBuilder.EXPLAIN_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     factory.explain(parser.booleanValue());
                 } else if (SearchSourceBuilder.TRACK_SCORES_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
@@ -745,7 +775,7 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
     @Override
     protected int doHashCode() {
         return Objects.hash(explain, fetchSourceContext, docValueFields, storedFieldsContext, from, highlightBuilder,
-            scriptFields, size, sorts, trackScores, version);
+            scriptFields, size, sorts, trackScores, version, seqNoAndPrimaryTerm);
     }
 
     @Override
@@ -761,7 +791,8 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
                 && Objects.equals(size, other.size)
                 && Objects.equals(sorts, other.sorts)
                 && Objects.equals(trackScores, other.trackScores)
-                && Objects.equals(version, other.version);
+                && Objects.equals(version, other.version)
+                && Objects.equals(seqNoAndPrimaryTerm, other.seqNoAndPrimaryTerm);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/metrics/TopHitsAggregatorFactory.java

@@ -44,6 +44,7 @@ class TopHitsAggregatorFactory extends AggregatorFactory<TopHitsAggregatorFactor
     private final int size;
     private final boolean explain;
     private final boolean version;
+    private final boolean seqNoAndPrimaryTerm;
     private final boolean trackScores;
     private final Optional<SortAndFormats> sort;
     private final HighlightBuilder highlightBuilder;
@@ -52,8 +53,8 @@ class TopHitsAggregatorFactory extends AggregatorFactory<TopHitsAggregatorFactor
     private final List<ScriptFieldsContext.ScriptField> scriptFields;
     private final FetchSourceContext fetchSourceContext;
 
-    TopHitsAggregatorFactory(String name, int from, int size, boolean explain, boolean version, boolean trackScores,
-            Optional<SortAndFormats> sort, HighlightBuilder highlightBuilder, StoredFieldsContext storedFieldsContext,
+    TopHitsAggregatorFactory(String name, int from, int size, boolean explain, boolean version, boolean seqNoAndPrimaryTerm,
+            boolean trackScores, Optional<SortAndFormats> sort, HighlightBuilder highlightBuilder, StoredFieldsContext storedFieldsContext,
             List<FieldAndFormat> docValueFields, List<ScriptFieldsContext.ScriptField> scriptFields, FetchSourceContext fetchSourceContext,
             SearchContext context, AggregatorFactory<?> parent, AggregatorFactories.Builder subFactories, Map<String, Object> metaData)
             throws IOException {
@@ -62,6 +63,7 @@ class TopHitsAggregatorFactory extends AggregatorFactory<TopHitsAggregatorFactor
         this.size = size;
         this.explain = explain;
         this.version = version;
+        this.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
         this.trackScores = trackScores;
         this.sort = sort;
         this.highlightBuilder = highlightBuilder;
@@ -78,6 +80,7 @@ class TopHitsAggregatorFactory extends AggregatorFactory<TopHitsAggregatorFactor
         subSearchContext.parsedQuery(context.parsedQuery());
         subSearchContext.explain(explain);
         subSearchContext.version(version);
+        subSearchContext.seqNoAndPrimaryTerm(seqNoAndPrimaryTerm);
         subSearchContext.trackScores(trackScores);
         subSearchContext.from(from);
         subSearchContext.size(size);

server/src/main/java/org/elasticsearch/search/builder/SearchSourceBuilder.java

@@ -92,6 +92,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
     public static final ParseField POST_FILTER_FIELD = new ParseField("post_filter");
     public static final ParseField MIN_SCORE_FIELD = new ParseField("min_score");
     public static final ParseField VERSION_FIELD = new ParseField("version");
+    public static final ParseField SEQ_NO_PRIMARY_TERM_FIELD = new ParseField("seq_no_primary_term");
     public static final ParseField EXPLAIN_FIELD = new ParseField("explain");
     public static final ParseField _SOURCE_FIELD = new ParseField("_source");
     public static final ParseField STORED_FIELDS_FIELD = new ParseField("stored_fields");
@@ -151,6 +152,8 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
 
     private Boolean version;
 
+    private Boolean seqNoAndPrimaryTerm;
+
     private List<SortBuilder<?>> sorts;
 
     private boolean trackScores = false;
@@ -247,6 +250,11 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         timeout = in.readOptionalTimeValue();
         trackScores = in.readBoolean();
         version = in.readOptionalBoolean();
+        if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
+            seqNoAndPrimaryTerm = in.readOptionalBoolean();
+        } else {
+            seqNoAndPrimaryTerm = null;
+        }
         extBuilders = in.readNamedWriteableList(SearchExtBuilder.class);
         profile = in.readBoolean();
         searchAfterBuilder = in.readOptionalWriteable(SearchAfterBuilder::new);
@@ -310,6 +318,9 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         out.writeOptionalTimeValue(timeout);
         out.writeBoolean(trackScores);
         out.writeOptionalBoolean(version);
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeOptionalBoolean(seqNoAndPrimaryTerm);
+        }
         out.writeNamedWriteableList(extBuilders);
         out.writeBoolean(profile);
         out.writeOptionalWriteable(searchAfterBuilder);
@@ -441,6 +452,23 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         return version;
     }
 
+    /**
+     * Should each {@link org.elasticsearch.search.SearchHit} be returned with the
+     * sequence number and primary term of the last modification of the document.
+     */
+    public SearchSourceBuilder seqNoAndPrimaryTerm(Boolean seqNoAndPrimaryTerm) {
+        this.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
+        return this;
+    }
+
+    /**
+     * Indicates whether {@link org.elasticsearch.search.SearchHit}s should be returned with the
+     * sequence number and primary term of the last modification of the document.
+     */
+    public Boolean seqNoAndPrimaryTerm() {
+        return seqNoAndPrimaryTerm;
+    }
+
     /**
      * An optional timeout to control how long search is allowed to take.
      */
@@ -999,6 +1027,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         rewrittenBuilder.trackScores = trackScores;
         rewrittenBuilder.trackTotalHitsUpTo = trackTotalHitsUpTo;
         rewrittenBuilder.version = version;
+        rewrittenBuilder.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
         rewrittenBuilder.collapse = collapse;
         return rewrittenBuilder;
     }
@@ -1038,6 +1067,8 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
                     minScore = parser.floatValue();
                 } else if (VERSION_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     version = parser.booleanValue();
+                } else if (SEQ_NO_PRIMARY_TERM_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
+                    seqNoAndPrimaryTerm = parser.booleanValue();
                 } else if (EXPLAIN_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     explain = parser.booleanValue();
                 } else if (TRACK_SCORES_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
@@ -1205,6 +1236,10 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
             builder.field(VERSION_FIELD.getPreferredName(), version);
         }
 
+        if (seqNoAndPrimaryTerm != null) {
+            builder.field(SEQ_NO_PRIMARY_TERM_FIELD.getPreferredName(), seqNoAndPrimaryTerm);
+        }
+
         if (explain != null) {
             builder.field(EXPLAIN_FIELD.getPreferredName(), explain);
         }
@@ -1523,7 +1558,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         return Objects.hash(aggregations, explain, fetchSourceContext, docValueFields, storedFieldsContext, from, highlightBuilder,
                 indexBoosts, minScore, postQueryBuilder, queryBuilder, rescoreBuilders, scriptFields, size,
                 sorts, searchAfterBuilder, sliceBuilder, stats, suggestBuilder, terminateAfter, timeout, trackScores, version,
-                profile, extBuilders, collapse, trackTotalHitsUpTo);
+                seqNoAndPrimaryTerm, profile, extBuilders, collapse, trackTotalHitsUpTo);
     }
 
     @Override
@@ -1558,6 +1593,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
                 && Objects.equals(timeout, other.timeout)
                 && Objects.equals(trackScores, other.trackScores)
                 && Objects.equals(version, other.version)
+                && Objects.equals(seqNoAndPrimaryTerm, other.seqNoAndPrimaryTerm)
                 && Objects.equals(profile, other.profile)
                 && Objects.equals(extBuilders, other.extBuilders)
                 && Objects.equals(collapse, other.collapse)

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

@@ -0,0 +1,69 @@
+/*
+ * 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.search.fetch.subphase;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.ReaderUtil;
+import org.elasticsearch.index.mapper.SeqNoFieldMapper;
+import org.elasticsearch.index.seqno.SequenceNumbers;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.fetch.FetchSubPhase;
+import org.elasticsearch.search.internal.SearchContext;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Comparator;
+
+public final class SeqNoPrimaryTermFetchSubPhase implements FetchSubPhase {
+    @Override
+    public void hitsExecute(SearchContext context, SearchHit[] hits) throws IOException {
+        if (context.seqNoAndPrimaryTerm() == false) {
+            return;
+        }
+
+        hits = hits.clone(); // don't modify the incoming hits
+        Arrays.sort(hits, Comparator.comparingInt(SearchHit::docId));
+
+        int lastReaderId = -1;
+        NumericDocValues seqNoField = null;
+        NumericDocValues primaryTermField = null;
+        for (SearchHit hit : hits) {
+            int readerId = ReaderUtil.subIndex(hit.docId(), context.searcher().getIndexReader().leaves());
+            LeafReaderContext subReaderContext = context.searcher().getIndexReader().leaves().get(readerId);
+            if (lastReaderId != readerId) {
+                seqNoField = subReaderContext.reader().getNumericDocValues(SeqNoFieldMapper.NAME);
+                primaryTermField = subReaderContext.reader().getNumericDocValues(SeqNoFieldMapper.PRIMARY_TERM_NAME);
+                lastReaderId = readerId;
+            }
+            int docId = hit.docId() - subReaderContext.docBase;
+            long seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
+            long primaryTerm = SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
+            // we have to check the primary term field as it is only assigned for non-nested documents
+            if (primaryTermField != null && primaryTermField.advanceExact(docId)) {
+                boolean found = seqNoField.advanceExact(docId);
+                assert found: "found seq no for " + docId + " but not a primary term";
+                seqNo = seqNoField.longValue();
+                primaryTerm = primaryTermField.longValue();
+            }
+            hit.setSeqNo(seqNo);
+            hit.setPrimaryTerm(primaryTerm);
+        }
+    }
+}

server/src/main/java/org/elasticsearch/search/internal/FilteredSearchContext.java

@@ -422,6 +422,16 @@ public abstract class FilteredSearchContext extends SearchContext {
         in.version(version);
     }
 
+    @Override
+    public boolean seqNoAndPrimaryTerm() {
+        return in.seqNoAndPrimaryTerm();
+    }
+
+    @Override
+    public void seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+        in.seqNoAndPrimaryTerm(seqNoAndPrimaryTerm);
+    }
+
     @Override
     public int[] docIdsToLoad() {
         return in.docIdsToLoad();

server/src/main/java/org/elasticsearch/search/internal/SearchContext.java

@@ -38,7 +38,6 @@ import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ObjectMapper;
-import org.elasticsearch.search.collapse.CollapseContext;
 import org.elasticsearch.index.query.ParsedQuery;
 import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.index.shard.IndexShard;
@@ -46,6 +45,7 @@ import org.elasticsearch.index.similarity.SimilarityService;
 import org.elasticsearch.search.SearchExtBuilder;
 import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.search.aggregations.SearchContextAggregations;
+import org.elasticsearch.search.collapse.CollapseContext;
 import org.elasticsearch.search.dfs.DfsSearchResult;
 import org.elasticsearch.search.fetch.FetchPhase;
 import org.elasticsearch.search.fetch.FetchSearchResult;
@@ -309,6 +309,12 @@ public abstract class SearchContext extends AbstractRefCounted implements Releas
 
     public abstract void version(boolean version);
 
+    /** indicates whether the sequence number and primary term of the last modification to each hit should be returned */
+    public abstract boolean seqNoAndPrimaryTerm();
+
+    /** controls whether the sequence number and primary term of the last modification to each hit should be returned */
+    public abstract void seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm);
+
     public abstract int[] docIdsToLoad();
 
     public abstract int docIdsToLoadFrom();

server/src/main/java/org/elasticsearch/search/internal/SubSearchContext.java

@@ -65,6 +65,7 @@ public class SubSearchContext extends FilteredSearchContext {
     private boolean explain;
     private boolean trackScores;
     private boolean version;
+    private boolean seqNoAndPrimaryTerm;
 
     public SubSearchContext(SearchContext context) {
         super(context);
@@ -294,6 +295,16 @@ public class SubSearchContext extends FilteredSearchContext {
         this.version = version;
     }
 
+    @Override
+    public boolean seqNoAndPrimaryTerm() {
+        return seqNoAndPrimaryTerm;
+    }
+
+    @Override
+    public void seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+        this.seqNoAndPrimaryTerm = seqNoAndPrimaryTerm;
+    }
+
     @Override
     public int[] docIdsToLoad() {
         return docIdsToLoad;

server/src/test/java/org/elasticsearch/action/search/ExpandSearchPhaseTests.java

@@ -241,6 +241,7 @@ public class ExpandSearchPhaseTests extends ESTestCase {
     public void testExpandRequestOptions() throws IOException {
         MockSearchPhaseContext mockSearchPhaseContext = new MockSearchPhaseContext(1);
         boolean version = randomBoolean();
+        final boolean seqNoAndTerm = randomBoolean();
 
         mockSearchPhaseContext.searchTransport = new SearchTransportService(null, null) {
             @Override
@@ -249,13 +250,14 @@ public class ExpandSearchPhaseTests extends ESTestCase {
                 assertTrue(request.requests().stream().allMatch((r) -> "foo".equals(r.preference())));
                 assertTrue(request.requests().stream().allMatch((r) -> "baz".equals(r.routing())));
                 assertTrue(request.requests().stream().allMatch((r) -> version == r.source().version()));
+                assertTrue(request.requests().stream().allMatch((r) -> seqNoAndTerm == r.source().seqNoAndPrimaryTerm()));
                 assertTrue(request.requests().stream().allMatch((r) -> postFilter.equals(r.source().postFilter())));
             }
         };
         mockSearchPhaseContext.getRequest().source(new SearchSourceBuilder()
             .collapse(
                 new CollapseBuilder("someField")
-                    .setInnerHits(new InnerHitBuilder().setName("foobarbaz").setVersion(version))
+                    .setInnerHits(new InnerHitBuilder().setName("foobarbaz").setVersion(version).setSeqNoAndPrimaryTerm(seqNoAndTerm))
             )
             .postFilter(QueryBuilders.existsQuery("foo")))
             .preference("foobar")

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

@@ -140,6 +140,11 @@ public class InnerHitBuilderTests extends ESTestCase {
         }
     }
 
+    public static InnerHitBuilder randomNestedInnerHits() {
+        InnerHitBuilder innerHitBuilder = randomInnerHits();
+        innerHitBuilder.setSeqNoAndPrimaryTerm(false); // not supported by nested queries
+        return innerHitBuilder;
+    }
     public static InnerHitBuilder randomInnerHits() {
         InnerHitBuilder innerHits = new InnerHitBuilder();
         innerHits.setName(randomAlphaOfLengthBetween(1, 16));
@@ -147,6 +152,7 @@ public class InnerHitBuilderTests extends ESTestCase {
         innerHits.setSize(randomIntBetween(0, 32));
         innerHits.setExplain(randomBoolean());
         innerHits.setVersion(randomBoolean());
+        innerHits.setSeqNoAndPrimaryTerm(randomBoolean());
         innerHits.setTrackScores(randomBoolean());
         if (randomBoolean()) {
             innerHits.setStoredFieldNames(randomListStuff(16, () -> randomAlphaOfLengthBetween(1, 16)));
@@ -189,6 +195,7 @@ public class InnerHitBuilderTests extends ESTestCase {
         modifiers.add(() -> copy.setSize(randomValueOtherThan(copy.getSize(), () -> randomIntBetween(0, 128))));
         modifiers.add(() -> copy.setExplain(!copy.isExplain()));
         modifiers.add(() -> copy.setVersion(!copy.isVersion()));
+        modifiers.add(() -> copy.setSeqNoAndPrimaryTerm(!copy.isSeqNoAndPrimaryTerm()));
         modifiers.add(() -> copy.setTrackScores(!copy.isTrackScores()));
         modifiers.add(() -> copy.setName(randomValueOtherThan(copy.getName(), () -> randomAlphaOfLengthBetween(1, 16))));
         modifiers.add(() -> {

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

@@ -45,7 +45,7 @@ import java.util.HashMap;
 import java.util.Map;
 
 import static org.elasticsearch.index.IndexSettingsTests.newIndexMeta;
-import static org.elasticsearch.index.query.InnerHitBuilderTests.randomInnerHits;
+import static org.elasticsearch.index.query.InnerHitBuilderTests.randomNestedInnerHits;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
@@ -267,7 +267,7 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
     }
 
     public void testInlineLeafInnerHitsNestedQuery() throws Exception {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
+        InnerHitBuilder leafInnerHits = randomNestedInnerHits();
         NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
         nestedQueryBuilder.innerHit(leafInnerHits);
         Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();
@@ -276,7 +276,7 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
     }
 
     public void testInlineLeafInnerHitsNestedQueryViaBoolQuery() {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
+        InnerHitBuilder leafInnerHits = randomNestedInnerHits();
         NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
             .innerHit(leafInnerHits);
         BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder().should(nestedQueryBuilder);
@@ -286,7 +286,7 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
     }
 
     public void testInlineLeafInnerHitsNestedQueryViaConstantScoreQuery() {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
+        InnerHitBuilder leafInnerHits = randomNestedInnerHits();
         NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
             .innerHit(leafInnerHits);
         ConstantScoreQueryBuilder constantScoreQueryBuilder = new ConstantScoreQueryBuilder(nestedQueryBuilder);
@@ -296,10 +296,10 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
     }
 
     public void testInlineLeafInnerHitsNestedQueryViaBoostingQuery() {
-        InnerHitBuilder leafInnerHits1 = randomInnerHits();
+        InnerHitBuilder leafInnerHits1 = randomNestedInnerHits();
         NestedQueryBuilder nestedQueryBuilder1 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
             .innerHit(leafInnerHits1);
-        InnerHitBuilder leafInnerHits2 = randomInnerHits();
+        InnerHitBuilder leafInnerHits2 = randomNestedInnerHits();
         NestedQueryBuilder nestedQueryBuilder2 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
             .innerHit(leafInnerHits2);
         BoostingQueryBuilder constantScoreQueryBuilder = new BoostingQueryBuilder(nestedQueryBuilder1, nestedQueryBuilder2);
@@ -310,7 +310,7 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
     }
 
     public void testInlineLeafInnerHitsNestedQueryViaFunctionScoreQuery() {
-        InnerHitBuilder leafInnerHits = randomInnerHits();
+        InnerHitBuilder leafInnerHits = randomNestedInnerHits();
         NestedQueryBuilder nestedQueryBuilder = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None)
             .innerHit(leafInnerHits);
         FunctionScoreQueryBuilder functionScoreQueryBuilder = new FunctionScoreQueryBuilder(nestedQueryBuilder);
@@ -330,7 +330,7 @@ public class NestedQueryBuilderTests extends AbstractQueryTestCase<NestedQueryBu
         when(mapperService.getIndexSettings()).thenReturn(settings);
         when(searchContext.mapperService()).thenReturn(mapperService);
 
-        InnerHitBuilder leafInnerHits = randomInnerHits();
+        InnerHitBuilder leafInnerHits = randomNestedInnerHits();
         NestedQueryBuilder query1 = new NestedQueryBuilder("path", new MatchAllQueryBuilder(), ScoreMode.None);
         query1.innerHit(leafInnerHits);
         final Map<String, InnerHitContextBuilder> innerHitBuilders = new HashMap<>();

server/src/test/java/org/elasticsearch/search/SearchHitTests.java

@@ -90,6 +90,11 @@ public class SearchHitTests extends AbstractStreamableTestCase<SearchHit> {
         if (randomBoolean()) {
             hit.version(randomLong());
         }
+
+        if (randomBoolean()) {
+            hit.version(randomNonNegativeLong());
+            hit.version(randomLongBetween(1, Long.MAX_VALUE));
+        }
         if (randomBoolean()) {
             hit.sortValues(SearchSortValuesTests.createTestItem(xContentType, transportSerialization));
         }

server/src/test/java/org/elasticsearch/search/aggregations/metrics/TopHitsIT.java

@@ -31,6 +31,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.query.MatchAllQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.script.MockScriptEngine;
 import org.elasticsearch.script.MockScriptPlugin;
@@ -83,6 +84,7 @@ import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
@@ -578,6 +580,7 @@ public class TopHitsIT extends ESIntegTestCase {
     }
 
     public void testFetchFeatures() {
+        final boolean seqNoAndTerm = randomBoolean();
         SearchResponse response = client().prepareSearch("idx")
                 .setQuery(matchQuery("text", "text").queryName("test"))
                 .addAggregation(terms("terms")
@@ -593,6 +596,7 @@ public class TopHitsIT extends ESIntegTestCase {
                                                 new Script(ScriptType.INLINE, MockScriptEngine.NAME, "5", Collections.emptyMap()))
                                             .fetchSource("text", null)
                                             .version(true)
+                                            .seqNoAndPrimaryTerm(seqNoAndTerm)
                                 )
                 )
                 .get();
@@ -620,6 +624,14 @@ public class TopHitsIT extends ESIntegTestCase {
             long version = hit.getVersion();
             assertThat(version, equalTo(1L));
 
+            if (seqNoAndTerm) {
+                assertThat(hit.getSeqNo(), greaterThanOrEqualTo(0L));
+                assertThat(hit.getPrimaryTerm(), greaterThanOrEqualTo(1L));
+            } else {
+                assertThat(hit.getSeqNo(), equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
+                assertThat(hit.getPrimaryTerm(), equalTo(SequenceNumbers.UNASSIGNED_PRIMARY_TERM));
+            }
+
             assertThat(hit.getMatchedQueries()[0], equalTo("test"));
 
             DocumentField field = hit.field("field1");

server/src/test/java/org/elasticsearch/search/aggregations/metrics/TopHitsTests.java

@@ -54,6 +54,9 @@ public class TopHitsTests extends BaseAggregationTestCase<TopHitsAggregationBuil
         if (randomBoolean()) {
             factory.version(randomBoolean());
         }
+        if (randomBoolean()) {
+            factory.seqNoAndPrimaryTerm(randomBoolean());
+        }
         if (randomBoolean()) {
             factory.trackScores(randomBoolean());
         }

test/framework/src/main/java/org/elasticsearch/search/RandomSearchRequestGenerator.java

@@ -145,6 +145,9 @@ public class RandomSearchRequestGenerator {
         if (randomBoolean()) {
             builder.version(randomBoolean());
         }
+        if (randomBoolean()) {
+            builder.seqNoAndPrimaryTerm(randomBoolean());
+        }
         if (randomBoolean()) {
             builder.trackScores(randomBoolean());
         }

test/framework/src/main/java/org/elasticsearch/test/TestSearchContext.java

@@ -504,6 +504,16 @@ public class TestSearchContext extends SearchContext {
     public void version(boolean version) {
     }
 
+    @Override
+    public boolean seqNoAndPrimaryTerm() {
+        return false;
+    }
+
+    @Override
+    public void seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+
+    }
+
     @Override
     public int[] docIdsToLoad() {
         return new int[0];