Fix term stats when talking to ES 6

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/30_dfs_query.yml

@@ -0,0 +1,14 @@
+---
+"Perform a dfs_query_then_fetch search on a keyword field":
+  - do:
+      search:
+        search_type: dfs_query_then_fetch
+        index: keyword_index
+        rest_total_hits_as_int: true
+        body:
+          query:
+            match:
+              field:
+                query: value
+
+  - match: { hits.total: 3 }

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/old_cluster/30_dfs_query.yml

@@ -0,0 +1,43 @@
+---
+"Perform a dfs_query_then_fetch search on a keyword field":
+  - do:
+      indices.create:
+        index: keyword_index
+        body:
+          mappings:
+            properties:
+              field:
+                type: keyword
+  - do:
+      index:
+        index: keyword_index
+        body:
+          field: value
+        refresh: true
+
+  - do:
+      index:
+        index: keyword_index
+        body:
+          field: value
+        refresh: true
+
+  - do:
+      index:
+        index: keyword_index
+        body:
+          field: value
+        refresh: true
+
+  - do:
+      search:
+        search_type: dfs_query_then_fetch
+        index: keyword_index
+        rest_total_hits_as_int: true
+        body:
+          query:
+            match:
+              field:
+                query: value
+
+  - match: { hits.total: 3 }

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/upgraded_cluster/30_dfs_query.yml

@@ -0,0 +1,14 @@
+---
+"Perform a dfs_query_then_fetch search on a keyword field":
+  - do:
+      search:
+        search_type: dfs_query_then_fetch
+        index: keyword_index
+        rest_total_hits_as_int: true
+        body:
+          query:
+            match:
+              field:
+                query: value
+
+  - match: { hits.total: 3 }

server/src/main/java/org/elasticsearch/search/dfs/AggregatedDfs.java

@@ -14,6 +14,8 @@
 import org.apache.lucene.index.Term;
 import org.apache.lucene.search.CollectionStatistics;
 import org.apache.lucene.search.TermStatistics;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.collect.HppcMaps;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -32,9 +34,17 @@ public AggregatedDfs(StreamInput in) throws IOException {
         termStatistics = HppcMaps.newMap(size);
         for (int i = 0; i < size; i++) {
             Term term = new Term(in.readString(), in.readBytesRef());
-            TermStatistics stats = new TermStatistics(in.readBytesRef(),
-                in.readVLong(),
-                DfsSearchResult.subOne(in.readVLong()));
+            BytesRef term2 = in.readBytesRef();
+            final long docFreq = in.readVLong();
+            assert docFreq >= 0;
+            long totalTermFreq = DfsSearchResult.subOne(in.readVLong());
+            if (in.getVersion().before(Version.V_7_0_0)) {
+                if (totalTermFreq == -1L) {
+                    // fallback used by Lucene in ES 6 (LUCENE-8007)
+                    totalTermFreq = docFreq;
+                }
+            }
+            TermStatistics stats = new TermStatistics(term2, docFreq, totalTermFreq);
             termStatistics.put(term, stats);
         }
         fieldStatistics = DfsSearchResult.readFieldStats(in);

server/src/main/java/org/elasticsearch/search/dfs/DfsSearchResult.java

@@ -116,7 +116,7 @@ public static void writeFieldStats(StreamOutput out, ObjectObjectHashMap<String,
         for (ObjectObjectCursor<String, CollectionStatistics> c : fieldStatistics) {
             out.writeString(c.key);
             CollectionStatistics statistics = c.value;
-            assert statistics.maxDoc() >= 0;
+            assert statistics.maxDoc() > 0;
             out.writeVLong(statistics.maxDoc());
             if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
                 // stats are always positive numbers
@@ -155,9 +155,9 @@ static ObjectObjectHashMap<String, CollectionStatistics> readFieldStats(StreamIn
         for (int i = 0; i < numFieldStatistics; i++) {
             final String field = in.readString();
             assert field != null;
-            final long maxDoc = in.readVLong();
+            long maxDoc = in.readVLong();
             final long docCount;
-            final long sumTotalTermFreq;
+            long sumTotalTermFreq;
             final long sumDocFreq;
             if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
                 // stats are always positive numbers
@@ -168,6 +168,23 @@ static ObjectObjectHashMap<String, CollectionStatistics> readFieldStats(StreamIn
                 docCount = subOne(in.readVLong());
                 sumTotalTermFreq = subOne(in.readVLong());
                 sumDocFreq = subOne(in.readVLong());
+                if (sumTotalTermFreq == -1L) {
+                    // fallback used by Lucene in ES 6 (LUCENE-8007)
+                    sumTotalTermFreq = sumDocFreq;
+                }
+                if (maxDoc == 0L) {
+                    // fallback used by Lucene in ES 6 (LUCENE-8007)
+                    maxDoc = docCount;
+                }
+                if (docCount == 0L) {
+                    // empty stats object (LUCENE-8020)
+                    assert maxDoc == 0 && docCount == 0 && sumTotalTermFreq == 0 && sumDocFreq == 0:
+                        " maxDoc:" + maxDoc +
+                        " docCount:" + docCount +
+                        " sumTotalTermFreq:" + sumTotalTermFreq +
+                        " sumDocFreq:" + sumDocFreq;
+                    continue;
+                }
             }
             CollectionStatistics stats = new CollectionStatistics(field, maxDoc, docCount, sumTotalTermFreq, sumDocFreq);
             fieldStatistics.put(field, stats);
@@ -187,10 +204,16 @@ static TermStatistics[] readTermStats(StreamInput in, Term[] terms) throws IOExc
                 BytesRef term = terms[i].bytes();
                 final long docFreq = in.readVLong();
                 assert docFreq >= 0;
-                final long totalTermFreq = subOne(in.readVLong());
+                long totalTermFreq = subOne(in.readVLong());
                 if (docFreq == 0) {
                     continue;
                 }
+                if (in.getVersion().before(Version.V_7_0_0)) {
+                    if (totalTermFreq == -1L) {
+                        // fallback used by Lucene in ES 6
+                        totalTermFreq = docFreq;
+                    }
+                }
                 termStatistics[i] = new TermStatistics(term, docFreq, totalTermFreq);
             }
         }