Term Vectors: support for distributed frequencies

Adds distributed frequencies support for the Term Vectors API. A new parameter
called `dfs` is introduced which defaults to `false`.

Closes #8144

Author: alexksikes

docs/reference/docs/termvectors.asciidoc

@@ -76,16 +76,23 @@ omit :
  * sum of total term frequencies (the sum of total term frequencies of
    each term in this field)
 
+[float]
+==== Distributed frequencies coming[1.5.0]
+
+Setting `dfs` to `true` (default is `false`) will return the term statistics
+or the field statistics of the entire index, and not just at the shard. Use it
+with caution as distributed frequencies can have a serious performance impact.
+
 [float]
 === Behaviour
 
 The term and field statistics are not accurate. Deleted documents
 are not taken into account. The information is only retrieved for the
-shard the requested document resides in. The term and field statistics
-are therefore only useful as relative measures whereas the absolute
-numbers have no meaning in this context. By default, when requesting
-term vectors of artificial documents, a shard to get the statistics from
-is randomly selected. Use `routing` only to hit a particular shard.
+shard the requested document resides in, unless `dfs` is set to `true`.
+The term and field statistics are therefore only useful as relative measures
+whereas the absolute numbers have no meaning in this context. By default,
+when requesting term vectors of artificial documents, a shard to get the statistics
+from is randomly selected. Use `routing` only to hit a particular shard.
 
 [float]
 === Example 1

rest-api-spec/api/termvector.json

@@ -35,6 +35,12 @@
            "default" : true,
            "required" : false
         },
+        "dfs" : {
+           "type" : "boolean",
+           "description" : "Specifies if distributed frequencies should be returned instead shard frequencies.",
+           "default" : false,
+           "required" : false
+        },
         "fields" : {
           "type" : "list",
           "description" : "A comma-separated list of fields to return.",

src/main/java/org/elasticsearch/action/ActionModule.java

@@ -160,6 +160,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.action.termvector.*;
+import org.elasticsearch.action.termvector.dfs.TransportDfsOnlyAction;
 import org.elasticsearch.action.update.TransportUpdateAction;
 import org.elasticsearch.action.update.UpdateAction;
 import org.elasticsearch.common.inject.AbstractModule;
@@ -280,7 +281,8 @@ protected void configure() {
 
         registerAction(IndexAction.INSTANCE, TransportIndexAction.class);
         registerAction(GetAction.INSTANCE, TransportGetAction.class);
-        registerAction(TermVectorAction.INSTANCE, TransportSingleShardTermVectorAction.class);
+        registerAction(TermVectorAction.INSTANCE, TransportSingleShardTermVectorAction.class,
+                TransportDfsOnlyAction.class);
         registerAction(MultiTermVectorsAction.INSTANCE, TransportMultiTermVectorsAction.class,
                 TransportSingleShardMultiTermsVectorAction.class);
         registerAction(DeleteAction.INSTANCE, TransportDeleteAction.class,

src/main/java/org/elasticsearch/action/termvector/TermVectorRequest.java

@@ -292,6 +292,22 @@ public TermVectorRequest fieldStatistics(boolean fieldStatistics) {
         return this;
     }
 
+    /**
+     * @return <code>true</code> if distributed frequencies should be returned. Otherwise
+     * <code>false</code>
+     */
+    public boolean dfs() {
+        return flagsEnum.contains(Flag.Dfs);
+    }
+
+    /**
+     * Use distributed frequencies instead of shard statistics.
+     */
+    public TermVectorRequest dfs(boolean dfs) {
+        setFlag(Flag.Dfs, dfs);
+        return this;
+    }
+
     /**
      * Return only term vectors for special selected fields. Returns for term
      * vectors for all fields if selectedFields == null
@@ -309,24 +325,30 @@ public TermVectorRequest selectedFields(String... fields) {
         return this;
     }
 
+    /**
+     * Return whether term vectors should be generated real-time (default to true).
+     */
     public boolean realtime() {
         return this.realtime == null ? true : this.realtime;
     }
 
+    /**
+     * Choose whether term vectors be generated real-time.
+     */
     public TermVectorRequest realtime(Boolean realtime) {
         this.realtime = realtime;
         return this;
     }
 
     /**
-     * Return the overridden analyzers at each field
+     * Return the overridden analyzers at each field.
      */
     public Map<String, String> perFieldAnalyzer() {
         return perFieldAnalyzer;
     }
 
     /**
-     * Override the analyzer used at each field when generating term vectors
+     * Override the analyzer used at each field when generating term vectors.
      */
     public TermVectorRequest perFieldAnalyzer(Map<String, String> perFieldAnalyzer) {
         this.perFieldAnalyzer = perFieldAnalyzer != null && perFieldAnalyzer.size() != 0 ? Maps.newHashMap(perFieldAnalyzer) : null;
@@ -444,7 +466,7 @@ public void writeTo(StreamOutput out) throws IOException {
     public static enum Flag {
         // Do not change the order of these flags we use
         // the ordinal for encoding! Only append to the end!
-        Positions, Offsets, Payloads, FieldStatistics, TermStatistics
+        Positions, Offsets, Payloads, FieldStatistics, TermStatistics, Dfs
     }
 
     /**
@@ -477,6 +499,8 @@ public static void parseRequest(TermVectorRequest termVectorRequest, XContentPar
                     termVectorRequest.termStatistics(parser.booleanValue());
                 } else if (currentFieldName.equals("field_statistics") || currentFieldName.equals("fieldStatistics")) {
                     termVectorRequest.fieldStatistics(parser.booleanValue());
+                } else if (currentFieldName.equals("dfs")) {
+                    termVectorRequest.dfs(parser.booleanValue());
                 } else if (currentFieldName.equals("per_field_analyzer") || currentFieldName.equals("perFieldAnalyzer")) {
                     termVectorRequest.perFieldAnalyzer(readPerFieldAnalyzer(parser.map()));
                 } else if ("_index".equals(currentFieldName)) { // the following is important for multi request parsing.

src/main/java/org/elasticsearch/action/termvector/TermVectorRequestBuilder.java

@@ -27,13 +27,23 @@
 import java.util.Map;
 
 /**
+ * The builder class for a term vector request.
+ * Returns the term vector (doc frequency, positions, offsets) for a document.
+ * <p/>
+ * Note, the {@code index}, {@code type} and {@code id} are
+ * required.
  */
 public class TermVectorRequestBuilder extends ActionRequestBuilder<TermVectorRequest, TermVectorResponse, TermVectorRequestBuilder, Client> {
 
     public TermVectorRequestBuilder(Client client) {
         super(client, new TermVectorRequest());
     }
 
+    /**
+     * Constructs a new term vector request builder for a document that will be fetch
+     * from the provided index. Use {@code index}, {@code type} and
+     * {@code id} to specify the document to load.
+     */
     public TermVectorRequestBuilder(Client client, String index, String type, String id) {
         super(client, new TermVectorRequest(index, type, id));
     }
@@ -92,47 +102,81 @@ public TermVectorRequestBuilder setParent(String parent) {
      * <tt>_local</tt> to prefer local shards, <tt>_primary</tt> to execute only on primary shards, or
      * a custom value, which guarantees that the same order will be used across different requests.
      */
-    
     public TermVectorRequestBuilder setPreference(String preference) {
         request.preference(preference);
         return this;
     }
-    
+
+    /**
+     * Sets whether to return the start and stop offsets for each term if they were stored or
+     * skip offsets.
+     */
     public TermVectorRequestBuilder setOffsets(boolean offsets) {
         request.offsets(offsets);
         return this;
     }
 
+
+    /**
+     * Sets whether to return the positions for each term if stored or skip.
+     */
     public TermVectorRequestBuilder setPositions(boolean positions) {
         request.positions(positions);
         return this;
     }
 
+    /**
+     * Sets whether to return the payloads for each term or skip.
+     */
     public TermVectorRequestBuilder setPayloads(boolean payloads) {
         request.payloads(payloads);
         return this;
     }
 
+    /**
+     * Sets whether to return the term statistics for each term in the shard or skip.
+     */
     public TermVectorRequestBuilder setTermStatistics(boolean termStatistics) {
         request.termStatistics(termStatistics);
         return this;
     }
 
+    /**
+     * Sets whether to return the field statistics for each term in the shard or skip.
+     */
     public TermVectorRequestBuilder setFieldStatistics(boolean fieldStatistics) {
         request.fieldStatistics(fieldStatistics);
         return this;
     }
 
+    /**
+     * Sets whether to use distributed frequencies instead of shard statistics.
+     */
+    public TermVectorRequestBuilder setDfs(boolean dfs) {
+        request.dfs(dfs);
+        return this;
+    }
+
+    /**
+     * Sets whether to return only term vectors for special selected fields. Returns the term
+     * vectors for all fields if selectedFields == null
+     */
     public TermVectorRequestBuilder setSelectedFields(String... fields) {
         request.selectedFields(fields);
         return this;
     }
 
+    /**
+     * Sets whether term vectors are generated real-time.
+     */
     public TermVectorRequestBuilder setRealtime(Boolean realtime) {
         request.realtime(realtime);
         return this;
     }
 
+    /**
+     * Sets the analyzer used at each field when generating term vectors.
+     */
     public TermVectorRequestBuilder setPerFieldAnalyzer(Map<String, String> perFieldAnalyzer) {
         request.perFieldAnalyzer(perFieldAnalyzer);
         return this;

src/main/java/org/elasticsearch/action/termvector/TermVectorResponse.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.termvector.TermVectorRequest.Flag;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -38,6 +39,7 @@
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
+import org.elasticsearch.search.dfs.AggregatedDfs;
 
 import java.io.IOException;
 import java.util.EnumSet;
@@ -320,10 +322,14 @@ public void setExists(boolean exists) {
     }
 
     public void setFields(Fields termVectorsByField, Set<String> selectedFields, EnumSet<Flag> flags, Fields topLevelFields) throws IOException {
+        setFields(termVectorsByField, selectedFields, flags, topLevelFields, null);
+    }
+
+    public void setFields(Fields termVectorsByField, Set<String> selectedFields, EnumSet<Flag> flags, Fields topLevelFields, @Nullable AggregatedDfs dfs) throws IOException {
         TermVectorWriter tvw = new TermVectorWriter(this);
 
         if (termVectorsByField != null) {
-            tvw.setFields(termVectorsByField, selectedFields, flags, topLevelFields);
+            tvw.setFields(termVectorsByField, selectedFields, flags, topLevelFields, dfs);
         }
 
     }

src/main/java/org/elasticsearch/action/termvector/TermVectorWriter.java

@@ -19,10 +19,14 @@
 package org.elasticsearch.action.termvector;
 
 import org.apache.lucene.index.*;
+import org.apache.lucene.search.CollectionStatistics;
+import org.apache.lucene.search.TermStatistics;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.action.termvector.TermVectorRequest.Flag;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.search.dfs.AggregatedDfs;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -45,7 +49,7 @@ final class TermVectorWriter {
         response = termVectorResponse;
     }
 
-    void setFields(Fields termVectorsByField, Set<String> selectedFields, EnumSet<Flag> flags, Fields topLevelFields) throws IOException {
+    void setFields(Fields termVectorsByField, Set<String> selectedFields, EnumSet<Flag> flags, Fields topLevelFields, @Nullable AggregatedDfs dfs) throws IOException {
         int numFieldsWritten = 0;
         TermsEnum iterator = null;
         DocsAndPositionsEnum docsAndPosEnum = null;
@@ -70,7 +74,11 @@ void setFields(Fields termVectorsByField, Set<String> selectedFields, EnumSet<Fl
             boolean payloads = flags.contains(Flag.Payloads) && fieldTermVector.hasPayloads();
             startField(field, fieldTermVector.size(), positions, offsets, payloads);
             if (flags.contains(Flag.FieldStatistics)) {
-                writeFieldStatistics(topLevelTerms);
+                if (dfs != null) {
+                    writeFieldStatistics(dfs.fieldStatistics().get(field));
+                } else {
+                    writeFieldStatistics(topLevelTerms);
+                }
             }
             iterator = fieldTermVector.iterator(iterator);
             final boolean useDocsAndPos = positions || offsets || payloads;
@@ -81,7 +89,11 @@ void setFields(Fields termVectorsByField, Set<String> selectedFields, EnumSet<Fl
                 boolean foundTerm = topLevelIterator.seekExact(term);
                 startTerm(term);
                 if (flags.contains(Flag.TermStatistics)) {
-                    writeTermStatistics(topLevelIterator);
+                    if (dfs != null) {
+                        writeTermStatistics(dfs.termStatistics().get(new Term(field, term.utf8ToString())));
+                    } else {
+                        writeTermStatistics(topLevelIterator);
+                    }
                 }
                 if (useDocsAndPos) {
                     // given we have pos or offsets
@@ -161,7 +173,6 @@ private void writePayload(BytesRef payload) throws IOException {
     }
 
     private void writeFreq(int termFreq) throws IOException {
-
         writePotentiallyNegativeVInt(termFreq);
     }
 
@@ -205,7 +216,15 @@ private void writeTermStatistics(TermsEnum topLevelIterator) throws IOException
         long ttf = topLevelIterator.totalTermFreq();
         assert (ttf >= -1);
         writePotentiallyNegativeVLong(ttf);
+    }
 
+    private void writeTermStatistics(TermStatistics termStatistics) throws IOException {
+        int docFreq = (int) termStatistics.docFreq();
+        assert (docFreq >= -1);
+        writePotentiallyNegativeVInt(docFreq);
+        long ttf = termStatistics.totalTermFreq();
+        assert (ttf >= -1);
+        writePotentiallyNegativeVLong(ttf);
     }
 
     private void writeFieldStatistics(Terms topLevelTerms) throws IOException {
@@ -218,7 +237,18 @@ private void writeFieldStatistics(Terms topLevelTerms) throws IOException {
         int dc = topLevelTerms.getDocCount();
         assert (dc >= -1);
         writePotentiallyNegativeVInt(dc);
+    }
 
+    private void writeFieldStatistics(CollectionStatistics fieldStats) throws IOException {
+        long sttf = fieldStats.sumTotalTermFreq();
+        assert (sttf >= -1);
+        writePotentiallyNegativeVLong(sttf);
+        long sdf = fieldStats.sumDocFreq();
+        assert (sdf >= -1);
+        writePotentiallyNegativeVLong(sdf);
+        int dc = (int) fieldStats.docCount();
+        assert (dc >= -1);
+        writePotentiallyNegativeVInt(dc);
     }
 
     private void writePotentiallyNegativeVInt(int value) throws IOException {