Skip to content

Commit

Permalink
HLRC: Add rollup search
Browse files Browse the repository at this point in the history
Relates to elastic#29827
  • Loading branch information
nik9000 committed Dec 6, 2018
1 parent e944764 commit fb6edda
Show file tree
Hide file tree
Showing 13 changed files with 257 additions and 10 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -354,8 +354,15 @@ static Request update(UpdateRequest updateRequest) throws IOException {
return request;
}

static Request search(SearchRequest searchRequest) throws IOException {
Request request = new Request(HttpPost.METHOD_NAME, endpoint(searchRequest.indices(), searchRequest.types(), "_search"));
/**
* Convert a {@linkplain SearchRequest} into a {@linkplain Request}.
* @param searchRequest the request to convert
* @param searchEndpoint the name of the search endpoint. {@literal _search}
* for standard searches and {@literal _rollup_search} for rollup
* searches.
*/
static Request search(SearchRequest searchRequest, String searchEndpoint) throws IOException {
Request request = new Request(HttpPost.METHOD_NAME, endpoint(searchRequest.indices(), searchRequest.types(), searchEndpoint));

Params params = new Params(request);
addSearchRequestParams(params, searchRequest);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -908,7 +908,12 @@ public final void deleteAsync(DeleteRequest deleteRequest, RequestOptions option
* @return the response
*/
public final SearchResponse search(SearchRequest searchRequest, RequestOptions options) throws IOException {
return performRequestAndParseEntity(searchRequest, RequestConverters::search, options, SearchResponse::fromXContent, emptySet());
return performRequestAndParseEntity(
searchRequest,
r -> RequestConverters.search(r, "_search"),
options,
SearchResponse::fromXContent,
emptySet());
}

/**
Expand All @@ -919,7 +924,12 @@ public final SearchResponse search(SearchRequest searchRequest, RequestOptions o
* @param listener the listener to be notified upon request completion
*/
public final void searchAsync(SearchRequest searchRequest, RequestOptions options, ActionListener<SearchResponse> listener) {
performRequestAsyncAndParseEntity(searchRequest, RequestConverters::search, options, SearchResponse::fromXContent, listener,
performRequestAsyncAndParseEntity(
searchRequest,
r -> RequestConverters.search(r, "_search"),
options,
SearchResponse::fromXContent,
listener,
emptySet());
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
package org.elasticsearch.client;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.core.AcknowledgedResponse;
import org.elasticsearch.client.rollup.DeleteRollupJobRequest;
import org.elasticsearch.client.rollup.GetRollupIndexCapsRequest;
Expand Down Expand Up @@ -224,6 +226,42 @@ public void getRollupJobAsync(GetRollupJobRequest request, RequestOptions option
listener, Collections.emptySet());
}

/**
* Perform a rollup search.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-search.html">
* the docs</a> for more.
* @param request 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 SearchResponse search(SearchRequest request, RequestOptions options) throws IOException {
return restHighLevelClient.performRequestAndParseEntity(
request,
RollupRequestConverters::search,
options,
SearchResponse::fromXContent,
Collections.emptySet());
}

/**
* Perform a rollup search.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-search.html">
* the docs</a> for more.
* @param request the request
* @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 void searchAsync(SearchRequest request, RequestOptions options, ActionListener<SearchResponse> listener) {
restHighLevelClient.performRequestAsyncAndParseEntity(
request,
RollupRequestConverters::search,
options,
SearchResponse::fromXContent,
listener,
Collections.emptySet());
}

/**
* Get the Rollup Capabilities of a target (non-rollup) index or pattern
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/master/rollup-get-rollup-caps.html">
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpPut;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.rollup.DeleteRollupJobRequest;
import org.elasticsearch.client.rollup.GetRollupCapsRequest;
import org.elasticsearch.client.rollup.GetRollupIndexCapsRequest;
Expand Down Expand Up @@ -93,6 +94,20 @@ static Request deleteJob(final DeleteRollupJobRequest deleteRollupJobRequest) th
return request;
}

static Request search(final SearchRequest request) throws IOException {
if (request.types().length > 0) {
/*
* Ideally we'd check this with the standard validation framework
* but we don't have a special request for rollup search so that'd
* be difficult.
*/
ValidationException ve = new ValidationException();
ve.addValidationError("types are not allowed in rollup search");
throw ve;
}
return RequestConverters.search(request, "_rollup_search");
}

static Request getRollupCaps(final GetRollupCapsRequest getRollupCapsRequest) throws IOException {
String endpoint = new RequestConverters.EndpointBuilder()
.addPathPartAsIs("_xpack", "rollup", "data")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -880,14 +880,16 @@ public void testGlobalPipelineOnBulkRequest() throws IOException {
}

public void testSearchNullSource() throws IOException {
String searchEndpoint = randomFrom("_" + randomAlphaOfLength(5));
SearchRequest searchRequest = new SearchRequest();
Request request = RequestConverters.search(searchRequest);
Request request = RequestConverters.search(searchRequest, searchEndpoint);
assertEquals(HttpPost.METHOD_NAME, request.getMethod());
assertEquals("/_search", request.getEndpoint());
assertEquals("/" + searchEndpoint, request.getEndpoint());
assertNull(request.getEntity());
}

public void testSearch() throws Exception {
String searchEndpoint = randomFrom("_" + randomAlphaOfLength(5));
String[] indices = randomIndicesNames(0, 5);
SearchRequest searchRequest = new SearchRequest(indices);

Expand Down Expand Up @@ -948,7 +950,7 @@ public void testSearch() throws Exception {
searchRequest.source(searchSourceBuilder);
}

Request request = RequestConverters.search(searchRequest);
Request request = RequestConverters.search(searchRequest, searchEndpoint);
StringJoiner endpoint = new StringJoiner("/", "/", "");
String index = String.join(",", indices);
if (Strings.hasLength(index)) {
Expand All @@ -958,7 +960,7 @@ public void testSearch() throws Exception {
if (Strings.hasLength(type)) {
endpoint.add(type);
}
endpoint.add("_search");
endpoint.add(searchEndpoint);
assertEquals(HttpPost.METHOD_NAME, request.getMethod());
assertEquals(endpoint.toString(), request.getEndpoint());
assertEquals(expectedParams, request.getParameters());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,11 +54,13 @@
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
import org.elasticsearch.search.aggregations.metrics.AvgAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.MaxAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.MinAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.Before;

import java.util.Arrays;
Expand All @@ -70,6 +72,7 @@
import java.util.Set;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.hamcrest.Matchers.closeTo;
import static org.hamcrest.Matchers.either;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
Expand Down Expand Up @@ -244,6 +247,33 @@ public void testPutStartAndGetRollupJob() throws Exception {
}
}

public void testSearch() throws Exception {
testPutStartAndGetRollupJob();
SearchRequest search = new SearchRequest(rollupIndex);
search.source(new SearchSourceBuilder()
.size(0)
.aggregation(new AvgAggregationBuilder("avg").field("value")));
SearchResponse response = highLevelClient().rollup().search(search, RequestOptions.DEFAULT);
assertEquals(0, response.getFailedShards());
assertEquals(0, response.getHits().getTotalHits().value);
NumericMetricsAggregation.SingleValue avg = response.getAggregations().get("avg");
assertThat(avg.value(), closeTo(sum / numDocs, 0.00000001));
}

public void testSearchWithType() throws Exception {
SearchRequest search = new SearchRequest(rollupIndex);
search.types("a", "b", "c");
search.source(new SearchSourceBuilder()
.size(0)
.aggregation(new AvgAggregationBuilder("avg").field("value")));
try {
highLevelClient().rollup().search(search, RequestOptions.DEFAULT);
fail("types are not allowed but didn't fail");
} catch (ValidationException e) {
assertEquals("Validation Failed: 1: types are not allowed in rollup search;", e.getMessage());
}
}

public void testGetMissingRollupJob() throws Exception {
GetRollupJobRequest getRollupJobRequest = new GetRollupJobRequest("missing");
RollupClient rollupClient = highLevelClient().rollup();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.client.ESRestHighLevelClientTestCase;
import org.elasticsearch.client.RequestOptions;
Expand Down Expand Up @@ -60,6 +62,9 @@
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
import org.elasticsearch.search.aggregations.metrics.MaxAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.Before;

import java.io.IOException;
Expand All @@ -72,6 +77,7 @@
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.hamcrest.Matchers.closeTo;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.isOneOf;
Expand All @@ -89,7 +95,7 @@ public void setUpDocs() throws IOException {
.field("timestamp", String.format(Locale.ROOT, "2018-01-01T00:%02d:00Z", i))
.field("hostname", 0)
.field("datacenter", 0)
.field("temperature", 0)
.field("temperature", i)
.field("voltage", 0)
.field("load", 0)
.field("net_in", 0)
Expand Down Expand Up @@ -330,6 +336,56 @@ public void onFailure(Exception e) {
assertTrue(latch.await(30L, TimeUnit.SECONDS));
}

public void testSearch() throws Exception {
// Setup a rollup index to query
testCreateRollupJob();

RestHighLevelClient client = highLevelClient();

// tag::search-request
SearchRequest request = new SearchRequest();
request.source(new SearchSourceBuilder()
.size(0)
.aggregation(new MaxAggregationBuilder("max_temperature")
.field("temperature")));
// end::search-request

// tag::search-execute
SearchResponse response =
client.rollup().search(request, RequestOptions.DEFAULT);
// end::search-execute

// tag::search-response
NumericMetricsAggregation.SingleValue maxTemperature =
response.getAggregations().get("max_temperature");
assertThat(maxTemperature.value(), closeTo(49.0, .00001));
// end::search-response-response

ActionListener<SearchResponse> listener;
// tag::search-execute-listener
listener = new ActionListener<SearchResponse>() {
@Override
public void onResponse(SearchResponse response) {
// <1>
}

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

final CountDownLatch latch = new CountDownLatch(1);
listener = new LatchedActionListener<>(listener, latch);

// tag::search-execute-async
client.rollup().searchAsync(request, RequestOptions.DEFAULT, listener); // <1>
// end::search-execute-async

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

@SuppressWarnings("unused")
public void testGetRollupCaps() throws Exception {
RestHighLevelClient client = highLevelClient();
Expand Down
45 changes: 45 additions & 0 deletions docs/java-rest/high-level/rollup/search.asciidoc
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
--
:api: search
:request: SearchRequest
:response: SearchResponse
--

[id="{upid}-{api}"]
=== Rollup Search API

The Rollup Search endpoint allows searching rolled-up data using the standard
query DSL. The Rollup Search endpoint is needed because, internally,
rolled-up documents utilize a different document structure than the original
data. The Rollup Search endpoint rewrites standard query DSL into a format that
matches the rollup documents, then takes the response and rewrites it back to
what a client would expect given the original query.

[id="{upid}-{api}-request"]
==== Request

Rollup Search uses the same +{request}+ that is used by the <<{mainid}-search>>
but it is mostly for aggregations you should set the `size` to 0 and add
aggregations like this:

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request]
--------------------------------------------------

NOTE:: Rollup Search is limited in many ways because only some query elements
can be translated into queries against the rollup indices. See the main
{ref}/rollup-search.html[Rollup Search] documentation for more.

include::../execution.asciidoc[]

[id="{upid}-{api}-response"]
==== Response

Rollup Search returns the same +{response}+ that is used by the
<<{mainid}-search>> and everything can be accessed in exactly the same way.
This will access the aggregation built by the example request above:

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-response]
--------------------------------------------------
2 changes: 2 additions & 0 deletions docs/java-rest/high-level/supported-apis.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -354,6 +354,7 @@ The Java High Level REST Client supports the following Rollup APIs:
* <<{upid}-rollup-stop-job>>
* <<{upid}-rollup-delete-job>>
* <<java-rest-high-x-pack-rollup-get-job>>
* <<{upid}-search>>
* <<{upid}-x-pack-rollup-get-rollup-caps>>
* <<{upid}-x-pack-rollup-get-rollup-index-caps>>

Expand All @@ -362,6 +363,7 @@ include::rollup/start_job.asciidoc[]
include::rollup/stop_job.asciidoc[]
include::rollup/delete_job.asciidoc[]
include::rollup/get_job.asciidoc[]
include::rollup/search.asciidoc[]
include::rollup/get_rollup_caps.asciidoc[]
include::rollup/get_rollup_index_caps.asciidoc[]

Expand Down
1 change: 1 addition & 0 deletions docs/reference/rollup/apis/rollup-search.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ and rewrites it back to what a client would expect given the original query.
indices.

Rules for the `index` parameter:

- At least one index/index-pattern must be specified. This can be either a rollup or non-rollup index. Omitting the index parameter,
or using `_all`, is not permitted
- Multiple non-rollup indices may be specified
Expand Down
Loading

0 comments on commit fb6edda

Please sign in to comment.