-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support for non global aggregations with concurrent segment search. T…
…his PR does not include the support for profile option with aggregations to work with concurrent model Signed-off-by: Sorabh Hamirwasia <sohami.apache@gmail.com>
- Loading branch information
Showing
19 changed files
with
694 additions
and
197 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 changes: 45 additions & 0 deletions
45
server/src/main/java/org/opensearch/search/aggregations/AggregationCollectorManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.search.aggregations; | ||
|
|
||
| import org.apache.lucene.search.Collector; | ||
| import org.apache.lucene.search.CollectorManager; | ||
| import org.opensearch.search.internal.SearchContext; | ||
| import org.opensearch.search.query.ReduceableSearchResult; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
|
|
||
| import static org.opensearch.search.aggregations.DefaultAggregationProcessor.createCollector; | ||
|
|
||
| /** | ||
| * {@link CollectorManager} to take care of aggregation operators both in case of concurrent and non-concurrent | ||
| * segment search | ||
| */ | ||
| public class AggregationCollectorManager implements CollectorManager<Collector, ReduceableSearchResult> { | ||
| private final SearchContext context; | ||
|
|
||
| public AggregationCollectorManager(SearchContext context) { | ||
| this.context = context; | ||
| } | ||
|
|
||
| @Override | ||
| public Collector newCollector() throws IOException { | ||
| List<Aggregator> nonGlobalAggregators = context.aggregations().factories().createTopLevelNonGlobalAggregators(context); | ||
| assert !nonGlobalAggregators.isEmpty() : "Expected atleast one non global aggregator to be present"; | ||
| context.aggregations().addNonGlobalAggregators(nonGlobalAggregators); | ||
| return createCollector(context, nonGlobalAggregators); | ||
| } | ||
|
|
||
| @Override | ||
| public ReduceableSearchResult reduce(Collection<Collector> collectors) throws IOException { | ||
| return (result) -> {}; | ||
| } | ||
| } |
36 changes: 36 additions & 0 deletions
36
server/src/main/java/org/opensearch/search/aggregations/AggregationProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.search.aggregations; | ||
|
|
||
| import org.opensearch.search.internal.SearchContext; | ||
|
|
||
| /** | ||
| * Interface to define different stages of aggregation processing before and after document collection | ||
| */ | ||
| public interface AggregationProcessor { | ||
|
|
||
| /** | ||
| * Callback invoked before collection of documents are done | ||
| * @param context {@link SearchContext} for the request | ||
| */ | ||
| void preProcess(SearchContext context); | ||
|
|
||
| /** | ||
| * Callback invoked after collection of documents are done | ||
| * @param context {@link SearchContext} for the request | ||
| */ | ||
| void postProcess(SearchContext context); | ||
|
|
||
| /** | ||
| * Callback to process the {@link org.opensearch.search.aggregations.bucket.global.GlobalAggregator} present | ||
| * in the search request | ||
| * @param context {@link SearchContext} for the request | ||
| */ | ||
| void processGlobalAggregators(SearchContext context); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
40 changes: 40 additions & 0 deletions
40
server/src/main/java/org/opensearch/search/aggregations/ConcurrentAggregationProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.search.aggregations; | ||
|
|
||
| import org.opensearch.search.internal.SearchContext; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * {@link AggregationProcessor} implementation to be used with {@link org.opensearch.search.query.ConcurrentQueryPhaseSearcher}. It takes | ||
| * care of performing shard level reduce on Aggregation results collected as part of concurrent execution among slices. This is done to | ||
| * avoid the increase in aggregation result sets returned by each shard to coordinator where final reduce happens for results received from | ||
| * all the shards | ||
| */ | ||
| public class ConcurrentAggregationProcessor extends DefaultAggregationProcessor { | ||
|
|
||
| @Override | ||
| public void populateResult(SearchContext context, List<InternalAggregation> aggregations) { | ||
| InternalAggregations internalAggregations = InternalAggregations.from(aggregations); | ||
| // Reduce the aggregations across slices before sending to the coordinator. We will perform shard level reduce iff multiple slices | ||
| // were created to execute this request and it used concurrent segment search path | ||
| // TODO: Add the check for flag that the request was executed using concurrent search | ||
| if (context.searcher().getSlices().length > 1) { | ||
| // using reduce is fine here instead of topLevelReduce as pipeline aggregation is evaluated on the coordinator after all | ||
| // documents are collected across shards for an aggregation | ||
| internalAggregations = InternalAggregations.reduce( | ||
| Collections.singletonList(internalAggregations), | ||
| context.aggregationReduceContext() | ||
| ); | ||
| } | ||
| context.queryResult().aggregations(internalAggregations); | ||
| } | ||
| } |
Oops, something went wrong.