Skip to content

Commit

Permalink
HLRC: Add delete by query API (#32782)
Browse files Browse the repository at this point in the history 
Adds the delete-by-query API to the High Level REST Client.
  • Loading branch information
@sohaibiftikhar @nik9000
sohaibiftikhar authored and nik9000 committed Sep 4, 2018
1 parent 685dc33 commit f9466fd
Show file tree
Hide file tree
Showing 17 changed files with 547 additions and 60 deletions.
27 changes: 27 additions & 0 deletions client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.rankeval.RankEvalRequest;
import org.elasticsearch.index.reindex.AbstractBulkByScrollRequest;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.protocol.xpack.XPackInfoRequest;
Expand Down Expand Up @@ -877,6 +878,32 @@ static Request updateByQuery(UpdateByQueryRequest updateByQueryRequest) throws I
return request;
}

static Request deleteByQuery(DeleteByQueryRequest deleteByQueryRequest) throws IOException {
String endpoint =
endpoint(deleteByQueryRequest.indices(), deleteByQueryRequest.getDocTypes(), "_delete_by_query");
Request request = new Request(HttpPost.METHOD_NAME, endpoint);
Params params = new Params(request)
.withRouting(deleteByQueryRequest.getRouting())
.withRefresh(deleteByQueryRequest.isRefresh())
.withTimeout(deleteByQueryRequest.getTimeout())
.withWaitForActiveShards(deleteByQueryRequest.getWaitForActiveShards(), ActiveShardCount.DEFAULT)
.withIndicesOptions(deleteByQueryRequest.indicesOptions());
if (deleteByQueryRequest.isAbortOnVersionConflict() == false) {
params.putParam("conflicts", "proceed");
}
if (deleteByQueryRequest.getBatchSize() != AbstractBulkByScrollRequest.DEFAULT_SCROLL_SIZE) {
params.putParam("scroll_size", Integer.toString(deleteByQueryRequest.getBatchSize()));
}
if (deleteByQueryRequest.getScrollTime() != AbstractBulkByScrollRequest.DEFAULT_SCROLL_TIMEOUT) {
params.putParam("scroll", deleteByQueryRequest.getScrollTime());
}
if (deleteByQueryRequest.getSize() > 0) {
params.putParam("size", Integer.toString(deleteByQueryRequest.getSize()));
}
request.setEntity(createEntity(deleteByQueryRequest, 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
30 changes: 30 additions & 0 deletions client/rest-high-level/src/main/java/org/elasticsearch/client/RestHighLevelClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@
import org.elasticsearch.index.rankeval.RankEvalRequest;
import org.elasticsearch.index.rankeval.RankEvalResponse;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.plugins.spi.NamedXContentProvider;
Expand Down Expand Up @@ -474,6 +475,35 @@ public final void updateByQueryAsync(UpdateByQueryRequest reindexRequest, Reques
);
}

/**
* Executes a delete by query request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-delete-by-query.html">
* Delete By Query API on elastic.co</a>
* @param deleteByQueryRequest 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 deleteByQuery(DeleteByQueryRequest deleteByQueryRequest, RequestOptions options) throws IOException {
return performRequestAndParseEntity(
deleteByQueryRequest, RequestConverters::deleteByQuery, options, BulkByScrollResponse::fromXContent, emptySet()
);
}

/**
* Asynchronously executes a delete by query request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-delete-by-query.html">
* Delete 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 deleteByQueryAsync(DeleteByQueryRequest reindexRequest, RequestOptions options,
ActionListener<BulkByScrollResponse> listener) {
performRequestAsyncAndParseEntity(
reindexRequest, RequestConverters::deleteByQuery, 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
48 changes: 48 additions & 0 deletions client/rest-high-level/src/test/java/org/elasticsearch/client/CrudIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
Expand All @@ -50,6 +51,7 @@
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.query.IdsQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
Expand Down Expand Up @@ -823,6 +825,52 @@ public void testUpdateByQuery() throws IOException {
}
}

public void testDeleteByQuery() 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: delete one doc
DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest();
deleteByQueryRequest.indices(sourceIndex);
deleteByQueryRequest.setQuery(new IdsQueryBuilder().addIds("1").types("type"));
deleteByQueryRequest.setRefresh(true);
BulkByScrollResponse bulkResponse =
execute(deleteByQueryRequest, highLevelClient()::deleteByQuery, highLevelClient()::deleteByQueryAsync);
assertEquals(1, bulkResponse.getTotal());
assertEquals(1, 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(
1,
highLevelClient().search(new SearchRequest(sourceIndex), RequestOptions.DEFAULT).getHits().totalHits
);
}
}

public void testBulkProcessorIntegration() throws IOException {
int nbItems = randomIntBetween(10, 100);
boolean[] errors = new boolean[nbItems];
Expand Down
52 changes: 50 additions & 2 deletions client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,7 @@
import org.elasticsearch.index.rankeval.RankEvalSpec;
import org.elasticsearch.index.rankeval.RatedRequest;
import org.elasticsearch.index.rankeval.RestRankEvalAction;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.RemoteInfo;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
Expand Down Expand Up @@ -536,6 +537,53 @@ public void testUpdateByQuery() throws IOException {
assertToXContentBody(updateByQueryRequest, request.getEntity());
}

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

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

Expand Down Expand Up @@ -2754,7 +2802,7 @@ public void testXPackPutWatch() throws Exception {
request.getEntity().writeTo(bos);
assertThat(bos.toString("UTF-8"), is(body));
}

public void testGraphExplore() throws Exception {
Map<String, String> expectedParams = new HashMap<>();

Expand Down Expand Up @@ -2782,7 +2830,7 @@ public void testGraphExplore() throws Exception {
assertEquals(expectedParams, request.getParameters());
assertThat(request.getEntity().getContentType().getValue(), is(XContentType.JSON.mediaTypeWithoutParameters()));
assertToXContentBody(graphExploreRequest, request.getEntity());
}
}

public void testXPackDeleteWatch() {
DeleteWatchRequest deleteWatchRequest = new DeleteWatchRequest();
Expand Down
1 change: 0 additions & 1 deletion client/rest-high-level/src/test/java/org/elasticsearch/client/RestHighLevelClientTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -667,7 +667,6 @@ public void testApiNamingConventions() throws Exception {
"cluster.remote_info",
"count",
"create",
"delete_by_query",
"exists_source",
"get_source",
"indices.delete_alias",
Expand Down
108 changes: 108 additions & 0 deletions ...-high-level/src/test/java/org/elasticsearch/client/documentation/CRUDDocumentationIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.RemoteInfo;
import org.elasticsearch.index.reindex.ScrollableHitSource;
Expand Down Expand Up @@ -1030,6 +1031,113 @@ public void onFailure(Exception e) {
}
}

public void testDeleteByQuery() 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);
}
{
// tag::delete-by-query-request
DeleteByQueryRequest request = new DeleteByQueryRequest("source1", "source2"); // <1>
// end::delete-by-query-request
// tag::delete-by-query-request-conflicts
request.setConflicts("proceed"); // <1>
// end::delete-by-query-request-conflicts
// tag::delete-by-query-request-typeOrQuery
request.setDocTypes("doc"); // <1>
request.setQuery(new TermQueryBuilder("user", "kimchy")); // <2>
// end::delete-by-query-request-typeOrQuery
// tag::delete-by-query-request-size
request.setSize(10); // <1>
// end::delete-by-query-request-size
// tag::delete-by-query-request-scrollSize
request.setBatchSize(100); // <1>
// end::delete-by-query-request-scrollSize
// tag::delete-by-query-request-timeout
request.setTimeout(TimeValue.timeValueMinutes(2)); // <1>
// end::delete-by-query-request-timeout
// tag::delete-by-query-request-refresh
request.setRefresh(true); // <1>
// end::delete-by-query-request-refresh
// tag::delete-by-query-request-slices
request.setSlices(2); // <1>
// end::delete-by-query-request-slices
// tag::delete-by-query-request-scroll
request.setScroll(TimeValue.timeValueMinutes(10)); // <1>
// end::delete-by-query-request-scroll
// tag::delete-by-query-request-routing
request.setRouting("=cat"); // <1>
// end::delete-by-query-request-routing
// tag::delete-by-query-request-indicesOptions
request.setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN); // <1>
// end::delete-by-query-request-indicesOptions

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

// tag::delete-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::delete-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::delete-by-query-execute-async
client.deleteByQueryAsync(request, RequestOptions.DEFAULT, listener); // <1>
// end::delete-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 f9466fd

Please sign in to comment.