Skip to content

Commit

Permalink
HLREST: add update by query API (#32760)
Browse files Browse the repository at this point in the history
Adds update by query to the high level rest client.
  • Loading branch information
sohaibiftikhar authored and nik9000 committed Sep 2, 2018
1 parent 6770a45 commit 389bf67
Show file tree
Hide file tree
Showing 19 changed files with 691 additions and 81 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,9 @@
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.rankeval.RankEvalRequest;
import org.elasticsearch.index.reindex.AbstractBulkByScrollRequest;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.protocol.xpack.XPackInfoRequest;
import org.elasticsearch.protocol.xpack.XPackUsageRequest;
import org.elasticsearch.protocol.xpack.license.DeleteLicenseRequest;
Expand Down Expand Up @@ -837,6 +839,33 @@ static Request reindex(ReindexRequest reindexRequest) throws IOException {
return request;
}

static Request updateByQuery(UpdateByQueryRequest updateByQueryRequest) throws IOException {
String endpoint =
endpoint(updateByQueryRequest.indices(), updateByQueryRequest.getDocTypes(), "_update_by_query");
Request request = new Request(HttpPost.METHOD_NAME, endpoint);
Params params = new Params(request)
.withRouting(updateByQueryRequest.getRouting())
.withPipeline(updateByQueryRequest.getPipeline())
.withRefresh(updateByQueryRequest.isRefresh())
.withTimeout(updateByQueryRequest.getTimeout())
.withWaitForActiveShards(updateByQueryRequest.getWaitForActiveShards())
.withIndicesOptions(updateByQueryRequest.indicesOptions());
if (updateByQueryRequest.isAbortOnVersionConflict() == false) {
params.putParam("conflicts", "proceed");
}
if (updateByQueryRequest.getBatchSize() != AbstractBulkByScrollRequest.DEFAULT_SCROLL_SIZE) {
params.putParam("scroll_size", Integer.toString(updateByQueryRequest.getBatchSize()));
}
if (updateByQueryRequest.getScrollTime() != AbstractBulkByScrollRequest.DEFAULT_SCROLL_TIMEOUT) {
params.putParam("scroll", updateByQueryRequest.getScrollTime());
}
if (updateByQueryRequest.getSize() > 0) {
params.putParam("size", Integer.toString(updateByQueryRequest.getSize()));
}
request.setEntity(createEntity(updateByQueryRequest, REQUEST_BODY_CONTENT_TYPE));
return request;
}

static Request rollover(RolloverRequest rolloverRequest) throws IOException {
String endpoint = new EndpointBuilder().addPathPart(rolloverRequest.getAlias()).addPathPartAsIs("_rollover")
.addPathPart(rolloverRequest.getNewIndexName()).build();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@
import org.elasticsearch.index.rankeval.RankEvalResponse;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.plugins.spi.NamedXContentProvider;
import org.elasticsearch.rest.BytesRestResponse;
import org.elasticsearch.rest.RestStatus;
Expand Down Expand Up @@ -424,6 +425,35 @@ public final void reindexAsync(ReindexRequest reindexRequest, RequestOptions opt
);
}

/**
* Executes a update by query request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">
* Update By Query API on elastic.co</a>
* @param updateByQueryRequest the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @return the response
* @throws IOException in case there is a problem sending the request or parsing back the response
*/
public final BulkByScrollResponse updateByQuery(UpdateByQueryRequest updateByQueryRequest, RequestOptions options) throws IOException {
return performRequestAndParseEntity(
updateByQueryRequest, RequestConverters::updateByQuery, options, BulkByScrollResponse::fromXContent, emptySet()
);
}

/**
* Asynchronously executes an update by query request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">
* Update By Query API on elastic.co</a>
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @param listener the listener to be notified upon request completion
*/
public final void updateByQueryAsync(UpdateByQueryRequest reindexRequest, RequestOptions options,
ActionListener<BulkByScrollResponse> listener) {
performRequestAsyncAndParseEntity(
reindexRequest, RequestConverters::updateByQuery, options, BulkByScrollResponse::fromXContent, listener, emptySet()
);
}

/**
* Pings the remote Elasticsearch cluster and returns true if the ping succeeded, false otherwise
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
import org.elasticsearch.index.query.IdsQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
Expand Down Expand Up @@ -691,6 +692,72 @@ public void testReindex() throws IOException {
}
}

public void testUpdateByQuery() throws IOException {
final String sourceIndex = "source1";
{
// Prepare
Settings settings = Settings.builder()
.put("number_of_shards", 1)
.put("number_of_replicas", 0)
.build();
createIndex(sourceIndex, settings);
assertEquals(
RestStatus.OK,
highLevelClient().bulk(
new BulkRequest()
.add(new IndexRequest(sourceIndex, "type", "1")
.source(Collections.singletonMap("foo", 1), XContentType.JSON))
.add(new IndexRequest(sourceIndex, "type", "2")
.source(Collections.singletonMap("foo", 2), XContentType.JSON))
.setRefreshPolicy(RefreshPolicy.IMMEDIATE),
RequestOptions.DEFAULT
).status()
);
}
{
// test1: create one doc in dest
UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();
updateByQueryRequest.indices(sourceIndex);
updateByQueryRequest.setQuery(new IdsQueryBuilder().addIds("1").types("type"));
updateByQueryRequest.setRefresh(true);
BulkByScrollResponse bulkResponse =
execute(updateByQueryRequest, highLevelClient()::updateByQuery, highLevelClient()::updateByQueryAsync);
assertEquals(1, bulkResponse.getTotal());
assertEquals(1, bulkResponse.getUpdated());
assertEquals(0, bulkResponse.getNoops());
assertEquals(0, bulkResponse.getVersionConflicts());
assertEquals(1, bulkResponse.getBatches());
assertTrue(bulkResponse.getTook().getMillis() > 0);
assertEquals(1, bulkResponse.getBatches());
assertEquals(0, bulkResponse.getBulkFailures().size());
assertEquals(0, bulkResponse.getSearchFailures().size());
}
{
// test2: update using script
UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();
updateByQueryRequest.indices(sourceIndex);
updateByQueryRequest.setScript(new Script("if (ctx._source.foo == 2) ctx._source.foo++;"));
updateByQueryRequest.setRefresh(true);
BulkByScrollResponse bulkResponse =
execute(updateByQueryRequest, highLevelClient()::updateByQuery, highLevelClient()::updateByQueryAsync);
assertEquals(2, bulkResponse.getTotal());
assertEquals(2, bulkResponse.getUpdated());
assertEquals(0, bulkResponse.getDeleted());
assertEquals(0, bulkResponse.getNoops());
assertEquals(0, bulkResponse.getVersionConflicts());
assertEquals(1, bulkResponse.getBatches());
assertTrue(bulkResponse.getTook().getMillis() > 0);
assertEquals(1, bulkResponse.getBatches());
assertEquals(0, bulkResponse.getBulkFailures().size());
assertEquals(0, bulkResponse.getSearchFailures().size());
assertEquals(
3,
(int) (highLevelClient().get(new GetRequest(sourceIndex, "type", "2"), RequestOptions.DEFAULT)
.getSourceAsMap().get("foo"))
);
}
}

public void testBulkProcessorIntegration() throws IOException {
int nbItems = randomIntBetween(10, 100);
boolean[] errors = new boolean[nbItems];
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@
import org.elasticsearch.index.rankeval.RestRankEvalAction;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.RemoteInfo;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.protocol.xpack.XPackInfoRequest;
import org.elasticsearch.protocol.xpack.migration.IndexUpgradeInfoRequest;
import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
Expand All @@ -137,6 +138,7 @@
import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
import org.elasticsearch.repositories.fs.FsRepository;
import org.elasticsearch.rest.action.search.RestSearchAction;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.script.mustache.MultiSearchTemplateRequest;
import org.elasticsearch.script.mustache.SearchTemplateRequest;
Expand Down Expand Up @@ -470,6 +472,60 @@ public void testReindex() throws IOException {
assertToXContentBody(reindexRequest, request.getEntity());
}

public void testUpdateByQuery() throws IOException {
UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();
updateByQueryRequest.indices(randomIndicesNames(1, 5));
Map<String, String> expectedParams = new HashMap<>();
if (randomBoolean()) {
updateByQueryRequest.setDocTypes(generateRandomStringArray(5, 5, false, false));
}
if (randomBoolean()) {
int batchSize = randomInt(100);
updateByQueryRequest.setBatchSize(batchSize);
expectedParams.put("scroll_size", Integer.toString(batchSize));
}
if (randomBoolean()) {
updateByQueryRequest.setPipeline("my_pipeline");
expectedParams.put("pipeline", "my_pipeline");
}
if (randomBoolean()) {
updateByQueryRequest.setRouting("=cat");
expectedParams.put("routing", "=cat");
}
if (randomBoolean()) {
int size = randomIntBetween(100, 1000);
updateByQueryRequest.setSize(size);
expectedParams.put("size", Integer.toString(size));
}
if (randomBoolean()) {
updateByQueryRequest.setAbortOnVersionConflict(false);
expectedParams.put("conflicts", "proceed");
}
if (randomBoolean()) {
String ts = randomTimeValue();
updateByQueryRequest.setScroll(TimeValue.parseTimeValue(ts, "scroll"));
expectedParams.put("scroll", ts);
}
if (randomBoolean()) {
updateByQueryRequest.setQuery(new TermQueryBuilder("foo", "fooval"));
}
if (randomBoolean()) {
updateByQueryRequest.setScript(new Script("ctx._source.last = \"lastname\""));
}
setRandomIndicesOptions(updateByQueryRequest::setIndicesOptions, updateByQueryRequest::indicesOptions, expectedParams);
setRandomTimeout(updateByQueryRequest::setTimeout, ReplicationRequest.DEFAULT_TIMEOUT, expectedParams);
Request request = RequestConverters.updateByQuery(updateByQueryRequest);
StringJoiner joiner = new StringJoiner("/", "/", "");
joiner.add(String.join(",", updateByQueryRequest.indices()));
if (updateByQueryRequest.getDocTypes().length > 0)
joiner.add(String.join(",", updateByQueryRequest.getDocTypes()));
joiner.add("_update_by_query");
assertEquals(joiner.toString(), request.getEndpoint());
assertEquals(HttpPost.METHOD_NAME, request.getMethod());
assertEquals(expectedParams, request.getParameters());
assertToXContentBody(updateByQueryRequest, request.getEntity());
}

public void testPutMapping() throws IOException {
PutMappingRequest putMappingRequest = new PutMappingRequest();

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -664,8 +664,7 @@ public void testApiNamingConventions() throws Exception {
"render_search_template",
"scripts_painless_execute",
"tasks.get",
"termvectors",
"update_by_query"
"termvectors"
};
//These API are not required for high-level client feature completeness
String[] notRequiredApi = new String[] {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
import org.elasticsearch.action.support.replication.ReplicationResponse;
Expand Down Expand Up @@ -67,6 +68,7 @@
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.RemoteInfo;
import org.elasticsearch.index.reindex.ScrollableHitSource;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
Expand Down Expand Up @@ -899,6 +901,125 @@ public void onFailure(Exception e) {
}
}

public void testUpdateByQuery() throws Exception {
RestHighLevelClient client = highLevelClient();
{
String mapping =
"\"doc\": {\n" +
" \"properties\": {\n" +
" \"user\": {\n" +
" \"type\": \"text\"\n" +
" },\n" +
" \"field1\": {\n" +
" \"type\": \"integer\"\n" +
" },\n" +
" \"field2\": {\n" +
" \"type\": \"integer\"\n" +
" }\n" +
" }\n" +
" }";
createIndex("source1", Settings.EMPTY, mapping);
createIndex("source2", Settings.EMPTY, mapping);
createPipeline("my_pipeline");
}
{
// tag::update-by-query-request
UpdateByQueryRequest request = new UpdateByQueryRequest("source1", "source2"); // <1>
// end::update-by-query-request
// tag::update-by-query-request-conflicts
request.setConflicts("proceed"); // <1>
// end::update-by-query-request-conflicts
// tag::update-by-query-request-typeOrQuery
request.setDocTypes("doc"); // <1>
request.setQuery(new TermQueryBuilder("user", "kimchy")); // <2>
// end::update-by-query-request-typeOrQuery
// tag::update-by-query-request-size
request.setSize(10); // <1>
// end::update-by-query-request-size
// tag::update-by-query-request-scrollSize
request.setBatchSize(100); // <1>
// end::update-by-query-request-scrollSize
// tag::update-by-query-request-pipeline
request.setPipeline("my_pipeline"); // <1>
// end::update-by-query-request-pipeline
// tag::update-by-query-request-script
request.setScript(
new Script(
ScriptType.INLINE, "painless",
"if (ctx._source.user == 'kimchy') {ctx._source.likes++;}",
Collections.emptyMap())); // <1>
// end::update-by-query-request-script
// tag::update-by-query-request-timeout
request.setTimeout(TimeValue.timeValueMinutes(2)); // <1>
// end::update-by-query-request-timeout
// tag::update-by-query-request-refresh
request.setRefresh(true); // <1>
// end::update-by-query-request-refresh
// tag::update-by-query-request-slices
request.setSlices(2); // <1>
// end::update-by-query-request-slices
// tag::update-by-query-request-scroll
request.setScroll(TimeValue.timeValueMinutes(10)); // <1>
// end::update-by-query-request-scroll
// tag::update-by-query-request-routing
request.setRouting("=cat"); // <1>
// end::update-by-query-request-routing
// tag::update-by-query-request-indicesOptions
request.setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN); // <1>
// end::update-by-query-request-indicesOptions

// tag::update-by-query-execute
BulkByScrollResponse bulkResponse = client.updateByQuery(request, RequestOptions.DEFAULT);
// end::update-by-query-execute
assertSame(0, bulkResponse.getSearchFailures().size());
assertSame(0, bulkResponse.getBulkFailures().size());
// tag::update-by-query-response
TimeValue timeTaken = bulkResponse.getTook(); // <1>
boolean timedOut = bulkResponse.isTimedOut(); // <2>
long totalDocs = bulkResponse.getTotal(); // <3>
long updatedDocs = bulkResponse.getUpdated(); // <4>
long deletedDocs = bulkResponse.getDeleted(); // <5>
long batches = bulkResponse.getBatches(); // <6>
long noops = bulkResponse.getNoops(); // <7>
long versionConflicts = bulkResponse.getVersionConflicts(); // <8>
long bulkRetries = bulkResponse.getBulkRetries(); // <9>
long searchRetries = bulkResponse.getSearchRetries(); // <10>
TimeValue throttledMillis = bulkResponse.getStatus().getThrottled(); // <11>
TimeValue throttledUntilMillis = bulkResponse.getStatus().getThrottledUntil(); // <12>
List<ScrollableHitSource.SearchFailure> searchFailures = bulkResponse.getSearchFailures(); // <13>
List<BulkItemResponse.Failure> bulkFailures = bulkResponse.getBulkFailures(); // <14>
// end::update-by-query-response
}
{
UpdateByQueryRequest request = new UpdateByQueryRequest();
request.indices("source1");

// tag::update-by-query-execute-listener
ActionListener<BulkByScrollResponse> listener = new ActionListener<BulkByScrollResponse>() {
@Override
public void onResponse(BulkByScrollResponse bulkResponse) {
// <1>
}

@Override
public void onFailure(Exception e) {
// <2>
}
};
// end::update-by-query-execute-listener

// Replace the empty listener by a blocking listener in test
final CountDownLatch latch = new CountDownLatch(1);
listener = new LatchedActionListener<>(listener, latch);

// tag::update-by-query-execute-async
client.updateByQueryAsync(request, RequestOptions.DEFAULT, listener); // <1>
// end::update-by-query-execute-async

assertTrue(latch.await(30L, TimeUnit.SECONDS));
}
}

public void testGet() throws Exception {
RestHighLevelClient client = highLevelClient();
{
Expand Down
Loading

0 comments on commit 389bf67

Please sign in to comment.