honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Multi GET API, closes #1068.

This commit is contained in:
kimchy 2011-06-27 22:23:49 +03:00
parent 475564449f
commit 9bf686ef62
24 changed files with 1549 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
modules/elasticsearch/src/main/java/org/elasticsearch/action/TransportActionModule.java
View File

@ -59,6 +59,8 @@ import org.elasticsearch.action.deletebyquery.TransportDeleteByQueryAction;
import org.elasticsearch.action.deletebyquery.TransportIndexDeleteByQueryAction;
import org.elasticsearch.action.deletebyquery.TransportShardDeleteByQueryAction;
import org.elasticsearch.action.get.TransportGetAction;
import org.elasticsearch.action.get.TransportMultiGetAction;
import org.elasticsearch.action.get.TransportShardMultiGetAction;
import org.elasticsearch.action.index.TransportIndexAction;
import org.elasticsearch.action.mlt.TransportMoreLikeThisAction;
import org.elasticsearch.action.percolate.TransportPercolateAction;
@ -124,6 +126,9 @@ public class TransportActionModule extends AbstractModule {
bind(TransportShardDeleteAction.class).asEagerSingleton();
bind(TransportCountAction.class).asEagerSingleton();
bind(TransportMultiGetAction.class).asEagerSingleton();
bind(TransportShardMultiGetAction.class).asEagerSingleton();
bind(TransportBulkAction.class).asEagerSingleton();
bind(TransportShardBulkAction.class).asEagerSingleton();

2
modules/elasticsearch/src/main/java/org/elasticsearch/action/TransportActions.java
View File

@ -36,6 +36,8 @@ public class TransportActions {
public static final String GET = "indices/get";
public static final String MULTI_GET = "indices/mget";
public static final String SEARCH = "indices/search";
public static final String SEARCH_SCROLL = "indices/searchScroll";

3
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/GetResponse.java
View File

@ -247,6 +247,7 @@ public class GetResponse implements ActionResponse, Streamable, Iterable<GetFiel
static final XContentBuilderString _TYPE = new XContentBuilderString("_type");
static final XContentBuilderString _ID = new XContentBuilderString("_id");
static final XContentBuilderString _VERSION = new XContentBuilderString("_version");
static final XContentBuilderString EXISTS = new XContentBuilderString("exists");
static final XContentBuilderString FIELDS = new XContentBuilderString("fields");
}
@ -256,6 +257,7 @@ public class GetResponse implements ActionResponse, Streamable, Iterable<GetFiel
builder.field(Fields._INDEX, index);
builder.field(Fields._TYPE, type);
builder.field(Fields._ID, id);
builder.field(Fields.EXISTS, false);
builder.endObject();
} else {
builder.startObject();
@ -265,6 +267,7 @@ public class GetResponse implements ActionResponse, Streamable, Iterable<GetFiel
if (version != -1) {
builder.field(Fields._VERSION, version);
}
builder.field(Fields.EXISTS, true);
if (source != null) {
RestXContentBuilder.restDocumentSource(source.bytes(), source.offset(), source.length(), builder, params);
}

162
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/MultiGetItemResponse.java Normal file
View File

@ -0,0 +1,162 @@
/*
* 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.action.get;
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 MultiGetItemResponse implements Streamable {
private GetResponse response;
private MultiGetResponse.Failure failure;
MultiGetItemResponse() {
}
public MultiGetItemResponse(GetResponse response, MultiGetResponse.Failure failure) {
this.response = response;
this.failure = failure;
}
/**
* The index name of the document.
*/
public String index() {
if (failure != null) {
return failure.index();
}
return response.index();
}
/**
* The index name of the document.
*/
public String getIndex() {
return index();
}
/**
* The type of the document.
*/
public String type() {
if (failure != null) {
return failure.type();
}
return response.type();
}
/**
* The type of the document.
*/
public String getType() {
return type();
}
/**
* The id of the document.
*/
public String id() {
if (failure != null) {
return failure.id();
}
return response.id();
}
/**
* The id of the document.
*/
public String getId() {
return id();
}
/**
* Is this a failed execution?
*/
public boolean failed() {
return failure != null;
}
/**
* Is this a failed execution?
*/
public boolean isFailed() {
return failed();
}
/**
* The actual get response, <tt>null</tt> if its a failure.
*/
public GetResponse response() {
return this.response;
}
/**
* The actual get response, <tt>null</tt> if its a failure.
*/
public GetResponse getResponse() {
return this.response;
}
/**
* The failure if relevant.
*/
public MultiGetResponse.Failure failure() {
return this.failure;
}
/**
* The failure if relevant.
*/
public MultiGetResponse.Failure getFailure() {
return failure();
}
public static MultiGetItemResponse readItemResponse(StreamInput in) throws IOException {
MultiGetItemResponse response = new MultiGetItemResponse();
response.readFrom(in);
return response;
}
@Override public void readFrom(StreamInput in) throws IOException {
if (in.readBoolean()) {
failure = MultiGetResponse.Failure.readFailure(in);
} else {
response = new GetResponse();
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);
}
}
}

291
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java Normal file
View File

@ -0,0 +1,291 @@
/*
* 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.action.get;
import org.elasticsearch.ElasticSearchIllegalArgumentException;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.Actions;
import org.elasticsearch.common.Nullable;
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.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
public class MultiGetRequest implements ActionRequest {
/**
* A single get item.
*/
public static class Item implements Streamable {
private String index;
private String type;
private String id;
private String routing;
Item() {
}
/**
* Constructs a single get item.
*
* @param index The index name
* @param type The type (can be null)
* @param id The id
*/
public Item(String index, @Nullable String type, String id) {
this.index = index;
this.type = type;
this.id = id;
}
public String index() {
return this.index;
}
public String type() {
return this.type;
}
public String id() {
return this.id;
}
/**
* The routing associated with this document.
*/
public Item routing(String routing) {
this.routing = routing;
return this;
}
public String routing() {
return this.routing;
}
public static Item readItem(StreamInput in) throws IOException {
Item item = new Item();
item.readFrom(in);
return item;
}
@Override public void readFrom(StreamInput in) throws IOException {
index = in.readUTF();
if (in.readBoolean()) {
type = in.readUTF();
}
id = in.readUTF();
if (in.readBoolean()) {
routing = in.readUTF();
}
}
@Override public void writeTo(StreamOutput out) throws IOException {
out.writeUTF(index);
if (type == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
out.writeUTF(type);
}
out.writeUTF(id);
if (routing == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
out.writeUTF(routing);
}
}
}
private boolean listenerThreaded = false;
String preference;
Boolean realtime;
boolean refresh;
List<Item> items = new ArrayList<Item>();
public MultiGetRequest add(Item item) {
items.add(item);
return this;
}
public MultiGetRequest add(String index, @Nullable String type, String id) {
items.add(new Item(index, type, id));
return this;
}
@Override public boolean listenerThreaded() {
return listenerThreaded;
}
@Override public MultiGetRequest listenerThreaded(boolean listenerThreaded) {
this.listenerThreaded = listenerThreaded;
return this;
}
@Override public ActionRequestValidationException validate() {
ActionRequestValidationException validationException = null;
if (items.isEmpty()) {
validationException = Actions.addValidationError("no documents to get", validationException);
} else {
for (int i = 0; i < items.size(); i++) {
Item item = items.get(i);
if (item.index() == null) {
validationException = Actions.addValidationError("index is missing for doc " + i, validationException);
}
if (item.id() == null) {
validationException = Actions.addValidationError("id is missing for doc " + i, validationException);
}
}
}
return validationException;
}
/**
* 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 MultiGetRequest preference(String preference) {
this.preference = preference;
return this;
}
public String preference() {
return this.preference;
}
public boolean realtime() {
return this.realtime == null ? true : this.realtime;
}
public MultiGetRequest realtime(Boolean realtime) {
this.realtime = realtime;
return this;
}
public boolean refresh() {
return this.refresh;
}
public MultiGetRequest refresh(boolean refresh) {
this.refresh = refresh;
return this;
}
public void add(@Nullable String defaultIndex, @Nullable String defaultType, byte[] data, int from, int length) throws Exception {
XContentParser parser = XContentFactory.xContent(data, from, length).createParser(data, from, length);
try {
XContentParser.Token token;
String currentFieldName = null;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} 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");
}
String index = defaultIndex;
String type = defaultType;
String id = null;
String routing = null;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else if (token.isValue()) {
if ("_index".equals(currentFieldName)) {
index = parser.text();
} else if ("_type".equals(currentFieldName)) {
type = parser.text();
} else if ("_id".equals(currentFieldName)) {
id = parser.text();
} else if ("_routing".equals(currentFieldName) || "routing".equals(currentFieldName)) {
routing = parser.text();
}
}
}
add(new Item(index, type, id).routing(routing));
}
} 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");
}
add(new Item(defaultIndex, defaultType, parser.text()));
}
}
}
}
} finally {
parser.close();
}
}
@Override public void readFrom(StreamInput in) throws IOException {
if (in.readBoolean()) {
preference = in.readUTF();
}
refresh = in.readBoolean();
byte realtime = in.readByte();
if (realtime == 0) {
this.realtime = false;
} else if (realtime == 1) {
this.realtime = true;
}
int size = in.readVInt();
items = new ArrayList<Item>(size);
for (int i = 0; i < size; i++) {
items.add(Item.readItem(in));
}
}
@Override public void writeTo(StreamOutput out) throws IOException {
if (preference == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
out.writeUTF(preference);
}
out.writeBoolean(refresh);
if (realtime == null) {
out.writeByte((byte) -1);
} else if (realtime == false) {
out.writeByte((byte) 0);
} else {
out.writeByte((byte) 1);
}
out.writeVInt(items.size());
for (Item item : items) {
item.writeTo(out);
}
}
}

200
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/MultiGetResponse.java Normal file
View File

@ -0,0 +1,200 @@
/*
* 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.action.get;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.common.collect.Iterators;
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 MultiGetResponse implements ActionResponse, Iterable<MultiGetItemResponse>, 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 index() {
return this.index;
}
/**
* The index name of the action.
*/
public String getIndex() {
return index();
}
/**
* The type of the action.
*/
public String type() {
return type;
}
/**
* The type of the action.
*/
public String getType() {
return type();
}
/**
* The id of the action.
*/
public String id() {
return id;
}
/**
* The id of the action.
*/
public String getId() {
return this.id;
}
/**
* The failure message.
*/
public String message() {
return this.message;
}
/**
* The failure message.
*/
public String getMessage() {
return 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.readUTF();
if (in.readBoolean()) {
type = in.readUTF();
}
id = in.readUTF();
message = in.readUTF();
}
@Override public void writeTo(StreamOutput out) throws IOException {
out.writeUTF(index);
if (type == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
out.writeUTF(type);
}
out.writeUTF(id);
out.writeUTF(message);
}
}
private MultiGetItemResponse[] responses;
MultiGetResponse() {
}
public MultiGetResponse(MultiGetItemResponse[] responses) {
this.responses = responses;
}
public MultiGetItemResponse[] responses() {
return this.responses;
}
@Override public Iterator<MultiGetItemResponse> iterator() {
return Iterators.forArray(responses);
}
@Override public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.startArray(Fields.DOCS);
for (MultiGetItemResponse response : responses) {
if (response.failed()) {
builder.startObject();
Failure failure = response.failure();
builder.field(Fields._INDEX, failure.index());
builder.field(Fields._TYPE, failure.type());
builder.field(Fields._ID, failure.id());
builder.field(Fields.ERROR, failure.message());
builder.endObject();
} else {
GetResponse 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 {
responses = new MultiGetItemResponse[in.readVInt()];
for (int i = 0; i < responses.length; i++) {
responses[i] = MultiGetItemResponse.readItemResponse(in);
}
}
@Override public void writeTo(StreamOutput out) throws IOException {
out.writeVInt(responses.length);
for (MultiGetItemResponse response : responses) {
response.writeTo(out);
}
}
}

154
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/MultiGetShardRequest.java Normal file
View File

@ -0,0 +1,154 @@
/*
* 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.action.get;
import org.elasticsearch.action.support.single.shard.SingleShardOperationRequest;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.trove.list.array.TIntArrayList;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
public class MultiGetShardRequest extends SingleShardOperationRequest {
private int shardId;
private String preference;
Boolean realtime;
boolean refresh;
TIntArrayList locations;
List<String> types;
List<String> ids;
MultiGetShardRequest() {
}
MultiGetShardRequest(String index, int shardId) {
super(index);
this.shardId = shardId;
locations = new TIntArrayList();
types = new ArrayList<String>();
ids = new ArrayList<String>();
}
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 MultiGetShardRequest preference(String preference) {
this.preference = preference;
return this;
}
public String preference() {
return this.preference;
}
public boolean realtime() {
return this.realtime == null ? true : this.realtime;
}
public MultiGetShardRequest realtime(Boolean realtime) {
this.realtime = realtime;
return this;
}
public boolean refresh() {
return this.refresh;
}
public MultiGetShardRequest refresh(boolean refresh) {
this.refresh = refresh;
return this;
}
public void add(int location, @Nullable String type, String id) {
locations.add(location);
types.add(type);
ids.add(id);
}
@Override public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
int size = in.readVInt();
locations = new TIntArrayList(size);
types = new ArrayList<String>(size);
ids = new ArrayList<String>(size);
for (int i = 0; i < size; i++) {
locations.add(in.readVInt());
if (in.readBoolean()) {
types.add(in.readUTF());
} else {
types.add(null);
}
ids.add(in.readUTF());
}
if (in.readBoolean()) {
preference = in.readUTF();
}
refresh = in.readBoolean();
byte realtime = in.readByte();
if (realtime == 0) {
this.realtime = false;
} else if (realtime == 1) {
this.realtime = true;
}
}
@Override public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
out.writeVInt(types.size());
for (int i = 0; i < types.size(); i++) {
out.writeVInt(locations.get(i));
if (types.get(i) == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
out.writeUTF(types.get(i));
}
out.writeUTF(ids.get(i));
}
if (preference == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
out.writeUTF(preference);
}
out.writeBoolean(refresh);
if (realtime == null) {
out.writeByte((byte) -1);
} else if (realtime == false) {
out.writeByte((byte) 0);
} else {
out.writeByte((byte) 1);
}
}
}

94
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/MultiGetShardResponse.java Normal file
View File

@ -0,0 +1,94 @@
/*
* 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.action.get;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.trove.list.array.TIntArrayList;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
public class MultiGetShardResponse implements ActionResponse {
TIntArrayList locations;
List<GetResponse> responses;
List<MultiGetResponse.Failure> failures;
MultiGetShardResponse() {
locations = new TIntArrayList();
responses = new ArrayList<GetResponse>();
failures = new ArrayList<MultiGetResponse.Failure>();
}
public void add(int location, GetResponse response) {
locations.add(location);
responses.add(response);
failures.add(null);
}
public void add(int location, MultiGetResponse.Failure failure) {
locations.add(location);
responses.add(null);
failures.add(failure);
}
@Override public void readFrom(StreamInput in) throws IOException {
int size = in.readVInt();
locations = new TIntArrayList(size);
responses = new ArrayList<GetResponse>(size);
failures = new ArrayList<MultiGetResponse.Failure>(size);
for (int i = 0; i < size; i++) {
locations.add(in.readVInt());
if (in.readBoolean()) {
GetResponse response = new GetResponse();
response.readFrom(in);
responses.add(response);
} else {
responses.add(null);
}
if (in.readBoolean()) {
failures.add(MultiGetResponse.Failure.readFailure(in));
} else {
failures.add(null);
}
}
}
@Override public void writeTo(StreamOutput out) throws IOException {
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 {
responses.get(i).writeTo(out);
}
if (failures.get(i) == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
failures.get(i).writeTo(out);
}
}
}
}

71
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/TransportGetAction.java
View File

@ -22,9 +22,6 @@ package org.elasticsearch.action.get;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.FieldSelector;
import org.apache.lucene.document.Fieldable;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.Term;
import org.apache.lucene.util.UnicodeUtil;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.TransportActions;
@ -35,12 +32,10 @@ import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.ShardIterator;
import org.elasticsearch.common.BytesHolder;
import org.elasticsearch.common.Unicode;
import org.elasticsearch.common.bloom.BloomFilter;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.lucene.uid.UidField;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.cache.bloom.BloomCache;
import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.mapper.DocumentMapper;
import org.elasticsearch.index.mapper.FieldMapper;
@ -122,50 +117,51 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
IndexService indexService = indicesService.indexServiceSafe(request.index());
IndexShard indexShard = indexService.shardSafe(shardId);
String type = null;
if (request.refresh() && !request.realtime()) {
indexShard.refresh(new Engine.Refresh(false));
}
return load(logger, scriptService, indexService, indexShard, request.index(), request.type(), request.id(), request.fields(), request.realtime());
}
public static GetResponse load(ESLogger logger, ScriptService scriptService, IndexService indexService, IndexShard indexShard, String index, String type, String id, String[] gFields, boolean realtime) throws ElasticSearchException {
Engine.GetResult get = null;
if (request.type() == null || request.type().equals("_all")) {
if (type == null || type.equals("_all")) {
for (String typeX : indexService.mapperService().types()) {
get = indexShard.get(new Engine.Get(request.realtime(), UidFieldMapper.TERM_FACTORY.createTerm(Uid.createUid(typeX, request.id()))));
get = indexShard.get(new Engine.Get(realtime, UidFieldMapper.TERM_FACTORY.createTerm(Uid.createUid(typeX, id))));
if (get.exists()) {
type = typeX;
break;
}
}
if (get == null || !get.exists()) {
return new GetResponse(request.index(), request.type(), request.id(), -1, false, null, null);
return new GetResponse(index, type, id, -1, false, null, null);
}
} else {
type = request.type();
get = indexShard.get(new Engine.Get(request.realtime(), UidFieldMapper.TERM_FACTORY.createTerm(Uid.createUid(type, request.id()))));
get = indexShard.get(new Engine.Get(realtime, UidFieldMapper.TERM_FACTORY.createTerm(Uid.createUid(type, id))));
if (!get.exists()) {
return new GetResponse(request.index(), request.type(), request.id(), -1, false, null, null);
return new GetResponse(index, type, id, -1, false, null, null);
}
}
DocumentMapper docMapper = indexService.mapperService().documentMapper(type);
if (docMapper == null) {
return new GetResponse(request.index(), request.type(), request.id(), -1, false, null, null);
return new GetResponse(index, type, id, -1, false, null, null);
}
if (request.refresh() && !request.realtime()) {
indexShard.refresh(new Engine.Refresh(false));
}
try {
// break between having loaded it from translog (so we only have _source), and having a document to load
if (get.docIdAndVersion() != null) {
Map<String, GetField> fields = null;
byte[] source = null;
UidField.DocIdAndVersion docIdAndVersion = get.docIdAndVersion();
FieldSelector fieldSelector = buildFieldSelectors(docMapper, request.fields());
FieldSelector fieldSelector = buildFieldSelectors(docMapper, gFields);
if (fieldSelector != null) {
Document doc;
try {
doc = docIdAndVersion.reader.document(docIdAndVersion.docId, fieldSelector);
} catch (IOException e) {
throw new ElasticSearchException("Failed to get type [" + request.type() + "] and id [" + request.id() + "]", e);
throw new ElasticSearchException("Failed to get type [" + type + "] and id [" + id + "]", e);
}
source = extractSource(doc, docMapper);
@ -203,9 +199,9 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
}
// now, go and do the script thingy if needed
if (request.fields() != null && request.fields().length > 0) {
if (gFields != null && gFields.length > 0) {
SearchLookup searchLookup = null;
for (String field : request.fields()) {
for (String field : gFields) {
String script = null;
if (field.contains("_source.") || field.contains("doc[")) {
script = field;
@ -244,7 +240,7 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
}
}
return new GetResponse(request.index(), request.type(), request.id(), get.version(), get.exists(), source == null ? null : new BytesHolder(source), fields);
return new GetResponse(index, type, id, get.version(), get.exists(), source == null ? null : new BytesHolder(source), fields);
} else {
BytesHolder source = get.source();
assert source != null;
@ -253,10 +249,10 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
boolean sourceRequested = false;
// we can only load scripts that can run against the source
if (request.fields() != null && request.fields().length > 0) {
if (gFields != null && gFields.length > 0) {
Map<String, Object> sourceAsMap = SourceLookup.sourceAsMap(source.bytes(), source.offset(), source.length());
SearchLookup searchLookup = null;
for (String field : request.fields()) {
for (String field : gFields) {
if (field.equals("_source")) {
sourceRequested = true;
continue;
@ -305,7 +301,7 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
sourceRequested = true;
}
return new GetResponse(request.index(), request.type(), request.id(), get.version(), get.exists(), sourceRequested ? source : null, fields);
return new GetResponse(index, type, id, get.version(), get.exists(), sourceRequested ? source : null, fields);
}
} finally {
if (get.searcher() != null) {
@ -314,7 +310,7 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
}
}
private FieldSelector buildFieldSelectors(DocumentMapper docMapper, String... fields) {
private static FieldSelector buildFieldSelectors(DocumentMapper docMapper, String... fields) {
if (fields == null) {
return docMapper.sourceMapper().fieldSelector();
}
@ -338,7 +334,7 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
return fieldSelector;
}
private byte[] extractSource(Document doc, DocumentMapper documentMapper) {
private static byte[] extractSource(Document doc, DocumentMapper documentMapper) {
byte[] source = null;
Fieldable sourceField = doc.getFieldable(documentMapper.sourceMapper().names().indexName());
if (sourceField != null) {
@ -348,23 +344,6 @@ public class TransportGetAction extends TransportShardSingleOperationAction<GetR
return source;
}
private UidField.DocIdAndVersion loadCurrentVersionFromIndex(BloomCache bloomCache, Engine.Searcher searcher, Term uid) {
UnicodeUtil.UTF8Result utf8 = Unicode.fromStringAsUtf8(uid.text());
for (IndexReader reader : searcher.searcher().subReaders()) {
BloomFilter filter = bloomCache.filter(reader, UidFieldMapper.NAME, true);
// we know that its not there...
if (!filter.isPresent(utf8.result, 0, utf8.length)) {
continue;
}
UidField.DocIdAndVersion docIdAndVersion = UidField.loadDocIdAndVersion(reader, uid);
// not null if it exists
if (docIdAndVersion != null) {
return docIdAndVersion;
}
}
return null;
}
@Override protected GetRequest newRequest() {
return new GetRequest();
}

142
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/TransportMultiGetAction.java Normal file
View File

@ -0,0 +1,142 @@
/*
* 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.action.get;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.TransportActions;
import org.elasticsearch.action.support.BaseAction;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
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 TransportMultiGetAction extends BaseAction<MultiGetRequest, MultiGetResponse> {
private final ClusterService clusterService;
private final TransportShardMultiGetAction shardAction;
@Inject public TransportMultiGetAction(Settings settings, ThreadPool threadPool, TransportService transportService, ClusterService clusterService, TransportShardMultiGetAction shardAction) {
super(settings, threadPool);
this.clusterService = clusterService;
this.shardAction = shardAction;
transportService.registerHandler(TransportActions.MULTI_GET, new TransportHandler());
}
@Override protected void doExecute(final MultiGetRequest request, final ActionListener<MultiGetResponse> listener) {
ClusterState clusterState = clusterService.state();
Map<ShardId, MultiGetShardRequest> shardRequests = new HashMap<ShardId, MultiGetShardRequest>();
for (int i = 0; i < request.items.size(); i++) {
MultiGetRequest.Item item = request.items.get(i);
ShardId shardId = clusterService.operationRouting()
.getShards(clusterState, item.index(), item.type(), item.id(), item.routing(), null).shardId();
MultiGetShardRequest shardRequest = shardRequests.get(shardId);
if (shardRequest == null) {
shardRequest = new MultiGetShardRequest(shardId.index().name(), shardId.id());
shardRequest.preference(request.preference);
shardRequest.realtime(request.realtime);
shardRequest.refresh(request.refresh);
shardRequests.put(shardId, shardRequest);
}
shardRequest.add(i, item.type(), item.id());
}
final MultiGetItemResponse[] responses = new MultiGetItemResponse[request.items.size()];
final AtomicInteger counter = new AtomicInteger(shardRequests.size());
for (final MultiGetShardRequest shardRequest : shardRequests.values()) {
shardAction.execute(shardRequest, new ActionListener<MultiGetShardResponse>() {
@Override public void onResponse(MultiGetShardResponse response) {
synchronized (responses) {
for (int i = 0; i < response.locations.size(); i++) {
responses[response.locations.get(i)] = new MultiGetItemResponse(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);
synchronized (responses) {
for (int i = 0; i < shardRequest.locations.size(); i++) {
responses[shardRequest.locations.get(i)] = new MultiGetItemResponse(null,
new MultiGetResponse.Failure(shardRequest.index(), shardRequest.types.get(i), shardRequest.ids.get(i), message));
}
}
if (counter.decrementAndGet() == 0) {
finishHim();
}
}
private void finishHim() {
listener.onResponse(new MultiGetResponse(responses));
}
});
}
}
class TransportHandler extends BaseTransportRequestHandler<MultiGetRequest> {
@Override public MultiGetRequest newInstance() {
return new MultiGetRequest();
}
@Override public void messageReceived(final MultiGetRequest 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<MultiGetResponse>() {
@Override public void onResponse(MultiGetResponse response) {
try {
channel.sendResponse(response);
} catch (Exception e) {
onFailure(e);
}
}
@Override public void onFailure(Throwable e) {
try {
channel.sendResponse(e);
} catch (Exception e1) {
logger.warn("Failed to send error response for action [" + TransportActions.MULTI_GET + "] and request [" + request + "]", e1);
}
}
});
}
@Override public String executor() {
return ThreadPool.Names.SAME;
}
}
}

117
modules/elasticsearch/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetAction.java Normal file
View File

@ -0,0 +1,117 @@
/*
* 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.action.get;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.single.shard.TransportShardSingleOperationAction;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
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.engine.Engine;
import org.elasticsearch.index.service.IndexService;
import org.elasticsearch.index.shard.service.IndexShard;
import org.elasticsearch.indices.IndicesService;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;
public class TransportShardMultiGetAction extends TransportShardSingleOperationAction<MultiGetShardRequest, MultiGetShardResponse> {
private final IndicesService indicesService;
private final ScriptService scriptService;
private final boolean realtime;
@Inject public TransportShardMultiGetAction(Settings settings, ClusterService clusterService, TransportService transportService,
IndicesService indicesService, ScriptService scriptService, ThreadPool threadPool) {
super(settings, threadPool, clusterService, transportService);
this.indicesService = indicesService;
this.scriptService = scriptService;
this.realtime = settings.getAsBoolean("action.get.realtime", true);
}
@Override protected String executor() {
return ThreadPool.Names.SEARCH;
}
@Override protected String transportAction() {
return "indices/mget/shard";
}
@Override protected String transportShardAction() {
return "indices/mget/shard/s";
}
@Override protected MultiGetShardRequest newRequest() {
return new MultiGetShardRequest();
}
@Override protected MultiGetShardResponse newResponse() {
return new MultiGetShardResponse();
}
@Override protected void checkBlock(MultiGetShardRequest request, ClusterState state) {
state.blocks().indexBlockedRaiseException(ClusterBlockLevel.READ, request.index());
}
@Override protected ShardIterator shards(ClusterState clusterState, MultiGetShardRequest request) {
return clusterService.operationRouting()
.getShards(clusterService.state(), request.index(), request.shardId(), request.preference());
}
@Override protected void doExecute(MultiGetShardRequest request, ActionListener<MultiGetShardResponse> listener) {
if (request.realtime == null) {
request.realtime = this.realtime;
}
super.doExecute(request, listener);
}
@Override protected MultiGetShardResponse shardOperation(MultiGetShardRequest request, int shardId) throws ElasticSearchException {
IndexService indexService = indicesService.indexServiceSafe(request.index());
IndexShard indexShard = indexService.shardSafe(shardId);
if (request.refresh() && !request.realtime()) {
indexShard.refresh(new Engine.Refresh(false));
}
MultiGetShardResponse response = new MultiGetShardResponse();
for (int i = 0; i < request.locations.size(); i++) {
String type = request.types.get(i);
String id = request.ids.get(i);
try {
GetResponse getResponse = TransportGetAction.load(logger, scriptService, indexService, indexShard, request.index(), type, id, null, request.realtime());
response.add(request.locations.get(i), getResponse);
} catch (Exception e) {
response.add(request.locations.get(i), new MultiGetResponse.Failure(request.index(), type, id, ExceptionsHelper.detailedMessage(e)));
}
}
return response;
}
}

18
modules/elasticsearch/src/main/java/org/elasticsearch/client/Client.java
View File

@ -31,6 +31,8 @@ import org.elasticsearch.action.deletebyquery.DeleteByQueryRequest;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.mlt.MoreLikeThisRequest;
@ -44,6 +46,7 @@ import org.elasticsearch.client.action.count.CountRequestBuilder;
import org.elasticsearch.client.action.delete.DeleteRequestBuilder;
import org.elasticsearch.client.action.deletebyquery.DeleteByQueryRequestBuilder;
import org.elasticsearch.client.action.get.GetRequestBuilder;
import org.elasticsearch.client.action.get.MultiGetRequestBuilder;
import org.elasticsearch.client.action.index.IndexRequestBuilder;
import org.elasticsearch.client.action.percolate.PercolateRequestBuilder;
import org.elasticsearch.client.action.search.SearchRequestBuilder;
@ -232,6 +235,21 @@ public interface Client {
*/
GetRequestBuilder prepareGet(String index, @Nullable String type, String id);
/**
* Multi get documents.
*/
ActionFuture<MultiGetResponse> multiGet(MultiGetRequest request);
/**
* Multi get documents.
*/
void multiGet(MultiGetRequest request, ActionListener<MultiGetResponse> listener);
/**
* Multi get documents.
*/
MultiGetRequestBuilder prepareMultiGet();
/**
* A count of all the documents matching a specific query.
*

86
modules/elasticsearch/src/main/java/org/elasticsearch/client/action/get/MultiGetRequestBuilder.java Normal file
View File

@ -0,0 +1,86 @@
/*
* 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.client.action.get;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.action.support.BaseRequestBuilder;
import org.elasticsearch.common.Nullable;
/**
* A multi get document action request builder.
*
* @author kimchy (shay.banon)
*/
public class MultiGetRequestBuilder extends BaseRequestBuilder<MultiGetRequest, MultiGetResponse> {
public MultiGetRequestBuilder(Client client) {
super(client, new MultiGetRequest());
}
public MultiGetRequestBuilder add(String index, @Nullable String type, String id) {
request.add(index, type, id);
return this;
}
public MultiGetRequestBuilder add(MultiGetRequest.Item item) {
request.add(item);
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 MultiGetRequestBuilder setPreference(String preference) {
request.preference(preference);
return this;
}
/**
* Should a refresh be executed before this get operation causing the operation to
* return the latest value. Note, heavy get should not set this to <tt>true</tt>. Defaults
* to <tt>false</tt>.
*/
public MultiGetRequestBuilder setRefresh(boolean refresh) {
request.refresh(refresh);
return this;
}
public MultiGetRequestBuilder setRealtime(Boolean realtime) {
request.realtime(realtime);
return this;
}
/**
* Should the listener be called on a separate thread if needed.
*/
public MultiGetRequestBuilder setListenerThreaded(boolean threadedListener) {
request.listenerThreaded(threadedListener);
return this;
}
@Override protected void doExecute(ActionListener<MultiGetResponse> listener) {
client.multiGet(request, listener);
}
}

22
modules/elasticsearch/src/main/java/org/elasticsearch/client/node/NodeClient.java
View File

@ -35,7 +35,10 @@ import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.deletebyquery.TransportDeleteByQueryAction;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.get.TransportGetAction;
import org.elasticsearch.action.get.TransportMultiGetAction;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.index.TransportIndexAction;
@ -44,7 +47,11 @@ import org.elasticsearch.action.mlt.TransportMoreLikeThisAction;
import org.elasticsearch.action.percolate.PercolateRequest;
import org.elasticsearch.action.percolate.PercolateResponse;
import org.elasticsearch.action.percolate.TransportPercolateAction;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.search.TransportSearchAction;
import org.elasticsearch.action.search.TransportSearchScrollAction;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.internal.InternalClient;
import org.elasticsearch.client.support.AbstractClient;
@ -71,6 +78,8 @@ public class NodeClient extends AbstractClient implements InternalClient {
private final TransportGetAction getAction;
private final TransportMultiGetAction multiGetAction;
private final TransportCountAction countAction;
private final TransportSearchAction searchAction;
@ -83,7 +92,7 @@ public class NodeClient extends AbstractClient implements InternalClient {
@Inject public NodeClient(Settings settings, ThreadPool threadPool, NodeAdminClient admin,
TransportIndexAction indexAction, TransportDeleteAction deleteAction, TransportBulkAction bulkAction,
TransportDeleteByQueryAction deleteByQueryAction, TransportGetAction getAction, TransportCountAction countAction,
TransportDeleteByQueryAction deleteByQueryAction, TransportGetAction getAction, TransportMultiGetAction multiGetAction, TransportCountAction countAction,
TransportSearchAction searchAction, TransportSearchScrollAction searchScrollAction,
TransportMoreLikeThisAction moreLikeThisAction, TransportPercolateAction percolateAction) {
this.threadPool = threadPool;
@ -93,6 +102,7 @@ public class NodeClient extends AbstractClient implements InternalClient {
this.bulkAction = bulkAction;
this.deleteByQueryAction = deleteByQueryAction;
this.getAction = getAction;
this.multiGetAction = multiGetAction;
this.countAction = countAction;
this.searchAction = searchAction;
this.searchScrollAction = searchScrollAction;
@ -152,6 +162,14 @@ public class NodeClient extends AbstractClient implements InternalClient {
getAction.execute(request, listener);
}
@Override public ActionFuture<MultiGetResponse> multiGet(MultiGetRequest request) {
return multiGetAction.execute(request);
}
@Override public void multiGet(MultiGetRequest request, ActionListener<MultiGetResponse> listener) {
multiGetAction.execute(request, listener);
}
@Override public ActionFuture<CountResponse> count(CountRequest request) {
return countAction.execute(request);
}

5
modules/elasticsearch/src/main/java/org/elasticsearch/client/support/AbstractClient.java
View File

@ -24,6 +24,7 @@ import org.elasticsearch.client.action.count.CountRequestBuilder;
import org.elasticsearch.client.action.delete.DeleteRequestBuilder;
import org.elasticsearch.client.action.deletebyquery.DeleteByQueryRequestBuilder;
import org.elasticsearch.client.action.get.GetRequestBuilder;
import org.elasticsearch.client.action.get.MultiGetRequestBuilder;
import org.elasticsearch.client.action.index.IndexRequestBuilder;
import org.elasticsearch.client.action.percolate.PercolateRequestBuilder;
import org.elasticsearch.client.action.search.SearchRequestBuilder;
@ -72,6 +73,10 @@ public abstract class AbstractClient implements InternalClient {
return prepareGet().setIndex(index).setType(type).setId(id);
}
@Override public MultiGetRequestBuilder prepareMultiGet() {
return new MultiGetRequestBuilder(this);
}
@Override public SearchRequestBuilder prepareSearch(String... indices) {
return new SearchRequestBuilder(this).setIndices(indices);
}

10
modules/elasticsearch/src/main/java/org/elasticsearch/client/transport/TransportClient.java
View File

@ -32,6 +32,8 @@ import org.elasticsearch.action.deletebyquery.DeleteByQueryRequest;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.mlt.MoreLikeThisRequest;
@ -284,6 +286,14 @@ public class TransportClient extends AbstractClient {
internalClient.get(request, listener);
}
@Override public ActionFuture<MultiGetResponse> multiGet(MultiGetRequest request) {
return internalClient.multiGet(request);
}
@Override public void multiGet(MultiGetRequest request, ActionListener<MultiGetResponse> listener) {
internalClient.multiGet(request, listener);
}
@Override public ActionFuture<CountResponse> count(CountRequest request) {
return internalClient.count(request);
}

2
modules/elasticsearch/src/main/java/org/elasticsearch/client/transport/action/ClientTransportActionModule.java
View File

@ -52,6 +52,7 @@ import org.elasticsearch.client.transport.action.count.ClientTransportCountActio
import org.elasticsearch.client.transport.action.delete.ClientTransportDeleteAction;
import org.elasticsearch.client.transport.action.deletebyquery.ClientTransportDeleteByQueryAction;
import org.elasticsearch.client.transport.action.get.ClientTransportGetAction;
import org.elasticsearch.client.transport.action.get.ClientTransportMultiGetAction;
import org.elasticsearch.client.transport.action.index.ClientTransportIndexAction;
import org.elasticsearch.client.transport.action.percolate.ClientTransportPercolateAction;
import org.elasticsearch.client.transport.action.search.ClientTransportSearchAction;
@ -68,6 +69,7 @@ public class ClientTransportActionModule extends AbstractModule {
bind(ClientTransportDeleteAction.class).asEagerSingleton();
bind(ClientTransportDeleteByQueryAction.class).asEagerSingleton();
bind(ClientTransportGetAction.class).asEagerSingleton();
bind(ClientTransportMultiGetAction.class).asEagerSingleton();
bind(ClientTransportCountAction.class).asEagerSingleton();
bind(ClientTransportSearchAction.class).asEagerSingleton();
bind(ClientTransportSearchScrollAction.class).asEagerSingleton();

42
modules/elasticsearch/src/main/java/org/elasticsearch/client/transport/action/get/ClientTransportMultiGetAction.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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.client.transport.action.get;
import org.elasticsearch.action.TransportActions;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.client.transport.action.support.BaseClientTransportAction;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.transport.TransportService;
/**
* @author kimchy (Shay Banon)
*/
public class ClientTransportMultiGetAction extends BaseClientTransportAction<MultiGetRequest, MultiGetResponse> {
@Inject public ClientTransportMultiGetAction(Settings settings, TransportService transportService) {
super(settings, transportService, MultiGetResponse.class);
}
@Override protected String action() {
return TransportActions.MULTI_GET;
}
}

25
modules/elasticsearch/src/main/java/org/elasticsearch/client/transport/support/InternalTransportClient.java
View File

@ -32,6 +32,8 @@ import org.elasticsearch.action.deletebyquery.DeleteByQueryRequest;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.mlt.MoreLikeThisRequest;
@ -49,6 +51,7 @@ import org.elasticsearch.client.transport.action.count.ClientTransportCountActio
import org.elasticsearch.client.transport.action.delete.ClientTransportDeleteAction;
import org.elasticsearch.client.transport.action.deletebyquery.ClientTransportDeleteByQueryAction;
import org.elasticsearch.client.transport.action.get.ClientTransportGetAction;
import org.elasticsearch.client.transport.action.get.ClientTransportMultiGetAction;
import org.elasticsearch.client.transport.action.index.ClientTransportIndexAction;
import org.elasticsearch.client.transport.action.mlt.ClientTransportMoreLikeThisAction;
import org.elasticsearch.client.transport.action.percolate.ClientTransportPercolateAction;
@ -78,6 +81,8 @@ public class InternalTransportClient extends AbstractClient implements InternalC
private final ClientTransportGetAction getAction;
private final ClientTransportMultiGetAction multiGetAction;
private final ClientTransportDeleteByQueryAction deleteByQueryAction;
private final ClientTransportCountAction countAction;
@ -92,7 +97,7 @@ public class InternalTransportClient extends AbstractClient implements InternalC
@Inject public InternalTransportClient(Settings settings, ThreadPool threadPool,
TransportClientNodesService nodesService, InternalTransportAdminClient adminClient,
ClientTransportIndexAction indexAction, ClientTransportDeleteAction deleteAction, ClientTransportBulkAction bulkAction, ClientTransportGetAction getAction,
ClientTransportIndexAction indexAction, ClientTransportDeleteAction deleteAction, ClientTransportBulkAction bulkAction, ClientTransportGetAction getAction, ClientTransportMultiGetAction multiGetAction,
ClientTransportDeleteByQueryAction deleteByQueryAction, ClientTransportCountAction countAction,
ClientTransportSearchAction searchAction, ClientTransportSearchScrollAction searchScrollAction,
ClientTransportMoreLikeThisAction moreLikeThisAction, ClientTransportPercolateAction percolateAction) {
@ -104,6 +109,7 @@ public class InternalTransportClient extends AbstractClient implements InternalC
this.deleteAction = deleteAction;
this.bulkAction = bulkAction;
this.getAction = getAction;
this.multiGetAction = multiGetAction;
this.deleteByQueryAction = deleteByQueryAction;
this.countAction = countAction;
this.searchAction = searchAction;
@ -209,6 +215,23 @@ public class InternalTransportClient extends AbstractClient implements InternalC
});
}
@Override public ActionFuture<MultiGetResponse> multiGet(final MultiGetRequest request) {
return nodesService.execute(new TransportClientNodesService.NodeCallback<ActionFuture<MultiGetResponse>>() {
@Override public ActionFuture<MultiGetResponse> doWithNode(DiscoveryNode node) throws ElasticSearchException {
return multiGetAction.execute(node, request);
}
});
}
@Override public void multiGet(final MultiGetRequest request, final ActionListener<MultiGetResponse> listener) {
nodesService.execute(new TransportClientNodesService.NodeCallback<Object>() {
@Override public Object doWithNode(DiscoveryNode node) throws ElasticSearchException {
multiGetAction.execute(node, request, listener);
return null;
}
});
}
@Override public ActionFuture<CountResponse> count(final CountRequest request) {
return nodesService.execute(new TransportClientNodesService.NodeCallback<ActionFuture<CountResponse>>() {
@Override public ActionFuture<CountResponse> doWithNode(DiscoveryNode node) throws ElasticSearchException {

2
modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/OperationRouting.java
View File

@ -42,6 +42,8 @@ public interface OperationRouting {
ShardIterator getShards(ClusterState clusterState, String index, String type, String id, @Nullable String routing, @Nullable String preference) throws IndexMissingException, IndexShardMissingException;
ShardIterator getShards(ClusterState clusterState, String index, int shardId, @Nullable String preference) throws IndexMissingException, IndexShardMissingException;
GroupShardsIterator deleteByQueryShards(ClusterState clusterState, String index, @Nullable Set<String> routing) throws IndexMissingException;
GroupShardsIterator searchShards(ClusterState clusterState, String[] indices, String[] concreteIndices, @Nullable String queryHint, @Nullable Map<String, Set<String>> routing, @Nullable String preference) throws IndexMissingException;

8
modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/plain/PlainOperationRouting.java
View File

@ -72,6 +72,10 @@ public class PlainOperationRouting extends AbstractComponent implements Operatio
return preferenceShardIterator(shards(clusterState, index, type, id, routing), clusterState.nodes().localNodeId(), preference);
}
@Override public ShardIterator getShards(ClusterState clusterState, String index, int shardId, @Nullable String preference) throws IndexMissingException, IndexShardMissingException {
return preferenceShardIterator(shards(clusterState, index, shardId), clusterState.nodes().localNodeId(), preference);
}
@Override public GroupShardsIterator broadcastDeleteShards(ClusterState clusterState, String index) throws IndexMissingException {
return indexRoutingTable(clusterState, index).groupByShardsIt();
}
@ -167,6 +171,10 @@ public class PlainOperationRouting extends AbstractComponent implements Operatio
protected IndexShardRoutingTable shards(ClusterState clusterState, String index, String type, String id, String routing) {
int shardId = shardId(clusterState, index, type, id, routing);
return shards(clusterState, index, shardId);
}
protected IndexShardRoutingTable shards(ClusterState clusterState, String index, int shardId) {
IndexShardRoutingTable indexShard = indexRoutingTable(clusterState, index).shard(shardId);
if (indexShard == null) {
throw new IndexShardMissingException(new ShardId(index, shardId));

2
modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/RestActionModule.java
View File

@ -59,6 +59,7 @@ import org.elasticsearch.rest.action.count.RestCountAction;
import org.elasticsearch.rest.action.delete.RestDeleteAction;
import org.elasticsearch.rest.action.deletebyquery.RestDeleteByQueryAction;
import org.elasticsearch.rest.action.get.RestGetAction;
import org.elasticsearch.rest.action.get.RestMultiGetAction;
import org.elasticsearch.rest.action.index.RestIndexAction;
import org.elasticsearch.rest.action.main.RestMainAction;
import org.elasticsearch.rest.action.mlt.RestMoreLikeThisAction;
@ -128,6 +129,7 @@ public class RestActionModule extends AbstractModule {
bind(RestIndexAction.class).asEagerSingleton();
bind(RestGetAction.class).asEagerSingleton();
bind(RestMultiGetAction.class).asEagerSingleton();
bind(RestDeleteAction.class).asEagerSingleton();

93
modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/get/RestMultiGetAction.java Normal file
View File

@ -0,0 +1,93 @@
/*
* 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.rest.action.get;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.rest.BaseRestHandler;
import org.elasticsearch.rest.RestChannel;
import org.elasticsearch.rest.RestController;
import org.elasticsearch.rest.RestRequest;
import org.elasticsearch.rest.XContentRestResponse;
import org.elasticsearch.rest.XContentThrowableRestResponse;
import java.io.IOException;
import static org.elasticsearch.rest.RestRequest.Method.*;
import static org.elasticsearch.rest.RestStatus.*;
import static org.elasticsearch.rest.action.support.RestXContentBuilder.*;
public class RestMultiGetAction extends BaseRestHandler {
@Inject public RestMultiGetAction(Settings settings, Client client, RestController controller) {
super(settings, client);
controller.registerHandler(GET, "/_mget", this);
controller.registerHandler(POST, "/_mget", this);
controller.registerHandler(GET, "/{index}/_mget", this);
controller.registerHandler(POST, "/{index}/_mget", this);
controller.registerHandler(GET, "/{index}/{type}/_mget", this);
controller.registerHandler(POST, "/{index}/{type}/_mget", this);
}
@Override public void handleRequest(final RestRequest request, final RestChannel channel) {
MultiGetRequest multiGetRequest = new MultiGetRequest();
multiGetRequest.listenerThreaded(false);
multiGetRequest.refresh(request.paramAsBoolean("refresh", multiGetRequest.refresh()));
multiGetRequest.preference(request.param("preference"));
multiGetRequest.realtime(request.paramAsBoolean("realtime", null));
try {
multiGetRequest.add(request.param("index"), request.param("type"), request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength());
} catch (Exception e) {
try {
XContentBuilder builder = restContentBuilder(request);
channel.sendResponse(new XContentRestResponse(request, BAD_REQUEST, builder.startObject().field("error", e.getMessage()).endObject()));
} catch (IOException e1) {
logger.error("Failed to send failure response", e1);
}
return;
}
client.multiGet(multiGetRequest, new ActionListener<MultiGetResponse>() {
@Override public void onResponse(MultiGetResponse response) {
try {
XContentBuilder builder = restContentBuilder(request);
response.toXContent(builder, request);
channel.sendResponse(new XContentRestResponse(request, OK, builder));
} catch (Exception e) {
onFailure(e);
}
}
@Override public void onFailure(Throwable e) {
try {
channel.sendResponse(new XContentThrowableRestResponse(request, e));
} catch (IOException e1) {
logger.error("Failed to send failure response", e1);
}
}
});
}
}

42
modules/test/integration/src/test/java/org/elasticsearch/test/integration/document/GetActionTests.java
View File

@ -23,6 +23,7 @@ import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.test.integration.AbstractNodesTests;
@ -135,4 +136,45 @@ public class GetActionTests extends AbstractNodesTests {
response = client.prepareGet("test", "type1", "1").execute().actionGet();
assertThat(response.exists(), equalTo(false));
}
@Test public void simpleMultiGetTests() throws Exception {
try {
client.admin().indices().prepareDelete("test").execute().actionGet();
} catch (Exception e) {
// fine
}
client.admin().indices().prepareCreate("test").setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1)).execute().actionGet();
ClusterHealthResponse clusterHealth = client.admin().cluster().health(clusterHealthRequest().waitForGreenStatus()).actionGet();
assertThat(clusterHealth.timedOut(), equalTo(false));
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
MultiGetResponse response = client.prepareMultiGet().add("test", "type1", "1").execute().actionGet();
assertThat(response.responses().length, equalTo(1));
assertThat(response.responses()[0].response().exists(), equalTo(false));
for (int i = 0; i < 10; i++) {
client.prepareIndex("test", "type1", Integer.toString(i)).setSource("field", "value" + i).execute().actionGet();
}
response = client.prepareMultiGet()
.add("test", "type1", "1")
.add("test", "type1", "15")
.add("test", "type1", "3")
.add("test", "type1", "9")
.add("test", "type1", "11")
.execute().actionGet();
assertThat(response.responses().length, equalTo(5));
assertThat(response.responses()[0].id(), equalTo("1"));
assertThat(response.responses()[0].response().exists(), equalTo(true));
assertThat(response.responses()[0].response().sourceAsMap().get("field").toString(), equalTo("value1"));
assertThat(response.responses()[1].id(), equalTo("15"));
assertThat(response.responses()[1].response().exists(), equalTo(false));
assertThat(response.responses()[2].id(), equalTo("3"));
assertThat(response.responses()[2].response().exists(), equalTo(true));
assertThat(response.responses()[3].id(), equalTo("9"));
assertThat(response.responses()[3].response().exists(), equalTo(true));
assertThat(response.responses()[4].id(), equalTo("11"));
assertThat(response.responses()[4].response().exists(), equalTo(false));
}
}