diff --git a/x-pack/docs/en/rest-api/rollup/rollup-search.asciidoc b/x-pack/docs/en/rest-api/rollup/rollup-search.asciidoc index f595d52ec10..115ef8fb043 100644 --- a/x-pack/docs/en/rest-api/rollup/rollup-search.asciidoc +++ b/x-pack/docs/en/rest-api/rollup/rollup-search.asciidoc @@ -101,6 +101,7 @@ GET /sensor_rollup/_rollup_search -------------------------------------------------- // CONSOLE // TEST[setup:sensor_prefab_data] +// TEST[s/_rollup_search/_rollup_search?filter_path=took,timed_out,terminated_early,_shards,hits,aggregations/] The query is targeting the `sensor_rollup` data, since this contains the rollup data as configured in the job. A `max` aggregation has been used on the `temperature` field, yielding the following response: @@ -194,6 +195,7 @@ GET sensor-1,sensor_rollup/_rollup_search <1> -------------------------------------------------- // CONSOLE // TEST[continued] +// TEST[s/_rollup_search/_rollup_search?filter_path=took,timed_out,terminated_early,_shards,hits,aggregations/] <1> Note the URI now searches `sensor-1` and `sensor_rollup` at the same time When the search is executed, the Rollup Search endpoint will do two things: diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java index 4042e98ef93..a38adf5d9de 100644 --- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java +++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java @@ -238,11 +238,23 @@ public class RollupResponseTranslator { ? (InternalAggregations)liveResponse.getAggregations() : InternalAggregations.EMPTY; - rolledResponses.forEach(r -> { - if (r == null || r.getAggregations() == null || r.getAggregations().asList().size() == 0) { - throw new RuntimeException("Expected to find aggregations in rollup response, but none found."); + int missingRollupAggs = rolledResponses.stream().mapToInt(searchResponse -> { + if (searchResponse == null + || searchResponse.getAggregations() == null + || searchResponse.getAggregations().asList().size() == 0) { + return 1; } - }); + return 0; + }).sum(); + + // We had no rollup aggs, so there is nothing to process + if (missingRollupAggs == rolledResponses.size()) { + // Return an empty response, but make sure we include all the shard, failure, etc stats + return mergeFinalResponse(liveResponse, rolledResponses, InternalAggregations.EMPTY); + } else if (missingRollupAggs > 0 && missingRollupAggs != rolledResponses.size()) { + // We were missing some but not all the aggs, unclear how to handle this. Bail. + throw new RuntimeException("Expected to find aggregations in rollup response, but none found."); + } // The combination process returns a tree that is identical to the non-rolled // which means we can use aggregation's reduce method to combine, just as if @@ -275,27 +287,39 @@ public class RollupResponseTranslator { new InternalAggregation.ReduceContext(reduceContext.bigArrays(), reduceContext.scriptService(), true)); } - // TODO allow profiling in the future - InternalSearchResponse combinedInternal = new InternalSearchResponse(SearchHits.empty(), currentTree, null, null, - rolledResponses.stream().anyMatch(SearchResponse::isTimedOut), - rolledResponses.stream().anyMatch(SearchResponse::isTimedOut), - rolledResponses.stream().mapToInt(SearchResponse::getNumReducePhases).sum()); + return mergeFinalResponse(liveResponse, rolledResponses, currentTree); + } + + private static SearchResponse mergeFinalResponse(SearchResponse liveResponse, List rolledResponses, + InternalAggregations aggs) { int totalShards = rolledResponses.stream().mapToInt(SearchResponse::getTotalShards).sum(); int sucessfulShards = rolledResponses.stream().mapToInt(SearchResponse::getSuccessfulShards).sum(); int skippedShards = rolledResponses.stream().mapToInt(SearchResponse::getSkippedShards).sum(); long took = rolledResponses.stream().mapToLong(r -> r.getTook().getMillis()).sum() ; + boolean isTimedOut = rolledResponses.stream().anyMatch(SearchResponse::isTimedOut); + boolean isTerminatedEarly = rolledResponses.stream() + .filter(r -> r.isTerminatedEarly() != null) + .anyMatch(SearchResponse::isTerminatedEarly); + int numReducePhases = rolledResponses.stream().mapToInt(SearchResponse::getNumReducePhases).sum(); + if (liveResponse != null) { totalShards += liveResponse.getTotalShards(); sucessfulShards += liveResponse.getSuccessfulShards(); skippedShards += liveResponse.getSkippedShards(); took = Math.max(took, liveResponse.getTook().getMillis()); + isTimedOut = isTimedOut && liveResponse.isTimedOut(); + isTerminatedEarly = isTerminatedEarly && liveResponse.isTerminatedEarly(); + numReducePhases += liveResponse.getNumReducePhases(); } + InternalSearchResponse combinedInternal = new InternalSearchResponse(SearchHits.empty(), aggs, null, null, + isTimedOut, isTerminatedEarly, numReducePhases); + // Shard failures are ignored atm, so returning an empty array is fine return new SearchResponse(combinedInternal, null, totalShards, sucessfulShards, skippedShards, - took, ShardSearchFailure.EMPTY_ARRAY, rolledResponses.get(0).getClusters()); + took, ShardSearchFailure.EMPTY_ARRAY, rolledResponses.get(0).getClusters()); } /** diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportRollupSearchAction.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportRollupSearchAction.java index c63ab96fa25..ea0319c3432 100644 --- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportRollupSearchAction.java +++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportRollupSearchAction.java @@ -155,6 +155,18 @@ public class TransportRollupSearchAction extends TransportAction validatedCaps = new HashSet<>(); sourceAgg.getAggregatorFactories() @@ -248,11 +260,6 @@ public class TransportRollupSearchAction extends TransportAction jobCaps) { diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java index 35d9f0d133a..73a4d0665c4 100644 --- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java +++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java @@ -198,10 +198,11 @@ public class RollupResponseTranslationTests extends AggregatorTestCase { BigArrays bigArrays = new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService()); ScriptService scriptService = mock(ScriptService.class); - Exception e = expectThrows(RuntimeException.class, - () -> RollupResponseTranslator.combineResponses(msearch, - new InternalAggregation.ReduceContext(bigArrays, scriptService, true))); - assertThat(e.getMessage(), equalTo("Expected to find aggregations in rollup response, but none found.")); + SearchResponse response = RollupResponseTranslator.combineResponses(msearch, + new InternalAggregation.ReduceContext(bigArrays, scriptService, true)); + assertNotNull(response); + Aggregations responseAggs = response.getAggregations(); + assertThat(responseAggs.asList().size(), equalTo(0)); } public void testMissingRolledIndex() { diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java index 069e23e4093..3cc6190db30 100644 --- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java +++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java @@ -307,21 +307,22 @@ public class SearchActionTests extends ESTestCase { assertThat(e.getMessage(), equalTo("Rollup search does not support explaining.")); } - public void testNoAgg() { - String[] normalIndices = new String[]{randomAlphaOfLength(10)}; + public void testNoRollupAgg() { + String[] normalIndices = new String[]{}; String[] rollupIndices = new String[]{randomAlphaOfLength(10)}; TransportRollupSearchAction.RollupSearchContext ctx = new TransportRollupSearchAction.RollupSearchContext(normalIndices, rollupIndices, Collections.emptySet()); SearchSourceBuilder source = new SearchSourceBuilder(); source.query(new MatchAllQueryBuilder()); source.size(0); - SearchRequest request = new SearchRequest(normalIndices, source); + SearchRequest request = new SearchRequest(rollupIndices, source); NamedWriteableRegistry registry = mock(NamedWriteableRegistry.class); - Exception e = expectThrows(IllegalArgumentException.class, - () -> TransportRollupSearchAction.createMSearchRequest(request, registry, ctx)); - assertThat(e.getMessage(), equalTo("Rollup requires at least one aggregation to be set.")); + MultiSearchRequest msearch = TransportRollupSearchAction.createMSearchRequest(request, registry, ctx); + assertThat(msearch.requests().size(), equalTo(1)); + assertThat(msearch.requests().get(0), equalTo(request)); } + public void testNoLiveNoRollup() { String[] normalIndices = new String[0]; String[] rollupIndices = new String[0]; diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/rollup/rollup_search.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/rollup/rollup_search.yml index d401d5c69ba..e2f1174665e 100644 --- a/x-pack/plugin/src/test/resources/rest-api-spec/test/rollup/rollup_search.yml +++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/rollup/rollup_search.yml @@ -152,6 +152,20 @@ setup: - match: { aggregations.histo.buckets.3.key_as_string: "2017-01-01T08:00:00.000Z" } - match: { aggregations.histo.buckets.3.doc_count: 20 } +--- +"Empty aggregation": + + - do: + xpack.rollup.rollup_search: + index: "foo_rollup" + body: + size: 0 + aggs: {} + + - length: { hits.hits: 0 } + - match: { hits.total: 0 } + - is_false: aggregations + --- "Search with Metric":