Added update by query api. #2230

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

@@ -150,6 +150,9 @@
 import org.elasticsearch.action.termvector.*;
 import org.elasticsearch.action.update.TransportUpdateAction;
 import org.elasticsearch.action.update.UpdateAction;
+import org.elasticsearch.action.updatebyquery.TransportUpdateByQueryAction;
+import org.elasticsearch.action.updatebyquery.TransportShardUpdateByQueryAction;
+import org.elasticsearch.action.updatebyquery.UpdateByQueryAction;
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.multibindings.MapBinder;
 
@@ -262,6 +265,8 @@ protected void configure() {
         registerAction(CountAction.INSTANCE, TransportCountAction.class);
         registerAction(SuggestAction.INSTANCE, TransportSuggestAction.class);
         registerAction(UpdateAction.INSTANCE, TransportUpdateAction.class);
+        registerAction(UpdateByQueryAction.INSTANCE, TransportUpdateByQueryAction.class,
+                TransportShardUpdateByQueryAction.class);
         registerAction(MultiGetAction.INSTANCE, TransportMultiGetAction.class,
                 TransportShardMultiGetAction.class);
         registerAction(BulkAction.INSTANCE, TransportBulkAction.class,

src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java

@@ -40,7 +40,7 @@ public class BulkShardRequest extends ShardReplicationOperationRequest<BulkShard
     BulkShardRequest() {
     }
 
-    BulkShardRequest(String index, int shardId, boolean refresh, BulkItemRequest[] items) {
+    public BulkShardRequest(String index, int shardId, boolean refresh, BulkItemRequest[] items) {
         this.index = index;
         this.shardId = shardId;
         this.items = items;

src/main/java/org/elasticsearch/action/bulk/BulkShardResponse.java

@@ -37,7 +37,7 @@ public class BulkShardResponse extends ActionResponse {
     BulkShardResponse() {
     }
 
-    BulkShardResponse(ShardId shardId, BulkItemResponse[] responses) {
+    public BulkShardResponse(ShardId shardId, BulkItemResponse[] responses) {
         this.shardId = shardId;
         this.responses = responses;
     }

src/main/java/org/elasticsearch/action/updatebyquery/BulkResponseOption.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.updatebyquery;
+
+import org.elasticsearch.ElasticsearchIllegalArgumentException;
+
+/**
+ * Specifies how bulk responses that have been created based on the documents that have matched with the update query
+ * are returned in the response.
+ */
+public enum BulkResponseOption {
+
+    /**
+     * Bulk responses aren't included in the response.
+     */
+    NONE((byte) 0),
+
+    /**
+     * Only failed bulk responses are included in the response.
+     */
+    FAILED((byte) 1),
+
+    /**
+     * All bulk responses are included in the response.
+     */
+    ALL((byte) 2);
+
+    private byte id;
+
+    private BulkResponseOption(byte id) {
+        this.id = id;
+    }
+
+    /**
+     * The internal representation of the operation type.
+     */
+    public byte id() {
+        return id;
+    }
+
+    /**
+     * Constructs the bulk response option from its internal representation.
+     */
+    public static BulkResponseOption fromId(byte id) {
+        if (id == 0) {
+            return NONE;
+        } else if (id == 1) {
+            return FAILED;
+        } else if (id == 2) {
+            return ALL;
+        } else {
+            throw new ElasticsearchIllegalArgumentException("No type match for [" + id + "]");
+        }
+    }
+
+    /**
+     * Parse the bulk response option from string.
+     */
+    public static BulkResponseOption fromString(String type) {
+        if ("none".equals(type)) {
+            return NONE;
+        } else if ("failed".equals(type)) {
+            return FAILED;
+        } else if ("all".equals(type)) {
+            return ALL;
+        }
+        throw new ElasticsearchIllegalArgumentException("no response type match for [" + type + "], should be either `none`, `failed` or `all`");
+    }
+
+}

src/main/java/org/elasticsearch/action/updatebyquery/IndexUpdateByQueryRequest.java

@@ -0,0 +1,128 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.updatebyquery;
+
+import com.google.common.collect.Sets;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.support.replication.IndexReplicationOperationRequest;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+import java.util.Set;
+
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+
+/**
+ * Represents an update by query request targeted for a specific index.
+ */
+public class IndexUpdateByQueryRequest extends IndexReplicationOperationRequest {
+
+    private String[] types = new String[0];
+    private BulkResponseOption bulkResponseOption;
+    private String[] filteringAliases = new String[0];
+    private Set<String> routing = Sets.newHashSet();
+
+    private BytesReference source;
+    private boolean sourceUnsafe;
+
+    IndexUpdateByQueryRequest() {
+    }
+
+    IndexUpdateByQueryRequest(UpdateByQueryRequest request, String index, String[] filteringAliases, Set<String> routing) {
+        this.replicationType = request.replicationType();
+        this.consistencyLevel = request.consistencyLevel();
+        this.timeout = request.timeout();
+        this.listenerThreaded(request.listenerThreaded());
+        this.index = index;
+        this.types = request.types();
+        this.bulkResponseOption = request.bulkResponseOptions();
+        this.source = request.source();
+        this.sourceUnsafe = request.sourceUnsafe();
+        if (filteringAliases != null) {
+            this.filteringAliases = filteringAliases;
+        }
+        if (routing != null) {
+            this.routing = routing;
+        }
+    }
+
+    public String[] types() {
+        return types;
+    }
+
+    public String[] filteringAliases() {
+        return filteringAliases;
+    }
+
+    public BulkResponseOption bulkResponseOptions() {
+        return bulkResponseOption;
+    }
+
+    public Set<String> routing() {
+        return routing;
+    }
+
+    public BytesReference source() {
+        return source;
+    }
+
+    public boolean sourceUnsafe() {
+        return sourceUnsafe;
+    }
+
+    public void beforeLocalFork() {
+        if (sourceUnsafe) {
+            source = source.copyBytesArray();
+        }
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = super.validate();
+        if (source == null) {
+            validationException = addValidationError("Source is missing", validationException);
+        }
+        return validationException;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        types = in.readStringArray();
+        bulkResponseOption = BulkResponseOption.fromId(in.readByte());
+        filteringAliases = in.readStringArray();
+        routing = Sets.newHashSet(in.readStringArray());
+        source = in.readBytesReference();
+        sourceUnsafe = false;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeStringArray(types);
+        out.writeByte(bulkResponseOption.id());
+        out.writeStringArray(filteringAliases);
+        out.writeStringArray(routing.toArray(new String[routing.size()]));
+        out.writeBytesReference(source);
+    }
+
+}