Multi term vector request

--------------------------

This feature allows to retrieve [term vectors](https://github.com/elasticsearch/elasticsearch/issues/3114) for a list of documents. The json request has exactly the same [format](https://github.com/elasticsearch/elasticsearch/issues/3484) as the ```_termvectors``` endpoint

It use it, call

```
curl -XGET 'http://localhost:9200/index/type/_mtermvectors' -d '{
    "fields": [
        "field1",
        "field2",
        ...
    ],
    "ids": [
        "docId1",
        "docId2",
        ...
    ],
    "offsets": false|true,
    "payloads": false|true,
    "positions": false|true,
    "term_statistics": false|true,
    "field_statistics": false|true
}'

```

The return format is an array, each entry of which conatins the term vector response for one document:

```
{
   "docs": [
      {
         "_index": "index",
         "_type": "type",
         "_id": "docId1",
         "_version": 1,
         "exists": true,
         "term_vectors": {
         	...
         }
      },
      {
         "_index": "index",
         "_type": "type",
         "_id": "docId2",
         "_version": 1,
         "exists": true,
         "term_vectors": {
         ...
         }
      }
   ]
}
```

Note that, like term vectors, the mult term vectors request will silenty skip over documents that have no term vectors stored in the index and will simply return an empty response in this case.

Closes #3536

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

@@ -126,8 +126,7 @@ import org.elasticsearch.action.search.type.*;
 import org.elasticsearch.action.suggest.SuggestAction;
 import org.elasticsearch.action.suggest.TransportSuggestAction;
 import org.elasticsearch.action.support.TransportAction;
-import org.elasticsearch.action.termvector.TermVectorAction;
-import org.elasticsearch.action.termvector.TransportSingleShardTermVectorAction;
+import org.elasticsearch.action.termvector.*;
 import org.elasticsearch.action.update.TransportUpdateAction;
 import org.elasticsearch.action.update.UpdateAction;
 import org.elasticsearch.common.inject.AbstractModule;
@@ -224,6 +223,8 @@ public class ActionModule extends AbstractModule {
         registerAction(IndexAction.INSTANCE, TransportIndexAction.class);
         registerAction(GetAction.INSTANCE, TransportGetAction.class);
         registerAction(TermVectorAction.INSTANCE, TransportSingleShardTermVectorAction.class);
+        registerAction(MultiTermVectorsAction.INSTANCE, TransportMultiTermVectorsAction.class,
+                TransportSingleShardMultiTermsVectorAction.class);
         registerAction(DeleteAction.INSTANCE, TransportDeleteAction.class,
                 TransportIndexDeleteAction.class, TransportShardDeleteAction.class);
         registerAction(CountAction.INSTANCE, TransportCountAction.class);

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

@@ -0,0 +1,45 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import org.elasticsearch.action.Action;
+import org.elasticsearch.client.Client;
+
+/**
+ */
+public class MultiTermVectorsAction extends Action<MultiTermVectorsRequest, MultiTermVectorsResponse, MultiTermVectorsRequestBuilder> {
+
+    public static final MultiTermVectorsAction INSTANCE = new MultiTermVectorsAction();
+    public static final String NAME = "mtv";
+
+    private MultiTermVectorsAction() {
+        super(NAME);
+    }
+
+    @Override
+    public MultiTermVectorsResponse newResponse() {
+        return new MultiTermVectorsResponse();
+    }
+
+    @Override
+    public MultiTermVectorsRequestBuilder newRequestBuilder(Client client) {
+        return new MultiTermVectorsRequestBuilder(client);
+    }
+}

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

@@ -0,0 +1,123 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Streamable;
+
+import java.io.IOException;
+
+/**
+ * A single multi get response.
+ */
+public class MultiTermVectorsItemResponse implements Streamable {
+
+    private TermVectorResponse response;
+    private MultiTermVectorsResponse.Failure failure;
+
+    MultiTermVectorsItemResponse() {
+
+    }
+
+    public MultiTermVectorsItemResponse(TermVectorResponse response, MultiTermVectorsResponse.Failure failure) {
+        assert (((response == null) && (failure != null)) || ((response != null) && (failure == null)));
+        this.response = response;
+        this.failure = failure;
+    }
+
+    /**
+     * The index name of the document.
+     */
+    public String getIndex() {
+        if (failure != null) {
+            return failure.getIndex();
+        }
+        return response.getIndex();
+    }
+
+    /**
+     * The type of the document.
+     */
+    public String getType() {
+        if (failure != null) {
+            return failure.getType();
+        }
+        return response.getType();
+    }
+
+    /**
+     * The id of the document.
+     */
+    public String getId() {
+        if (failure != null) {
+            return failure.getId();
+        }
+        return response.getId();
+    }
+
+    /**
+     * Is this a failed execution?
+     */
+    public boolean isFailed() {
+        return failure != null;
+    }
+
+    /**
+     * The actual get response, <tt>null</tt> if its a failure.
+     */
+    public TermVectorResponse getResponse() {
+        return this.response;
+    }
+
+    /**
+     * The failure if relevant.
+     */
+    public MultiTermVectorsResponse.Failure getFailure() {
+        return this.failure;
+    }
+
+    public static MultiTermVectorsItemResponse readItemResponse(StreamInput in) throws IOException {
+        MultiTermVectorsItemResponse response = new MultiTermVectorsItemResponse();
+        response.readFrom(in);
+        return response;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        if (in.readBoolean()) {
+            failure = MultiTermVectorsResponse.Failure.readFailure(in);
+        } else {
+            response = new TermVectorResponse();
+            response.readFrom(in);
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        if (failure != null) {
+            out.writeBoolean(true);
+            failure.writeTo(out);
+        } else {
+            out.writeBoolean(false);
+            response.writeTo(out);
+        }
+    }
+}

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

@@ -0,0 +1,210 @@
+package org.elasticsearch.action.termvector;
+
+/*
+ * 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.
+ */
+
+import org.elasticsearch.ElasticSearchIllegalArgumentException;
+import org.elasticsearch.ElasticSearchParseException;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ValidateActions;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MultiTermVectorsRequest extends ActionRequest<MultiTermVectorsRequest> {
+
+    String preference;
+    List<TermVectorRequest> requests = new ArrayList<TermVectorRequest>();
+
+    public MultiTermVectorsRequest add(TermVectorRequest termVectorRequest) {
+        requests.add(termVectorRequest);
+        return this;
+    }
+
+    public MultiTermVectorsRequest add(String index, @Nullable String type, String id) {
+        requests.add(new TermVectorRequest(index, type, id));
+        return this;
+    }
+
+    /**
+     * Sets the preference to execute the search. Defaults to randomize across
+     * shards. Can be set to <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 MultiTermVectorsRequest preference(String preference) {
+        this.preference = preference;
+        return this;
+    }
+
+    public String preference() {
+        return this.preference;
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        if (requests.isEmpty()) {
+            validationException = ValidateActions.addValidationError("multi term vectors: no documents requested", validationException);
+        } else {
+            for (int i = 0; i < requests.size(); i++) {
+                TermVectorRequest termVectorRequest = requests.get(i);
+                ActionRequestValidationException validationExceptionForDoc = termVectorRequest.validate();
+                if (validationExceptionForDoc != null) {
+                    validationException = ValidateActions.addValidationError("at multi term vectors for doc " + i,
+                            validationExceptionForDoc);
+                }
+            }
+        }
+        return validationException;
+    }
+
+    public void add(@Nullable String defaultIndex, @Nullable String defaultType, @Nullable String[] defaultFields, byte[] data, int from,
+            int length) throws Exception {
+        add(defaultIndex, defaultType, defaultFields, new BytesArray(data, from, length));
+    }
+
+    public void add(@Nullable String defaultIndex, @Nullable String defaultType, @Nullable String[] defaultFields, BytesReference data)
+            throws Exception {
+        XContentParser parser = XContentFactory.xContent(data).createParser(data);
+        try {
+            XContentParser.Token token;
+            String currentFieldName = null;
+            boolean offsets = true;
+            boolean offsetsFound = false;
+            boolean positions = true;
+            boolean positionsFound = false;
+            boolean payloads = true;
+            boolean payloadsFound = false;
+            boolean termStatistics = false;
+            boolean termStatisticsFound = false;
+            boolean fieldStatistics = true;
+            boolean fieldStatisticsFound = false;
+            while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+                if (token == XContentParser.Token.FIELD_NAME) {
+                    currentFieldName = parser.currentName();
+                } else if (token == XContentParser.Token.VALUE_BOOLEAN) {
+                    if (currentFieldName.equals("offsets")) {
+                        offsets = parser.booleanValue();
+                        offsetsFound = true;
+                    } else if (currentFieldName.equals("positions")) {
+                        positions = parser.booleanValue();
+                        positionsFound = true;
+                    } else if (currentFieldName.equals("payloads")) {
+                        payloads = parser.booleanValue();
+                        payloadsFound = true;
+                    } else if (currentFieldName.equals("term_statistics") || currentFieldName.equals("termStatistics")) {
+                        termStatistics = parser.booleanValue();
+                        termStatisticsFound = true;
+                    } else if (currentFieldName.equals("field_statistics") || currentFieldName.equals("fieldStatistics")) {
+                        fieldStatistics = parser.booleanValue();
+                        fieldStatisticsFound = true;
+                    } else {
+                        throw new ElasticSearchParseException("_mtermvectors: Parameter " + currentFieldName + "not supported");
+                    }
+                } else if (token == XContentParser.Token.START_ARRAY) {
+
+                    if ("docs".equals(currentFieldName)) {
+                        while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+                            if (token != XContentParser.Token.START_OBJECT) {
+                                throw new ElasticSearchIllegalArgumentException("docs array element should include an object");
+                            }
+                            TermVectorRequest termVectorRequest = new TermVectorRequest(defaultIndex, defaultType, null);
+
+                            TermVectorRequest.parseRequest(termVectorRequest, parser);
+
+                            if (defaultFields != null) {
+                                termVectorRequest.selectedFields(defaultFields.clone());
+                            }
+
+                            add(termVectorRequest);
+                        }
+                    } else if ("ids".equals(currentFieldName)) {
+                        while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+                            if (!token.isValue()) {
+                                throw new ElasticSearchIllegalArgumentException("ids array element should only contain ids");
+                            }
+                            TermVectorRequest tvr = new TermVectorRequest(defaultIndex, defaultType, parser.text());
+                            if (defaultFields != null) {
+                                tvr.selectedFields(defaultFields.clone());
+                            }
+                            add(tvr);
+                        }
+                    } else {
+                        throw new ElasticSearchParseException("_mtermvectors: Parameter " + currentFieldName + "not supported");
+                    }
+                } else if (currentFieldName != null) {
+                    throw new ElasticSearchParseException("_mtermvectors: Parameter " + currentFieldName + "not supported");
+                }
+            }
+            for (int i = 0; i < requests.size(); i++) {
+                TermVectorRequest curRequest = requests.get(i);
+                if (offsetsFound) {
+                    curRequest.offsets(offsets);
+                }
+                if (payloadsFound) {
+                    curRequest.payloads(payloads);
+                }
+                if (fieldStatisticsFound) {
+                    curRequest.fieldStatistics(fieldStatistics);
+                }
+                if (positionsFound) {
+                    curRequest.positions(positions);
+                }
+                if (termStatisticsFound) {
+                    curRequest.termStatistics(termStatistics);
+                }
+                requests.set(i, curRequest);
+            }
+        } finally {
+            parser.close();
+        }
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        preference = in.readOptionalString();
+        int size = in.readVInt();
+        requests = new ArrayList<TermVectorRequest>(size);
+        for (int i = 0; i < size; i++) {
+            requests.add(TermVectorRequest.readTermVectorRequest(in));
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeOptionalString(preference);
+        out.writeVInt(requests.size());
+        for (TermVectorRequest termVectorRequest : requests) {
+            termVectorRequest.writeTo(out);
+        }
+    }
+}

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

@@ -0,0 +1,66 @@
+package org.elasticsearch.action.termvector;
+/*
+ * 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.
+ */
+
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRequestBuilder;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.internal.InternalClient;
+import org.elasticsearch.common.Nullable;
+
+public class MultiTermVectorsRequestBuilder extends ActionRequestBuilder<MultiTermVectorsRequest, MultiTermVectorsResponse, MultiTermVectorsRequestBuilder> {
+    public MultiTermVectorsRequestBuilder(Client client) {
+        super((InternalClient) client, new MultiTermVectorsRequest());
+    }
+
+    public MultiTermVectorsRequestBuilder add(String index, @Nullable String type, Iterable<String> ids) {
+        for (String id : ids) {
+            request.add(index, type, id);
+        }
+        return this;
+    }
+
+    public MultiTermVectorsRequestBuilder add(String index, @Nullable String type, String... ids) {
+        for (String id : ids) {
+            request.add(index, type, id);
+        }
+        return this;
+    }
+
+    public MultiTermVectorsRequestBuilder add(TermVectorRequest termVectorRequest) {
+        request.add(termVectorRequest);
+        return this;
+    }
+
+    /**
+     * Sets the preference to execute the search. Defaults to randomize across shards. Can be set to
+     * <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 MultiTermVectorsRequestBuilder setPreference(String preference) {
+        request.preference(preference);
+        return this;
+    }
+
+    @Override
+    protected void doExecute(ActionListener<MultiTermVectorsResponse> listener) {
+        ((Client) client).multiTermVectors(request, listener);
+    }
+}

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

@@ -0,0 +1,173 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import com.google.common.collect.Iterators;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentBuilderString;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+public class MultiTermVectorsResponse extends ActionResponse implements Iterable<MultiTermVectorsItemResponse>, ToXContent {
+
+    /**
+     * Represents a failure.
+     */
+    public static class Failure implements Streamable {
+        private String index;
+        private String type;
+        private String id;
+        private String message;
+
+        Failure() {
+
+        }
+
+        public Failure(String index, String type, String id, String message) {
+            this.index = index;
+            this.type = type;
+            this.id = id;
+            this.message = message;
+        }
+
+        /**
+         * The index name of the action.
+         */
+        public String getIndex() {
+            return this.index;
+        }
+
+        /**
+         * The type of the action.
+         */
+        public String getType() {
+            return type;
+        }
+
+        /**
+         * The id of the action.
+         */
+        public String getId() {
+            return id;
+        }
+
+        /**
+         * The failure message.
+         */
+        public String getMessage() {
+            return this.message;
+        }
+
+        public static Failure readFailure(StreamInput in) throws IOException {
+            Failure failure = new Failure();
+            failure.readFrom(in);
+            return failure;
+        }
+
+        @Override
+        public void readFrom(StreamInput in) throws IOException {
+            index = in.readString();
+            type = in.readOptionalString();
+            id = in.readString();
+            message = in.readString();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeString(index);
+            out.writeOptionalString(type);
+            out.writeString(id);
+            out.writeString(message);
+        }
+    }
+
+    private MultiTermVectorsItemResponse[] responses;
+
+    MultiTermVectorsResponse() {
+    }
+
+    public MultiTermVectorsResponse(MultiTermVectorsItemResponse[] responses) {
+        this.responses = responses;
+    }
+
+    public MultiTermVectorsItemResponse[] getResponses() {
+        return this.responses;
+    }
+
+    @Override
+    public Iterator<MultiTermVectorsItemResponse> iterator() {
+        return Iterators.forArray(responses);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.startArray(Fields.DOCS);
+        for (MultiTermVectorsItemResponse response : responses) {
+            if (response.isFailed()) {
+                builder.startObject();
+                Failure failure = response.getFailure();
+                builder.field(Fields._INDEX, failure.getIndex());
+                builder.field(Fields._TYPE, failure.getType());
+                builder.field(Fields._ID, failure.getId());
+                builder.field(Fields.ERROR, failure.getMessage());
+                builder.endObject();
+            } else {
+                TermVectorResponse getResponse = response.getResponse();
+                getResponse.toXContent(builder, params);
+            }
+        }
+        builder.endArray();
+        builder.endObject();
+        return builder;
+    }
+
+    static final class Fields {
+        static final XContentBuilderString DOCS = new XContentBuilderString("docs");
+        static final XContentBuilderString _INDEX = new XContentBuilderString("_index");
+        static final XContentBuilderString _TYPE = new XContentBuilderString("_type");
+        static final XContentBuilderString _ID = new XContentBuilderString("_id");
+        static final XContentBuilderString ERROR = new XContentBuilderString("error");
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        responses = new MultiTermVectorsItemResponse[in.readVInt()];
+        for (int i = 0; i < responses.length; i++) {
+            responses[i] = MultiTermVectorsItemResponse.readItemResponse(in);
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeVInt(responses.length);
+        for (MultiTermVectorsItemResponse response : responses) {
+            response.writeTo(out);
+        }
+    }
+}

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

@@ -0,0 +1,99 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import gnu.trove.list.array.TIntArrayList;
+import org.elasticsearch.action.support.single.shard.SingleShardOperationRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MultiTermVectorsShardRequest extends SingleShardOperationRequest<MultiTermVectorsShardRequest> {
+
+    private int shardId;
+    private String preference;
+
+    TIntArrayList locations;
+    List<TermVectorRequest> requests;
+
+    MultiTermVectorsShardRequest() {
+
+    }
+
+    MultiTermVectorsShardRequest(String index, int shardId) {
+        super(index);
+        this.shardId = shardId;
+        locations = new TIntArrayList();
+        requests = new ArrayList<TermVectorRequest>();
+    }
+
+    public int shardId() {
+        return this.shardId;
+    }
+
+    /**
+     * Sets the preference to execute the search. Defaults to randomize across shards. Can be set to
+     * <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 MultiTermVectorsShardRequest preference(String preference) {
+        this.preference = preference;
+        return this;
+    }
+
+    public String preference() {
+        return this.preference;
+    }
+
+
+    public void add(int location, TermVectorRequest request) {
+        this.locations.add(location);
+        this.requests.add(request);
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        int size = in.readVInt();
+        locations = new TIntArrayList(size);
+        requests = new ArrayList<TermVectorRequest>(size);
+        for (int i = 0; i < size; i++) {
+            locations.add(in.readVInt());
+            requests.add(TermVectorRequest.readTermVectorRequest(in));
+        }
+
+        preference = in.readOptionalString();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeVInt(locations.size());
+        for (int i = 0; i < locations.size(); i++) {
+            out.writeVInt(locations.get(i));
+            requests.get(i).writeTo(out);
+        }
+
+        out.writeOptionalString(preference);
+    }
+}

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

@@ -0,0 +1,99 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import gnu.trove.list.array.TIntArrayList;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MultiTermVectorsShardResponse extends ActionResponse {
+
+    TIntArrayList locations;
+    List<TermVectorResponse> responses;
+    List<MultiTermVectorsResponse.Failure> failures;
+
+    MultiTermVectorsShardResponse() {
+        locations = new TIntArrayList();
+        responses = new ArrayList<TermVectorResponse>();
+        failures = new ArrayList<MultiTermVectorsResponse.Failure>();
+    }
+
+    public void add(int location, TermVectorResponse response) {
+        locations.add(location);
+        responses.add(response);
+        failures.add(null);
+    }
+
+    public void add(int location, MultiTermVectorsResponse.Failure failure) {
+        locations.add(location);
+        responses.add(null);
+        failures.add(failure);
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        int size = in.readVInt();
+        locations = new TIntArrayList(size);
+        responses = new ArrayList<TermVectorResponse>(size);
+        failures = new ArrayList<MultiTermVectorsResponse.Failure>(size);
+        for (int i = 0; i < size; i++) {
+            locations.add(in.readVInt());
+            if (in.readBoolean()) {
+                TermVectorResponse response = new TermVectorResponse();
+                response.readFrom(in);
+                responses.add(response);
+            } else {
+                responses.add(null);
+            }
+            if (in.readBoolean()) {
+                failures.add(MultiTermVectorsResponse.Failure.readFailure(in));
+            } else {
+                failures.add(null);
+            }
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeVInt(locations.size());
+        for (int i = 0; i < locations.size(); i++) {
+            out.writeVInt(locations.get(i));
+            if (responses.get(i) == null) {
+                out.writeBoolean(false);
+            } else {
+                out.writeBoolean(true);
+                responses.get(i).writeTo(out);
+            }
+            if (failures.get(i) == null) {
+                out.writeBoolean(false);
+            } else {
+                out.writeBoolean(true);
+                failures.get(i).writeTo(out);
+            }
+        }
+    }
+}

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

@@ -0,0 +1,166 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.get.MultiGetAction;
+import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.cluster.ClusterService;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.AtomicArray;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.BaseTransportRequestHandler;
+import org.elasticsearch.transport.TransportChannel;
+import org.elasticsearch.transport.TransportService;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TransportMultiTermVectorsAction extends TransportAction<MultiTermVectorsRequest, MultiTermVectorsResponse> {
+
+    private final ClusterService clusterService;
+
+    private final TransportSingleShardMultiTermsVectorAction shardAction;
+
+    @Inject
+    public TransportMultiTermVectorsAction(Settings settings, ThreadPool threadPool, TransportService transportService,
+            ClusterService clusterService, TransportSingleShardMultiTermsVectorAction shardAction) {
+        super(settings, threadPool);
+        this.clusterService = clusterService;
+        this.shardAction = shardAction;
+
+        transportService.registerHandler(MultiGetAction.NAME, new TransportHandler());
+    }
+
+    @Override
+    protected void doExecute(final MultiTermVectorsRequest request, final ActionListener<MultiTermVectorsResponse> listener) {
+        ClusterState clusterState = clusterService.state();
+
+        clusterState.blocks().globalBlockedRaiseException(ClusterBlockLevel.READ);
+
+        final AtomicArray<MultiTermVectorsItemResponse> responses = new AtomicArray<MultiTermVectorsItemResponse>(request.requests.size());
+
+        Map<ShardId, MultiTermVectorsShardRequest> shardRequests = new HashMap<ShardId, MultiTermVectorsShardRequest>();
+        for (int i = 0; i < request.requests.size(); i++) {
+            TermVectorRequest termVectorRequest = request.requests.get(i);
+            termVectorRequest.routing(clusterState.metaData().resolveIndexRouting(termVectorRequest.routing(), termVectorRequest.index()));
+            if (!clusterState.metaData().hasConcreteIndex(termVectorRequest.index())) {
+                responses.set(i, new MultiTermVectorsItemResponse(null, new MultiTermVectorsResponse.Failure(termVectorRequest.index(),
+                        termVectorRequest.type(), termVectorRequest.id(), "[" + termVectorRequest.index() + "] missing")));
+                continue;
+            }
+            termVectorRequest.index(clusterState.metaData().concreteIndex(termVectorRequest.index()));
+            ShardId shardId = clusterService
+                    .operationRouting()
+                    .getShards(clusterState, termVectorRequest.index(), termVectorRequest.type(), termVectorRequest.id(),
+                            termVectorRequest.routing(), null).shardId();
+            MultiTermVectorsShardRequest shardRequest = shardRequests.get(shardId);
+            if (shardRequest == null) {
+                shardRequest = new MultiTermVectorsShardRequest(shardId.index().name(), shardId.id());
+                shardRequest.preference(request.preference);
+
+                shardRequests.put(shardId, shardRequest);
+            }
+            shardRequest.add(i, termVectorRequest);
+        }
+
+        final AtomicInteger counter = new AtomicInteger(shardRequests.size());
+
+        for (final MultiTermVectorsShardRequest shardRequest : shardRequests.values()) {
+            shardAction.execute(shardRequest, new ActionListener<MultiTermVectorsShardResponse>() {
+                @Override
+                public void onResponse(MultiTermVectorsShardResponse response) {
+                    for (int i = 0; i < response.locations.size(); i++) {
+                        responses.set(response.locations.get(i), new MultiTermVectorsItemResponse(response.responses.get(i),
+                                response.failures.get(i)));
+                    }
+                    if (counter.decrementAndGet() == 0) {
+                        finishHim();
+                    }
+                }
+
+                @Override
+                public void onFailure(Throwable e) {
+                    // create failures for all relevant requests
+                    String message = ExceptionsHelper.detailedMessage(e);
+                    for (int i = 0; i < shardRequest.locations.size(); i++) {
+                        TermVectorRequest termVectorRequest = shardRequest.requests.get(i);
+                        responses.set(shardRequest.locations.get(i), new MultiTermVectorsItemResponse(null,
+                                new MultiTermVectorsResponse.Failure(shardRequest.index(), termVectorRequest.type(),
+                                        termVectorRequest.id(), message)));
+                    }
+                    if (counter.decrementAndGet() == 0) {
+                        finishHim();
+                    }
+                }
+
+                private void finishHim() {
+                    listener.onResponse(new MultiTermVectorsResponse(
+                            responses.toArray(new MultiTermVectorsItemResponse[responses.length()])));
+                }
+            });
+        }
+    }
+
+    class TransportHandler extends BaseTransportRequestHandler<MultiTermVectorsRequest> {
+
+        @Override
+        public MultiTermVectorsRequest newInstance() {
+            return new MultiTermVectorsRequest();
+        }
+
+        @Override
+        public void messageReceived(final MultiTermVectorsRequest request, final TransportChannel channel) throws Exception {
+            // no need to use threaded listener, since we just send a response
+            request.listenerThreaded(false);
+            execute(request, new ActionListener<MultiTermVectorsResponse>() {
+                @Override
+                public void onResponse(MultiTermVectorsResponse response) {
+                    try {
+                        channel.sendResponse(response);
+                    } catch (Throwable t) {
+                        onFailure(t);
+                    }
+                }
+
+                @Override
+                public void onFailure(Throwable e) {
+                    try {
+                        channel.sendResponse(e);
+                    } catch (Throwable t) {
+                        logger.warn("Failed to send error response for action [" + MultiTermVectorsAction.NAME + "] and request ["
+                                + request + "]", t);
+                    }
+                }
+            });
+        }
+
+        @Override
+        public String executor() {
+            return ThreadPool.Names.SAME;
+        }
+    }
+}

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

@@ -0,0 +1,113 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.action.termvector;
+
+import org.elasticsearch.ElasticSearchException;
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.support.single.shard.TransportShardSingleOperationAction;
+import org.elasticsearch.cluster.ClusterService;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.routing.ShardIterator;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.service.IndexService;
+import org.elasticsearch.index.shard.service.IndexShard;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+public class TransportSingleShardMultiTermsVectorAction extends TransportShardSingleOperationAction<MultiTermVectorsShardRequest, MultiTermVectorsShardResponse> {
+
+    private final IndicesService indicesService;
+
+    @Inject
+    public TransportSingleShardMultiTermsVectorAction(Settings settings, ClusterService clusterService, TransportService transportService,
+                                                      IndicesService indicesService, ThreadPool threadPool) {
+        super(settings, threadPool, clusterService, transportService);
+        this.indicesService = indicesService;
+    }
+
+    @Override
+    protected String executor() {
+        return ThreadPool.Names.GET;
+    }
+
+    @Override
+    protected String transportAction() {
+        return MultiTermVectorsAction.NAME + "/shard";
+    }
+
+    @Override
+    protected MultiTermVectorsShardRequest newRequest() {
+        return new MultiTermVectorsShardRequest();
+    }
+
+    @Override
+    protected MultiTermVectorsShardResponse newResponse() {
+        return new MultiTermVectorsShardResponse();
+    }
+
+    @Override
+    protected ClusterBlockException checkGlobalBlock(ClusterState state, MultiTermVectorsShardRequest request) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.READ);
+    }
+
+    @Override
+    protected ClusterBlockException checkRequestBlock(ClusterState state, MultiTermVectorsShardRequest request) {
+        return state.blocks().indexBlockedException(ClusterBlockLevel.READ, request.index());
+    }
+
+    @Override
+    protected ShardIterator shards(ClusterState state, MultiTermVectorsShardRequest request) {
+        return clusterService.operationRouting()
+                .getShards(clusterService.state(), request.index(), request.shardId(), request.preference());
+    }
+
+    @Override
+    protected void resolveRequest(ClusterState state, MultiTermVectorsShardRequest request) {
+        // no need to set concrete index and routing here, it has already been set by the multi term vectors action on the item
+        // request.index(state.metaData().concreteIndex(request.index()));
+    }
+
+    @Override
+    protected MultiTermVectorsShardResponse shardOperation(MultiTermVectorsShardRequest request, int shardId) throws ElasticSearchException {
+
+        MultiTermVectorsShardResponse response = new MultiTermVectorsShardResponse();
+        for (int i = 0; i < request.locations.size(); i++) {
+            TermVectorRequest termVectorRequest = request.requests.get(i);
+
+            try {
+                IndexService indexService = indicesService.indexServiceSafe(request.index());
+                IndexShard indexShard = indexService.shardSafe(shardId);
+                TermVectorResponse termVectorResponse = indexShard.termVectorService().getTermVector(termVectorRequest);
+                response.add(request.locations.get(i), termVectorResponse);
+            } catch (Throwable t) {
+                logger.debug("[{}][{}] failed to execute multi term vectors for [{}]/[{}]", t, request.index(), shardId, termVectorRequest.type(), termVectorRequest.id());
+                response.add(request.locations.get(i),
+                        new MultiTermVectorsResponse.Failure(request.index(), termVectorRequest.type(), termVectorRequest.id(), ExceptionsHelper.detailedMessage(t)));
+            }
+        }
+
+        return response;
+    }
+
+}

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

@@ -19,10 +19,6 @@
 
 package org.elasticsearch.action.termvector;
 
-import org.apache.lucene.index.Fields;
-import org.apache.lucene.index.IndexReader;
-import org.apache.lucene.index.MultiFields;
-import org.apache.lucene.index.Term;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.action.support.single.shard.TransportShardSingleOperationAction;
 import org.elasticsearch.cluster.ClusterService;
@@ -31,11 +27,7 @@ import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.index.engine.Engine;
-import org.elasticsearch.index.mapper.Uid;
-import org.elasticsearch.index.mapper.internal.UidFieldMapper;
 import org.elasticsearch.index.service.IndexService;
 import org.elasticsearch.index.shard.service.IndexShard;
 import org.elasticsearch.indices.IndicesService;
@@ -51,7 +43,7 @@ public class TransportSingleShardTermVectorAction extends TransportShardSingleOp
 
     @Inject
     public TransportSingleShardTermVectorAction(Settings settings, ClusterService clusterService, TransportService transportService,
-            IndicesService indicesService, ThreadPool threadPool) {
+                                                IndicesService indicesService, ThreadPool threadPool) {
         super(settings, threadPool, clusterService, transportService);
         this.indicesService = indicesService;
     }
@@ -94,31 +86,7 @@ public class TransportSingleShardTermVectorAction extends TransportShardSingleOp
     protected TermVectorResponse shardOperation(TermVectorRequest request, int shardId) throws ElasticSearchException {
         IndexService indexService = indicesService.indexServiceSafe(request.index());
         IndexShard indexShard = indexService.shardSafe(shardId);
-
-        final Engine.Searcher searcher = indexShard.searcher();
-        IndexReader topLevelReader = searcher.reader();
-
-        final TermVectorResponse termVectorResponse = new TermVectorResponse(request.index(), request.type(), request.id());
-        final Term uidTerm = new Term(UidFieldMapper.NAME, Uid.createUidAsBytes(request.type(), request.id()));
-        try {
-            Fields topLevelFields = MultiFields.getFields(topLevelReader);
-            Versions.DocIdAndVersion docIdAndVersion = Versions.loadDocIdAndVersion(topLevelReader, uidTerm);
-            if (docIdAndVersion != null) {
-                termVectorResponse.setFields(docIdAndVersion.context.reader().getTermVectors(docIdAndVersion.docId), request.selectedFields(),
-                        request.getFlags(), topLevelFields);
-                termVectorResponse.setDocVersion(docIdAndVersion.version);
-                termVectorResponse.setExists(true);
-            } else {
-                
-            }
-
-        } catch (Throwable ex) {
-            
-            throw new ElasticSearchException("failed to execute term vector request", ex);
-        } finally {
-            searcher.release();
-        }
-        return termVectorResponse;
+        return indexShard.termVectorService().getTermVector(request);
     }
 
     @Override

src/main/java/org/elasticsearch/client/Client.java

@@ -46,9 +46,7 @@ import org.elasticsearch.action.search.*;
 import org.elasticsearch.action.suggest.SuggestRequest;
 import org.elasticsearch.action.suggest.SuggestRequestBuilder;
 import org.elasticsearch.action.suggest.SuggestResponse;
-import org.elasticsearch.action.termvector.TermVectorRequest;
-import org.elasticsearch.action.termvector.TermVectorRequestBuilder;
-import org.elasticsearch.action.termvector.TermVectorResponse;
+import org.elasticsearch.action.termvector.*;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.elasticsearch.action.update.UpdateResponse;
@@ -434,7 +432,7 @@ public interface Client {
      * @param listener A listener to be notified of the result
      */
     void moreLikeThis(MoreLikeThisRequest request, ActionListener<SearchResponse> listener);
-    
+
     /**
      * A more like this action to search for documents that are "like" a specific document.
      *
@@ -444,7 +442,7 @@ public interface Client {
      */
     MoreLikeThisRequestBuilder prepareMoreLikeThis(String index, String type, String id);
 
-    
+
     /**
      * An action that returns the term vectors for a specific document.
      *
@@ -471,6 +469,23 @@ public interface Client {
      */
     TermVectorRequestBuilder prepareTermVector(String index, String type, String id);
 
+
+    /**
+     * Multi get term vectors.
+     */
+    ActionFuture<MultiTermVectorsResponse> multiTermVectors(MultiTermVectorsRequest request);
+
+    /**
+     * Multi get term vectors.
+     */
+    void multiTermVectors(MultiTermVectorsRequest request, ActionListener<MultiTermVectorsResponse> listener);
+
+    /**
+     * Multi get term vectors.
+     */
+    MultiTermVectorsRequestBuilder prepareMultiTermVectors();
+
+
     /**
      * Percolates a request returning the matches documents.
      */

src/main/java/org/elasticsearch/client/support/AbstractClient.java

@@ -54,10 +54,7 @@ import org.elasticsearch.action.suggest.SuggestAction;
 import org.elasticsearch.action.suggest.SuggestRequest;
 import org.elasticsearch.action.suggest.SuggestRequestBuilder;
 import org.elasticsearch.action.suggest.SuggestResponse;
-import org.elasticsearch.action.termvector.TermVectorAction;
-import org.elasticsearch.action.termvector.TermVectorRequest;
-import org.elasticsearch.action.termvector.TermVectorRequestBuilder;
-import org.elasticsearch.action.termvector.TermVectorResponse;
+import org.elasticsearch.action.termvector.*;
 import org.elasticsearch.action.update.UpdateAction;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
@@ -294,7 +291,7 @@ public abstract class AbstractClient implements InternalClient {
     public MoreLikeThisRequestBuilder prepareMoreLikeThis(String index, String type, String id) {
         return new MoreLikeThisRequestBuilder(this, index, type, id);
     }
-    
+
     @Override
     public ActionFuture<TermVectorResponse> termVector(final TermVectorRequest request) {
         return execute(TermVectorAction.INSTANCE, request);
@@ -310,6 +307,21 @@ public abstract class AbstractClient implements InternalClient {
         return new TermVectorRequestBuilder(this, index, type, id);
     }
 
+    @Override
+    public ActionFuture<MultiTermVectorsResponse> multiTermVectors(final MultiTermVectorsRequest request) {
+        return execute(MultiTermVectorsAction.INSTANCE, request);
+    }
+
+    @Override
+    public void multiTermVectors(final MultiTermVectorsRequest request, final ActionListener<MultiTermVectorsResponse> listener) {
+        execute(MultiTermVectorsAction.INSTANCE, request, listener);
+    }
+
+    @Override
+    public MultiTermVectorsRequestBuilder prepareMultiTermVectors() {
+        return new MultiTermVectorsRequestBuilder(this);
+    }
+
     @Override
     public ActionFuture<PercolateResponse> percolate(final PercolateRequest request) {
         return execute(PercolateAction.INSTANCE, request);

src/main/java/org/elasticsearch/client/transport/TransportClient.java

@@ -45,6 +45,8 @@ import org.elasticsearch.action.percolate.PercolateResponse;
 import org.elasticsearch.action.search.*;
 import org.elasticsearch.action.suggest.SuggestRequest;
 import org.elasticsearch.action.suggest.SuggestResponse;
+import org.elasticsearch.action.termvector.MultiTermVectorsRequest;
+import org.elasticsearch.action.termvector.MultiTermVectorsResponse;
 import org.elasticsearch.action.termvector.TermVectorRequest;
 import org.elasticsearch.action.termvector.TermVectorResponse;
 import org.elasticsearch.action.update.UpdateRequest;
@@ -439,11 +441,26 @@ public class TransportClient extends AbstractClient {
         internalClient.moreLikeThis(request, listener);
     }
 
+    @Override
+    public ActionFuture<TermVectorResponse> termVector(TermVectorRequest request) {
+        return internalClient.termVector(request);
+    }
+
     @Override
     public void termVector(TermVectorRequest request, ActionListener<TermVectorResponse> listener) {
         internalClient.termVector(request, listener);
     }
 
+    @Override
+    public ActionFuture<MultiTermVectorsResponse> multiTermVectors(final MultiTermVectorsRequest request) {
+        return internalClient.multiTermVectors(request);
+    }
+
+    @Override
+    public void multiTermVectors(final MultiTermVectorsRequest request, final ActionListener<MultiTermVectorsResponse> listener) {
+        internalClient.multiTermVectors(request, listener);
+    }
+
     @Override
     public ActionFuture<PercolateResponse> percolate(PercolateRequest request) {
         return internalClient.percolate(request);

src/main/java/org/elasticsearch/index/service/InternalIndexService.java

@@ -65,6 +65,7 @@ import org.elasticsearch.index.similarity.SimilarityService;
 import org.elasticsearch.index.store.IndexStore;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.store.StoreModule;
+import org.elasticsearch.index.termvectors.ShardTermVectorModule;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogModule;
 import org.elasticsearch.index.translog.TranslogService;
@@ -330,6 +331,7 @@ public class InternalIndexService extends AbstractIndexComponent implements Inde
         modules.add(new EngineModule(indexSettings));
         modules.add(new IndexShardGatewayModule(injector.getInstance(IndexGateway.class)));
         modules.add(new PercolatorShardModule());
+        modules.add(new ShardTermVectorModule());
 
         Injector shardInjector;
         try {

src/main/java/org/elasticsearch/index/shard/service/IndexShard.java

@@ -48,6 +48,7 @@ import org.elasticsearch.index.shard.DocsStats;
 import org.elasticsearch.index.shard.IndexShardComponent;
 import org.elasticsearch.index.shard.IndexShardState;
 import org.elasticsearch.index.store.StoreStats;
+import org.elasticsearch.index.termvectors.ShardTermVectorService;
 import org.elasticsearch.index.warmer.ShardIndexWarmerService;
 import org.elasticsearch.index.warmer.WarmerStats;
 import org.elasticsearch.search.suggest.completion.CompletionStats;
@@ -103,6 +104,8 @@ public interface IndexShard extends IndexShardComponent {
 
     ShardPercolateService shardPercolateService();
 
+    ShardTermVectorService termVectorService();
+
     IndexShardState state();
 
     Engine.Create prepareCreate(SourceToParse source) throws ElasticSearchException;

src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java

@@ -70,6 +70,7 @@ import org.elasticsearch.index.settings.IndexSettingsService;
 import org.elasticsearch.index.shard.*;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.store.StoreStats;
+import org.elasticsearch.index.termvectors.ShardTermVectorService;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.warmer.ShardIndexWarmerService;
 import org.elasticsearch.index.warmer.WarmerStats;
@@ -114,6 +115,7 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
     private final PercolatorQueriesRegistry percolatorQueriesRegistry;
     private final ShardPercolateService shardPercolateService;
     private final CodecService codecService;
+    private final ShardTermVectorService termVectorService;
 
     private final Object mutex = new Object();
     private final String checkIndexOnStartup;
@@ -138,7 +140,8 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
     public InternalIndexShard(ShardId shardId, @IndexSettings Settings indexSettings, IndexSettingsService indexSettingsService, IndicesLifecycle indicesLifecycle, Store store, Engine engine, MergeSchedulerProvider mergeScheduler, Translog translog,
                               ThreadPool threadPool, MapperService mapperService, IndexQueryParserService queryParserService, IndexCache indexCache, IndexAliasesService indexAliasesService, ShardIndexingService indexingService, ShardGetService getService, ShardSearchService searchService, ShardIndexWarmerService shardWarmerService,
                               ShardFilterCache shardFilterCache, ShardIdCache shardIdCache, ShardFieldData shardFieldData,
-                              PercolatorQueriesRegistry percolatorQueriesRegistry, ShardPercolateService shardPercolateService, CodecService codecService) {
+                              PercolatorQueriesRegistry percolatorQueriesRegistry, ShardPercolateService shardPercolateService, CodecService codecService,
+                              ShardTermVectorService termVectorService) {
         super(shardId, indexSettings);
         this.indicesLifecycle = (InternalIndicesLifecycle) indicesLifecycle;
         this.indexSettingsService = indexSettingsService;
@@ -153,6 +156,7 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
         this.indexAliasesService = indexAliasesService;
         this.indexingService = indexingService;
         this.getService = getService.setIndexShard(this);
+        this.termVectorService = termVectorService.setIndexShard(this);
         this.searchService = searchService;
         this.shardWarmerService = shardWarmerService;
         this.shardFilterCache = shardFilterCache;
@@ -198,6 +202,12 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
         return this.getService;
     }
 
+    @Override
+    public ShardTermVectorService termVectorService() {
+        return termVectorService;
+    }
+
+
     @Override
     public ShardSearchService searchService() {
         return this.searchService;

src/main/java/org/elasticsearch/index/termvectors/ShardTermVectorModule.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search 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.termvectors;
+
+import org.elasticsearch.common.inject.AbstractModule;
+
+/**
+ *
+ */
+public class ShardTermVectorModule extends AbstractModule {
+
+    @Override
+    protected void configure() {
+        bind(ShardTermVectorService.class).asEagerSingleton();
+    }
+}

src/main/java/org/elasticsearch/index/termvectors/ShardTermVectorService.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.termvectors;
+
+import org.apache.lucene.index.Fields;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.MultiFields;
+import org.apache.lucene.index.Term;
+import org.elasticsearch.ElasticSearchException;
+import org.elasticsearch.action.termvector.TermVectorRequest;
+import org.elasticsearch.action.termvector.TermVectorResponse;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.lucene.uid.Versions;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.mapper.Uid;
+import org.elasticsearch.index.mapper.internal.UidFieldMapper;
+import org.elasticsearch.index.settings.IndexSettings;
+import org.elasticsearch.index.shard.AbstractIndexShardComponent;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.shard.service.IndexShard;
+
+/**
+ */
+
+public class ShardTermVectorService extends AbstractIndexShardComponent {
+
+    private IndexShard indexShard;
+    private MapperService mapperService;
+
+    @Inject
+    public ShardTermVectorService(ShardId shardId, @IndexSettings Settings indexSettings, MapperService mapperService) {
+        super(shardId, indexSettings);
+    }
+
+    // sadly, to overcome cyclic dep, we need to do this and inject it ourselves...
+    public ShardTermVectorService setIndexShard(IndexShard indexShard) {
+        this.indexShard = indexShard;
+        return this;
+    }
+
+    public TermVectorResponse getTermVector(TermVectorRequest request) {
+        final Engine.Searcher searcher = indexShard.searcher();
+        IndexReader topLevelReader = searcher.reader();
+        final TermVectorResponse termVectorResponse = new TermVectorResponse(request.index(), request.type(), request.id());
+        final Term uidTerm = new Term(UidFieldMapper.NAME, Uid.createUidAsBytes(request.type(), request.id()));
+        try {
+            Fields topLevelFields = MultiFields.getFields(topLevelReader);
+            Versions.DocIdAndVersion docIdAndVersion = Versions.loadDocIdAndVersion(topLevelReader, uidTerm);
+            if (docIdAndVersion != null) {
+
+                Fields termVectorsByField = docIdAndVersion.context.reader().getTermVectors(docIdAndVersion.docId);
+                termVectorResponse.setFields(termVectorsByField, request.selectedFields(), request.getFlags(), topLevelFields);
+                termVectorResponse.setExists(true);
+                termVectorResponse.setDocVersion(docIdAndVersion.version);
+            } else {
+                termVectorResponse.setExists(false);
+            }
+        } catch (Throwable ex) {
+            throw new ElasticSearchException("failed to execute term vector request", ex);
+        } finally {
+            searcher.release();
+        }
+        return termVectorResponse;
+    }
+}

src/main/java/org/elasticsearch/rest/action/RestActionModule.java

@@ -90,6 +90,7 @@ import org.elasticsearch.rest.action.search.RestMultiSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchScrollAction;
 import org.elasticsearch.rest.action.suggest.RestSuggestAction;
+import org.elasticsearch.rest.action.termvector.RestMultiTermVectorsAction;
 import org.elasticsearch.rest.action.termvector.RestTermVectorAction;
 import org.elasticsearch.rest.action.update.RestUpdateAction;
 
@@ -176,6 +177,7 @@ public class RestActionModule extends AbstractModule {
         bind(RestCountAction.class).asEagerSingleton();
         bind(RestSuggestAction.class).asEagerSingleton();
         bind(RestTermVectorAction.class).asEagerSingleton();
+        bind(RestMultiTermVectorsAction.class).asEagerSingleton();
         bind(RestBulkAction.class).asEagerSingleton();
         bind(RestUpdateAction.class).asEagerSingleton();
         bind(RestPercolateAction.class).asEagerSingleton();

src/main/java/org/elasticsearch/rest/action/termvector/RestMultiTermVectorsAction.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.rest.action.termvector;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.termvector.MultiTermVectorsRequest;
+import org.elasticsearch.action.termvector.MultiTermVectorsResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.rest.*;
+
+import static org.elasticsearch.rest.RestRequest.Method.GET;
+import static org.elasticsearch.rest.RestRequest.Method.POST;
+import static org.elasticsearch.rest.RestStatus.OK;
+import static org.elasticsearch.rest.action.support.RestXContentBuilder.restContentBuilder;
+
+public class RestMultiTermVectorsAction extends BaseRestHandler {
+
+    @Inject
+    public RestMultiTermVectorsAction(Settings settings, Client client, RestController controller) {
+        super(settings, client);
+        controller.registerHandler(GET, "/_mtermvectors", this);
+        controller.registerHandler(POST, "/_mtermvectors", this);
+        controller.registerHandler(GET, "/{index}/_mtermvectors", this);
+        controller.registerHandler(POST, "/{index}/_mtermvectors", this);
+        controller.registerHandler(GET, "/{index}/{type}/_mtermvectors", this);
+        controller.registerHandler(POST, "/{index}/{type}/_mtermvectors", this);
+    }
+
+    @Override
+    public void handleRequest(final RestRequest request, final RestChannel channel) {
+        MultiTermVectorsRequest multiTermVectorsRequest = new MultiTermVectorsRequest();
+        multiTermVectorsRequest.listenerThreaded(false);
+        multiTermVectorsRequest.preference(request.param("preference"));
+
+        String[] sFields = null;
+        String sField = request.param("fields");
+        if (sField != null) {
+            sFields = Strings.splitStringByCommaToArray(sField);
+        }
+
+        try {
+            multiTermVectorsRequest.add(request.param("index"), request.param("type"), sFields, request.content());
+        } catch (Throwable t) {
+            try {
+                channel.sendResponse(new XContentThrowableRestResponse(request, t));
+            } catch (Throwable tIO) {
+                logger.error("Failed to send failure response", tIO);
+            }
+            return;
+        }
+
+        client.multiTermVectors(multiTermVectorsRequest, new ActionListener<MultiTermVectorsResponse>() {
+            @Override
+            public void onResponse(MultiTermVectorsResponse response) {
+                try {
+                    XContentBuilder builder = restContentBuilder(request);
+                    response.toXContent(builder, request);
+                    channel.sendResponse(new XContentRestResponse(request, OK, builder));
+                } catch (Throwable t) {
+                    onFailure(t);
+                }
+            }
+
+            @Override
+            public void onFailure(Throwable e) {
+                try {
+                    channel.sendResponse(new XContentThrowableRestResponse(request, e));
+                } catch (Throwable t) {
+                    logger.error("Failed to send failure response", t);
+                }
+            }
+        });
+    }
+}

src/test/java/org/elasticsearch/test/integration/termvectors/MultiTermVectorsTests.java

@@ -0,0 +1,64 @@
+package org.elasticsearch.test.integration.termvectors;
+/*
+ * 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.
+ */
+
+
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.Fields;
+import org.elasticsearch.action.termvector.MultiTermVectorsItemResponse;
+import org.elasticsearch.action.termvector.MultiTermVectorsRequestBuilder;
+import org.junit.Test;
+
+public class MultiTermVectorsTests extends AbstractTermVectorTests {
+
+    @Test
+    public void testDuelESLucene() throws Exception {
+        AbstractTermVectorTests.TestFieldSetting[] testFieldSettings = getFieldSettings();
+        createIndexBasedOnFieldSettings(testFieldSettings, -1);
+        AbstractTermVectorTests.TestDoc[] testDocs = generateTestDocs(5, testFieldSettings);
+
+        DirectoryReader directoryReader = indexDocsWithLucene(testDocs);
+        AbstractTermVectorTests.TestConfig[] testConfigs = generateTestConfigs(20, testDocs, testFieldSettings);
+
+        MultiTermVectorsRequestBuilder requestBuilder = client().prepareMultiTermVectors();
+        for (AbstractTermVectorTests.TestConfig test : testConfigs) {
+            requestBuilder.add(getRequestForConfig(test).request());
+        }
+
+        MultiTermVectorsItemResponse[] responseItems = run(requestBuilder).getResponses();
+
+        for (int i = 0; i < testConfigs.length; i++) {
+            TestConfig test = testConfigs[i];
+            try {
+                MultiTermVectorsItemResponse item = responseItems[i];
+                if (test.expectedException != null) {
+                    assertTrue(item.isFailed());
+                    continue;
+                } else if (item.isFailed()) {
+                    fail(item.getFailure().getMessage());
+                }
+                Fields luceneTermVectors = getTermVectorsFromLucene(directoryReader, test.doc);
+                validateResponse(item.getResponse(), luceneTermVectors, test);
+            } catch (Throwable t) {
+                throw new Exception("Test exception while running " + test.toString(), t);
+            }
+        }
+
+    }
+}