From c2eec8c74a716e09dd3c32b1d75a256aa6f49eb6 Mon Sep 17 00:00:00 2001 From: amordleq <62701743+amordleq@users.noreply.github.com> Date: Mon, 20 Apr 2020 12:33:35 -0400 Subject: [PATCH] DATAES-567 - Add aggregation support to reactive client. (#430) Original PR: #430 --- .../DefaultReactiveElasticsearchClient.java | 23 ++- .../reactive/ReactiveElasticsearchClient.java | 41 +++++ .../client/util/NamedXContents.java | 163 ++++++++++++++++++ .../core/ReactiveElasticsearchTemplate.java | 39 +++++ .../core/ReactiveSearchOperations.java | 23 +++ .../ReactiveElasticsearchClientTests.java | 26 +++ .../ReactiveElasticsearchClientUnitTests.java | 82 +++++++++ .../ReactiveElasticsearchTemplateTests.java | 38 ++++ .../client/aggregate-ok-multiple-results.json | 35 ++++ .../client/aggregate-ok-no-results.json | 22 +++ .../client/aggregate-ok-single-result.json | 31 ++++ 11 files changed, 521 insertions(+), 2 deletions(-) create mode 100644 src/main/java/org/springframework/data/elasticsearch/client/util/NamedXContents.java create mode 100644 src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-multiple-results.json create mode 100644 src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-no-results.json create mode 100644 src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-single-result.json diff --git a/src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java b/src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java index b63861c1b..71936494c 100644 --- a/src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java +++ b/src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java @@ -94,6 +94,7 @@ import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.Scroll; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; +import org.elasticsearch.search.aggregations.Aggregation; import org.reactivestreams.Publisher; import org.springframework.data.elasticsearch.client.ClientConfiguration; import org.springframework.data.elasticsearch.client.ClientLogger; @@ -101,6 +102,7 @@ import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.NoReachableHostException; import org.springframework.data.elasticsearch.client.reactive.HostProvider.Verification; import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient.Indices; +import org.springframework.data.elasticsearch.client.util.NamedXContents; import org.springframework.data.util.Lazy; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; @@ -128,6 +130,7 @@ import org.springframework.web.reactive.function.client.WebClient.RequestBodySpe * @author Huw Ayling-Miller * @author Henrique Amaral * @author Roman Puchkovskiy + * @author Russell Parry * @since 3.2 * @see ClientConfiguration * @see ReactiveRestClients @@ -401,6 +404,23 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch .flatMap(Flux::fromIterable); } + /* + * (non-Javadoc) + * @see org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient#aggregate(org.springframework.http.HttpHeaders, org.elasticsearch.action.search.SearchRequest) + */ + @Override + public Flux aggregate(HttpHeaders headers, SearchRequest searchRequest) { + + Assert.notNull(headers, "headers must not be null"); + Assert.notNull(searchRequest, "searchRequest must not be null"); + + searchRequest.source().size(0); + + return sendRequest(searchRequest, requestCreator.search(), SearchResponse.class, headers) // + .map(SearchResponse::getAggregations) // + .flatMap(Flux::fromIterable); + } + /* * (non-Javadoc) * @see org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient#scroll(org.springframework.http.HttpHeaders, org.elasticsearch.action.search.SearchRequest) @@ -751,10 +771,9 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch } private static XContentParser createParser(String mediaType, String content) throws IOException { - return XContentType.fromMediaTypeOrFormat(mediaType) // .xContent() // - .createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, content); + .createParser(new NamedXContentRegistry(NamedXContents.getDefaultNamedXContents()), DeprecationHandler.THROW_UNSUPPORTED_OPERATION, content); } private static Publisher handleServerError(Request request, ClientResponse response) { diff --git a/src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java b/src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java index 914a43395..708fba9df 100644 --- a/src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java +++ b/src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java @@ -15,6 +15,8 @@ */ package org.springframework.data.elasticsearch.client.reactive; +import org.elasticsearch.search.aggregations.Aggregation; +import org.springframework.util.Assert; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -411,6 +413,45 @@ public interface ReactiveElasticsearchClient { */ Flux search(HttpHeaders headers, SearchRequest searchRequest); + /** + * Execute the given {@link SearchRequest} with aggregations against the {@literal search} API. + * + * @param consumer never {@literal null}. + * @see Search API on + * elastic.co + * @return the {@link Flux} emitting {@link Aggregation} one by one. + * @since 4.0 + */ + default Flux aggregate(Consumer consumer) { + Assert.notNull(consumer, "consumer must not be null"); + SearchRequest request = new SearchRequest(); + consumer.accept(request); + return aggregate(request); + } + + /** + * Execute the given {@link SearchRequest} with aggregations against the {@literal search} API. + * + * @param searchRequest must not be {@literal null}. + * @see Search API on + * elastic.co + * @return the {@link Flux} emitting {@link Aggregation} one by one. + * @since 4.0 + */ + default Flux aggregate(SearchRequest searchRequest) { return aggregate(HttpHeaders.EMPTY, searchRequest); } + + /** + * Execute the given {@link SearchRequest} with aggregations against the {@literal search} API. + * + * @param headers Use {@link HttpHeaders} to provide eg. authentication data. Must not be {@literal null}. + * @param searchRequest must not be {@literal null}. + * @see Search API on + * elastic.co + * @return the {@link Flux} emitting {@link Aggregation} one by one. + * @since 4.0 + */ + Flux aggregate(HttpHeaders headers, SearchRequest searchRequest); + /** * Execute the given {@link SearchRequest} against the {@literal search scroll} API. * diff --git a/src/main/java/org/springframework/data/elasticsearch/client/util/NamedXContents.java b/src/main/java/org/springframework/data/elasticsearch/client/util/NamedXContents.java new file mode 100644 index 000000000..edf5f1f41 --- /dev/null +++ b/src/main/java/org/springframework/data/elasticsearch/client/util/NamedXContents.java @@ -0,0 +1,163 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.data.elasticsearch.client.util; + +import org.elasticsearch.common.ParseField; +import org.elasticsearch.common.xcontent.ContextParser; +import org.elasticsearch.common.xcontent.NamedXContentRegistry; +import org.elasticsearch.search.aggregations.Aggregation; +import org.elasticsearch.search.aggregations.bucket.adjacency.AdjacencyMatrixAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.adjacency.ParsedAdjacencyMatrix; +import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.composite.ParsedComposite; +import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.filter.FiltersAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter; +import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilters; +import org.elasticsearch.search.aggregations.bucket.geogrid.GeoHashGridAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.geogrid.GeoTileGridAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.geogrid.ParsedGeoHashGrid; +import org.elasticsearch.search.aggregations.bucket.geogrid.ParsedGeoTileGrid; +import org.elasticsearch.search.aggregations.bucket.global.GlobalAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.global.ParsedGlobal; +import org.elasticsearch.search.aggregations.bucket.histogram.AutoDateHistogramAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.histogram.ParsedAutoDateHistogram; +import org.elasticsearch.search.aggregations.bucket.histogram.ParsedDateHistogram; +import org.elasticsearch.search.aggregations.bucket.histogram.ParsedHistogram; +import org.elasticsearch.search.aggregations.bucket.missing.MissingAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.missing.ParsedMissing; +import org.elasticsearch.search.aggregations.bucket.nested.NestedAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.nested.ParsedNested; +import org.elasticsearch.search.aggregations.bucket.nested.ParsedReverseNested; +import org.elasticsearch.search.aggregations.bucket.nested.ReverseNestedAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.range.DateRangeAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.range.GeoDistanceAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.range.IpRangeAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.range.ParsedBinaryRange; +import org.elasticsearch.search.aggregations.bucket.range.ParsedDateRange; +import org.elasticsearch.search.aggregations.bucket.range.ParsedGeoDistance; +import org.elasticsearch.search.aggregations.bucket.range.ParsedRange; +import org.elasticsearch.search.aggregations.bucket.range.RangeAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.sampler.InternalSampler; +import org.elasticsearch.search.aggregations.bucket.sampler.ParsedSampler; +import org.elasticsearch.search.aggregations.bucket.significant.ParsedSignificantLongTerms; +import org.elasticsearch.search.aggregations.bucket.significant.ParsedSignificantStringTerms; +import org.elasticsearch.search.aggregations.bucket.significant.SignificantLongTerms; +import org.elasticsearch.search.aggregations.bucket.significant.SignificantStringTerms; +import org.elasticsearch.search.aggregations.bucket.terms.DoubleTerms; +import org.elasticsearch.search.aggregations.bucket.terms.LongTerms; +import org.elasticsearch.search.aggregations.bucket.terms.ParsedDoubleTerms; +import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms; +import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms; +import org.elasticsearch.search.aggregations.bucket.terms.StringTerms; +import org.elasticsearch.search.aggregations.metrics.*; +import org.elasticsearch.search.aggregations.pipeline.*; +import org.elasticsearch.search.suggest.Suggest; +import org.elasticsearch.search.suggest.completion.CompletionSuggestion; +import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder; +import org.elasticsearch.search.suggest.phrase.PhraseSuggestion; +import org.elasticsearch.search.suggest.phrase.PhraseSuggestionBuilder; +import org.elasticsearch.search.suggest.term.TermSuggestion; +import org.elasticsearch.search.suggest.term.TermSuggestionBuilder; +import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + *

+ * Original implementation source {@link org.elasticsearch.client.RestHighLevelClient#getDefaultNamedXContents()} by {@literal Elasticsearch} + * (https://www.elastic.co) licensed under the Apache License, Version 2.0. + *

+ * Modified for usage with {@link ReactiveElasticsearchClient}. + *

+ * Only intended for internal use. + * + * @author Russell Parry + * @since 4.0 + */ +public class NamedXContents { + + private NamedXContents() { + // contains only utility methods + } + + public static List getDefaultNamedXContents() { + Map> map = new HashMap<>(); + map.put(CardinalityAggregationBuilder.NAME, (p, c) -> ParsedCardinality.fromXContent(p, (String) c)); + map.put(InternalHDRPercentiles.NAME, (p, c) -> ParsedHDRPercentiles.fromXContent(p, (String) c)); + map.put(InternalHDRPercentileRanks.NAME, (p, c) -> ParsedHDRPercentileRanks.fromXContent(p, (String) c)); + map.put(InternalTDigestPercentiles.NAME, (p, c) -> ParsedTDigestPercentiles.fromXContent(p, (String) c)); + map.put(InternalTDigestPercentileRanks.NAME, (p, c) -> ParsedTDigestPercentileRanks.fromXContent(p, (String) c)); + map.put(PercentilesBucketPipelineAggregationBuilder.NAME, (p, c) -> ParsedPercentilesBucket.fromXContent(p, (String) c)); + map.put(MedianAbsoluteDeviationAggregationBuilder.NAME, (p, c) -> ParsedMedianAbsoluteDeviation.fromXContent(p, (String) c)); + map.put(MinAggregationBuilder.NAME, (p, c) -> ParsedMin.fromXContent(p, (String) c)); + map.put(MaxAggregationBuilder.NAME, (p, c) -> ParsedMax.fromXContent(p, (String) c)); + map.put(SumAggregationBuilder.NAME, (p, c) -> ParsedSum.fromXContent(p, (String) c)); + map.put(AvgAggregationBuilder.NAME, (p, c) -> ParsedAvg.fromXContent(p, (String) c)); + map.put(WeightedAvgAggregationBuilder.NAME, (p, c) -> ParsedWeightedAvg.fromXContent(p, (String) c)); + map.put(ValueCountAggregationBuilder.NAME, (p, c) -> ParsedValueCount.fromXContent(p, (String) c)); + map.put(InternalSimpleValue.NAME, (p, c) -> ParsedSimpleValue.fromXContent(p, (String) c)); + map.put(DerivativePipelineAggregationBuilder.NAME, (p, c) -> ParsedDerivative.fromXContent(p, (String) c)); + map.put(InternalBucketMetricValue.NAME, (p, c) -> ParsedBucketMetricValue.fromXContent(p, (String) c)); + map.put(StatsAggregationBuilder.NAME, (p, c) -> ParsedStats.fromXContent(p, (String) c)); + map.put(StatsBucketPipelineAggregationBuilder.NAME, (p, c) -> ParsedStatsBucket.fromXContent(p, (String) c)); + map.put(ExtendedStatsAggregationBuilder.NAME, (p, c) -> ParsedExtendedStats.fromXContent(p, (String) c)); + map.put(ExtendedStatsBucketPipelineAggregationBuilder.NAME, + (p, c) -> ParsedExtendedStatsBucket.fromXContent(p, (String) c)); + map.put(GeoBoundsAggregationBuilder.NAME, (p, c) -> ParsedGeoBounds.fromXContent(p, (String) c)); + map.put(GeoCentroidAggregationBuilder.NAME, (p, c) -> ParsedGeoCentroid.fromXContent(p, (String) c)); + map.put(HistogramAggregationBuilder.NAME, (p, c) -> ParsedHistogram.fromXContent(p, (String) c)); + map.put(DateHistogramAggregationBuilder.NAME, (p, c) -> ParsedDateHistogram.fromXContent(p, (String) c)); + map.put(AutoDateHistogramAggregationBuilder.NAME, (p, c) -> ParsedAutoDateHistogram.fromXContent(p, (String) c)); + map.put(StringTerms.NAME, (p, c) -> ParsedStringTerms.fromXContent(p, (String) c)); + map.put(LongTerms.NAME, (p, c) -> ParsedLongTerms.fromXContent(p, (String) c)); + map.put(DoubleTerms.NAME, (p, c) -> ParsedDoubleTerms.fromXContent(p, (String) c)); + map.put(MissingAggregationBuilder.NAME, (p, c) -> ParsedMissing.fromXContent(p, (String) c)); + map.put(NestedAggregationBuilder.NAME, (p, c) -> ParsedNested.fromXContent(p, (String) c)); + map.put(ReverseNestedAggregationBuilder.NAME, (p, c) -> ParsedReverseNested.fromXContent(p, (String) c)); + map.put(GlobalAggregationBuilder.NAME, (p, c) -> ParsedGlobal.fromXContent(p, (String) c)); + map.put(FilterAggregationBuilder.NAME, (p, c) -> ParsedFilter.fromXContent(p, (String) c)); + map.put(InternalSampler.PARSER_NAME, (p, c) -> ParsedSampler.fromXContent(p, (String) c)); + map.put(GeoHashGridAggregationBuilder.NAME, (p, c) -> ParsedGeoHashGrid.fromXContent(p, (String) c)); + map.put(GeoTileGridAggregationBuilder.NAME, (p, c) -> ParsedGeoTileGrid.fromXContent(p, (String) c)); + map.put(RangeAggregationBuilder.NAME, (p, c) -> ParsedRange.fromXContent(p, (String) c)); + map.put(DateRangeAggregationBuilder.NAME, (p, c) -> ParsedDateRange.fromXContent(p, (String) c)); + map.put(GeoDistanceAggregationBuilder.NAME, (p, c) -> ParsedGeoDistance.fromXContent(p, (String) c)); + map.put(FiltersAggregationBuilder.NAME, (p, c) -> ParsedFilters.fromXContent(p, (String) c)); + map.put(AdjacencyMatrixAggregationBuilder.NAME, (p, c) -> ParsedAdjacencyMatrix.fromXContent(p, (String) c)); + map.put(SignificantLongTerms.NAME, (p, c) -> ParsedSignificantLongTerms.fromXContent(p, (String) c)); + map.put(SignificantStringTerms.NAME, (p, c) -> ParsedSignificantStringTerms.fromXContent(p, (String) c)); + map.put(ScriptedMetricAggregationBuilder.NAME, (p, c) -> ParsedScriptedMetric.fromXContent(p, (String) c)); + map.put(IpRangeAggregationBuilder.NAME, (p, c) -> ParsedBinaryRange.fromXContent(p, (String) c)); + map.put(TopHitsAggregationBuilder.NAME, (p, c) -> ParsedTopHits.fromXContent(p, (String) c)); + map.put(CompositeAggregationBuilder.NAME, (p, c) -> ParsedComposite.fromXContent(p, (String) c)); + List entries = map.entrySet().stream() + .map(entry -> new NamedXContentRegistry.Entry(Aggregation.class, new ParseField(entry.getKey()), entry.getValue())) + .collect(Collectors.toList()); + entries.add(new NamedXContentRegistry.Entry(Suggest.Suggestion.class, new ParseField(TermSuggestionBuilder.SUGGESTION_NAME), + (parser, context) -> TermSuggestion.fromXContent(parser, (String)context))); + entries.add(new NamedXContentRegistry.Entry(Suggest.Suggestion.class, new ParseField(PhraseSuggestionBuilder.SUGGESTION_NAME), + (parser, context) -> PhraseSuggestion.fromXContent(parser, (String)context))); + entries.add(new NamedXContentRegistry.Entry(Suggest.Suggestion.class, new ParseField(CompletionSuggestionBuilder.SUGGESTION_NAME), + (parser, context) -> CompletionSuggestion.fromXContent(parser, (String)context))); + return entries; + } +} diff --git a/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java index 7b49f8dba..a19090abc 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java @@ -17,6 +17,7 @@ package org.springframework.data.elasticsearch.core; import static org.elasticsearch.index.VersionType.*; +import org.elasticsearch.search.aggregations.Aggregation; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -102,6 +103,7 @@ import org.springframework.util.Assert; * @author Mathias Teier * @author Aleksei Arsenev * @author Roman Puchkovskiy + * @author Russell Parry * @since 3.2 */ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOperations, ApplicationContextAware { @@ -608,6 +610,27 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera }); } + @Override + public Flux aggregate(Query query, Class entityType) { + return aggregate(query, entityType, getIndexCoordinatesFor(entityType)); + } + + @Override + public Flux aggregate(Query query, Class entityType, IndexCoordinates index) { + return doAggregate(query, entityType, index); + } + + private Flux doAggregate(Query query, Class entityType, IndexCoordinates index) { + return Flux.defer(() -> { + SearchRequest request = requestFactory.searchRequest(query, entityType, index); + request = prepareSearchRequest(request); + request.source().size(0); + request.source().trackTotalHits(false); + + return doAggregate(request); + }); + } + @Override public Mono count(Query query, Class entityType) { return count(query, entityType, getIndexCoordinatesFor(entityType)); @@ -682,6 +705,22 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera .onErrorResume(NoSuchIndexException.class, it -> Mono.empty()); } + /** + * Customization hook on the actual execution result {@link Publisher}.
+ * + * @param request the already prepared {@link SearchRequest} ready to be executed. + * @return a {@link Flux} emitting the result of the operation. + */ + protected Flux doAggregate(SearchRequest request) { + + if (QUERY_LOGGER.isDebugEnabled()) { + QUERY_LOGGER.debug("Executing doCount: {}", request); + } + + return Flux.from(execute(client -> client.aggregate(request))) // + .onErrorResume(NoSuchIndexException.class, it -> Flux.empty()); + } + /** * Customization hook on the actual execution result {@link Publisher}.
* diff --git a/src/main/java/org/springframework/data/elasticsearch/core/ReactiveSearchOperations.java b/src/main/java/org/springframework/data/elasticsearch/core/ReactiveSearchOperations.java index 1006b833b..32e3e4edc 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/ReactiveSearchOperations.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/ReactiveSearchOperations.java @@ -15,6 +15,7 @@ */ package org.springframework.data.elasticsearch.core; +import org.elasticsearch.search.aggregations.Aggregation; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -30,6 +31,7 @@ import org.springframework.data.elasticsearch.core.query.StringQuery; * APIs. * * @author Peter-Josef Meisch + * @author Russell Parry * @since 4.0 */ public interface ReactiveSearchOperations { @@ -183,4 +185,25 @@ public interface ReactiveSearchOperations { default Flux> search(Query query, Class entityType, IndexCoordinates index) { return search(query, entityType, entityType, index); } + + /** + * Perform an aggregation specified by the given {@link Query query}.
+ * + * @param query must not be {@literal null}. + * @param entityType must not be {@literal null}. + * @return a {@link Flux} emitting matching aggregations one by one. + * @since 4.0 + */ + Flux aggregate(Query query, Class entityType); + + /** + * Perform an aggregation specified by the given {@link Query query}.
+ * + * @param query must not be {@literal null}. + * @param entityType must not be {@literal null}. + * @param index the target index, must not be {@literal null} + * @return a {@link Flux} emitting matching aggregations one by one. + * @since 4.0 + */ + Flux aggregate(Query query, Class entityType, IndexCoordinates index); } diff --git a/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java b/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java index 0d7d1858c..fb13bb6bc 100644 --- a/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java +++ b/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java @@ -19,6 +19,10 @@ import static org.assertj.core.api.Assertions.*; import lombok.SneakyThrows; import org.elasticsearch.client.indices.GetIndexRequest; +import org.elasticsearch.client.indices.PutMappingRequest; +import org.elasticsearch.search.aggregations.AggregationBuilders; +import org.elasticsearch.search.aggregations.bucket.terms.StringTerms; +import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder; import reactor.test.StepVerifier; import java.io.IOException; @@ -67,6 +71,7 @@ import org.springframework.test.context.ContextConfiguration; * @author Mark Paluch * @author Peter-Josef Meisch * @author Henrique Amaral + * @author Russell Parry */ @SpringIntegrationTest @ContextConfiguration(classes = { ElasticsearchRestTemplateConfiguration.class }) @@ -656,6 +661,27 @@ public class ReactiveElasticsearchClientTests { }).verifyComplete(); } + @Test //DATAES-567 + public void aggregateReturnsAggregationResults() throws IOException { + syncClient.indices().create(new CreateIndexRequest(INDEX_I), RequestOptions.DEFAULT); + Map jsonMap = Collections.singletonMap("properties", + Collections.singletonMap("firstname", Collections.singletonMap("type", "keyword"))); + syncClient.indices().putMapping(new PutMappingRequest(INDEX_I).source(jsonMap), RequestOptions.DEFAULT); + + addSourceDocument().ofType(TYPE_I).to(INDEX_I); + + SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()); + searchSourceBuilder.aggregation(AggregationBuilders.terms("terms").field("firstname")); + + SearchRequest request = new SearchRequest(INDEX_I) // + .source(searchSourceBuilder); + + client.aggregate(request) + .as(StepVerifier::create) + .expectNextMatches(aggregation -> aggregation.getType().equals(StringTerms.NAME)) + .verifyComplete(); + } + private AddToIndexOfType addSourceDocument() { return add(DOC_SOURCE); } diff --git a/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientUnitTests.java b/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientUnitTests.java index 513c23846..feb86a9a6 100644 --- a/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientUnitTests.java +++ b/src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientUnitTests.java @@ -19,11 +19,15 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; import static org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive.*; +import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms; +import org.elasticsearch.search.aggregations.metrics.ParsedMax; +import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.io.IOException; import java.net.URI; +import java.time.Instant; import java.util.Collections; import org.elasticsearch.ElasticsearchStatusException; @@ -53,6 +57,7 @@ import org.springframework.util.StreamUtils; /** * @author Christoph Strobl * @author Henrique Amaral + * @author Russell Parry */ public class ReactiveElasticsearchClientUnitTests { @@ -577,6 +582,83 @@ public class ReactiveElasticsearchClientUnitTests { .verifyComplete(); } + // --> AGGREGATE + + @Test // DATAES-567 + public void aggregateShouldHitSearchEndpoint() { + + hostProvider.when(HOST) // + .receive(Receive::json) // + .body(fromPath("aggregate-ok-no-results")); + + client.search(new SearchRequest("twitter")).as(StepVerifier::create).verifyComplete(); + + verify(hostProvider.client(HOST)).method(HttpMethod.POST); + URI uri = hostProvider.when(HOST).captureUri(); + assertThat(uri.getRawPath()).isEqualTo("/twitter/_search"); + } + + @Test // DATAES-567 + public void aggregateShouldReturnSingleResultCorrectly() { + hostProvider.when(HOST) // + .receive(Receive::json) // + .body(fromPath("aggregate-ok-single-result")); + + client.aggregate(new SearchRequest("twitter")) // + .as(StepVerifier::create) // + .consumeNextWith(aggregation -> { + assertThat(aggregation.getName()).isEqualTo("users"); + assertThat(aggregation instanceof ParsedStringTerms); + ParsedStringTerms parsedStringTerms = (ParsedStringTerms) aggregation; + assertThat(parsedStringTerms.getBuckets().size()).isEqualTo(2); + assertThat(parsedStringTerms.getBucketByKey("kimchy").getDocCount()).isEqualTo(2); + assertThat(parsedStringTerms.getBucketByKey("elastic").getDocCount()).isEqualTo(1); + }).verifyComplete(); + } + + @Test // DATAES-567 + public void aggregateShouldReturnMultipleResultsCorrectly() { + + hostProvider.when(HOST) // + .receive(Receive::json) // + .body(fromPath("aggregate-ok-multiple-results")); + + client.aggregate(new SearchRequest("twitter")) // + .as(StepVerifier::create) // + .consumeNextWith(aggregation -> { + assertThat(aggregation.getName()).isEqualTo("users"); + assertThat(aggregation instanceof ParsedStringTerms); + ParsedStringTerms parsedStringTerms = (ParsedStringTerms) aggregation; + assertThat(parsedStringTerms.getBuckets().size()).isEqualTo(2); + assertThat(parsedStringTerms.getBucketByKey("kimchy").getDocCount()).isEqualTo(2); + assertThat(parsedStringTerms.getBucketByKey("elastic").getDocCount()).isEqualTo(1); + }) // + .consumeNextWith(aggregation -> { + assertThat(aggregation.getName()).isEqualTo("max_post_date"); + assertThat(aggregation instanceof ParsedMax); + ParsedMax parsedMax = (ParsedMax) aggregation; + assertThat(Instant.ofEpochMilli((long)parsedMax.getValue())).isEqualTo(Instant.parse("2010-01-15T01:46:38Z")); + }).verifyComplete(); + } + + @Test // DATAES-567 + public void aggregateShouldReturnAggregationWithNoValuesWhenNoResultsFound() { + + hostProvider.when(HOST) // + .receive(Receive::json) // + .body(fromPath("aggregate-ok-no-results")); + + + client.aggregate(new SearchRequest("twitter")) // + .as(StepVerifier::create) // + .consumeNextWith(aggregation -> { + assertThat(aggregation.getName()).isEqualTo("users"); + assertThat(aggregation instanceof ParsedStringTerms); + ParsedStringTerms parsedStringTerms = (ParsedStringTerms) aggregation; + assertThat(parsedStringTerms.getBuckets().size()).isEqualTo(0); + }).verifyComplete(); + } + // --> SCROLL @Test // DATAES-510 diff --git a/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateTests.java b/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateTests.java index 34649fe32..f39cbca3f 100644 --- a/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateTests.java +++ b/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateTests.java @@ -24,6 +24,8 @@ import lombok.Builder; import lombok.Data; import lombok.EqualsAndHashCode; import lombok.NoArgsConstructor; +import org.elasticsearch.search.aggregations.AggregationBuilders; +import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -78,6 +80,7 @@ import org.springframework.util.StringUtils; * @author Farid Azaza * @author Martin Choraine * @author Aleksei Arsenev + * @author Russell Parry */ @SpringIntegrationTest public class ReactiveElasticsearchTemplateTests { @@ -486,6 +489,41 @@ public class ReactiveElasticsearchTemplateTests { .verifyComplete(); } + @Test // DATAES-567 + public void aggregateShouldReturnAggregations() { + + SampleEntity sampleEntity1 = randomEntity("some message"); + SampleEntity sampleEntity2 = randomEntity("some message"); + SampleEntity sampleEntity3 = randomEntity("other message"); + + index(sampleEntity1, sampleEntity2, sampleEntity3); + + NativeSearchQuery query = new NativeSearchQueryBuilder() + .withQuery(matchAllQuery()) + .addAggregation(AggregationBuilders.terms("messages").field("message")) + .build(); + + template.aggregate(query, SampleEntity.class) // + .as(StepVerifier::create) // + .consumeNextWith(aggregation -> { + assertThat(aggregation.getName()).isEqualTo("messages"); + assertThat(aggregation instanceof ParsedStringTerms); + ParsedStringTerms parsedStringTerms = (ParsedStringTerms) aggregation; + assertThat(parsedStringTerms.getBuckets().size()).isEqualTo(3); + assertThat(parsedStringTerms.getBucketByKey("message").getDocCount()).isEqualTo(3); + assertThat(parsedStringTerms.getBucketByKey("some").getDocCount()).isEqualTo(2); + assertThat(parsedStringTerms.getBucketByKey("other").getDocCount()).isEqualTo(1); + }).verifyComplete(); + } + + @Test // DATAES-567 + public void aggregateShouldReturnEmptyWhenIndexDoesNotExist() { + template.aggregate(new CriteriaQuery(Criteria.where("message").is("some message")), SampleEntity.class, + IndexCoordinates.of("no-such-index")) // + .as(StepVerifier::create) // + .verifyComplete(); + } + @Test // DATAES-519 public void countShouldReturnZeroWhenIndexDoesNotExist() { diff --git a/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-multiple-results.json b/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-multiple-results.json new file mode 100644 index 000000000..262414083 --- /dev/null +++ b/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-multiple-results.json @@ -0,0 +1,35 @@ +{ + "took": 52, + "timed_out": false, + "_shards": { + "total": 5, + "successful": 5, + "skipped": 0, + "failed": 0 + }, + "hits": { + "total": 0, + "max_score": null, + "hits": [ ] + }, + "aggregations": { + "sterms#users": { + "doc_count_error_upper_bound": 0, + "sum_other_doc_count": 0, + "buckets": [ + { + "key": "kimchy", + "doc_count": 2 + }, + { + "key": "elastic", + "doc_count": 1 + } + ] + }, + "max#max_post_date": { + "value" : 1.263519998E12, + "value_as_string" : "2010-01-15T01:46:38.000Z" + } + } +} \ No newline at end of file diff --git a/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-no-results.json b/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-no-results.json new file mode 100644 index 000000000..157326a81 --- /dev/null +++ b/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-no-results.json @@ -0,0 +1,22 @@ +{ + "took" : 226, + "timed_out" : false, + "_shards" : { + "total" : 5, + "successful" : 5, + "skipped" : 0, + "failed" : 0 + }, + "hits" : { + "total" : 0, + "max_score" : null, + "hits" : [ ] + }, + "aggregations" : { + "sterms#users" : { + "doc_count_error_upper_bound" : 0, + "sum_other_doc_count" : 0, + "buckets" : [ ] + } + } +} diff --git a/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-single-result.json b/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-single-result.json new file mode 100644 index 000000000..a08726da4 --- /dev/null +++ b/src/test/resources/org/springframework/data/elasticsearch/client/aggregate-ok-single-result.json @@ -0,0 +1,31 @@ +{ + "took": 52, + "timed_out": false, + "_shards": { + "total": 5, + "successful": 5, + "skipped": 0, + "failed": 0 + }, + "hits": { + "total": 0, + "max_score": null, + "hits": [ ] + }, + "aggregations": { + "sterms#users": { + "doc_count_error_upper_bound": 0, + "sum_other_doc_count": 0, + "buckets": [ + { + "key": "kimchy", + "doc_count": 2 + }, + { + "key": "elastic", + "doc_count": 1 + } + ] + } + } +} \ No newline at end of file