honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup SearchPhaseController interface (#23844) 

SearchPhaseController is tighly coupled to AtomicArray which makes
non-dense representations of results very difficult. This commit removes
the coupling and cuts over to Collection rather than List to ensure no
order or random access lookup is implied.
This commit is contained in:
Simon Willnauer 2017-03-31 16:25:15 +02:00 committed by GitHub
parent a8250b26e7
commit 135eae42b9
5 changed files with 24 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java
View File

@ -202,7 +202,7 @@ final class FetchSearchPhase extends SearchPhase {
String scrollId, SearchPhaseController.ReducedQueryPhase reducedQueryPhase,
AtomicArray<? extends SearchPhaseResult> fetchResultsArr) {
final InternalSearchResponse internalResponse = searchPhaseController.merge(context.getRequest().scroll() != null,
sortedDocs, reducedQueryPhase, fetchResultsArr);
sortedDocs, reducedQueryPhase, fetchResultsArr.asList(), fetchResultsArr::get);
context.executeNextPhase(this, nextPhaseFactory.apply(context.buildSearchResponse(internalResponse, scrollId)));
}

36
core/src/main/java/org/elasticsearch/action/search/SearchPhaseController.java
View File

@ -36,7 +36,6 @@ import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.lucene.Lucene;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.BigArrays;
import org.elasticsearch.common.util.concurrent.AtomicArray;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
@ -61,14 +60,16 @@ import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.IntFunction;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
public class SearchPhaseController extends AbstractComponent {
public final class SearchPhaseController extends AbstractComponent {
private static final ScoreDoc[] EMPTY_DOCS = new ScoreDoc[0];
@ -81,7 +82,7 @@ public class SearchPhaseController extends AbstractComponent {
this.scriptService = scriptService;
}
public AggregatedDfs aggregateDfs(List<DfsSearchResult> results) {
public AggregatedDfs aggregateDfs(Collection<DfsSearchResult> results) {
ObjectObjectHashMap<Term, TermStatistics> termStatistics = HppcMaps.newNoNullKeysMap();
ObjectObjectHashMap<String, CollectionStatistics> fieldStatistics = HppcMaps.newNoNullKeysMap();
long aggMaxDoc = 0;
@ -148,7 +149,7 @@ public class SearchPhaseController extends AbstractComponent {
* Enabled only for scroll search, because that only retrieves hits of length 'size' in the query phase.
* @param results Shard result holder
*/
public ScoreDoc[] sortDocs(boolean ignoreFrom, List<? extends SearchPhaseResult> results, int numShards) throws IOException {
public ScoreDoc[] sortDocs(boolean ignoreFrom, Collection<? extends SearchPhaseResult> results, int numShards) throws IOException {
if (results.isEmpty()) {
return EMPTY_DOCS;
}
@ -158,7 +159,7 @@ public class SearchPhaseController extends AbstractComponent {
int shardIndex = -1;
if (results.size() == 1) {
canOptimize = true;
result = results.get(0).queryResult();
result = results.stream().findFirst().get().queryResult();
shardIndex = result.getShardIndex();
} else {
boolean hasResult = false;
@ -176,7 +177,7 @@ public class SearchPhaseController extends AbstractComponent {
shardIndex = resultToOptimize.getShardIndex();
}
}
result = canOptimize ? resultToOptimize : results.get(0).queryResult();
result = canOptimize ? resultToOptimize : results.stream().findFirst().get().queryResult();
assert result != null;
}
if (canOptimize) {
@ -284,8 +285,7 @@ public class SearchPhaseController extends AbstractComponent {
}
static <T extends TopDocs> void fillTopDocs(T[] shardTopDocs,
List<? extends SearchPhaseResult> results,
T empytTopDocs) {
Collection<? extends SearchPhaseResult> results, T empytTopDocs) {
if (results.size() != shardTopDocs.length) {
// TopDocs#merge can't deal with null shard TopDocs
Arrays.fill(shardTopDocs, empytTopDocs);
@ -338,12 +338,11 @@ public class SearchPhaseController extends AbstractComponent {
*/
public InternalSearchResponse merge(boolean ignoreFrom, ScoreDoc[] sortedDocs,
ReducedQueryPhase reducedQueryPhase,
AtomicArray<? extends SearchPhaseResult> fetchResultsArr) {
Collection<? extends SearchPhaseResult> fetchResults, IntFunction<SearchPhaseResult> resultsLookup) {
if (reducedQueryPhase.isEmpty()) {
return InternalSearchResponse.empty();
}
List<? extends SearchPhaseResult> fetchResults = fetchResultsArr.asList();
SearchHits hits = getHits(reducedQueryPhase, ignoreFrom, sortedDocs, fetchResultsArr);
SearchHits hits = getHits(reducedQueryPhase, ignoreFrom, sortedDocs, fetchResults, resultsLookup);
if (reducedQueryPhase.suggest != null) {
if (!fetchResults.isEmpty()) {
int currentOffset = hits.getHits().length;
@ -351,7 +350,7 @@ public class SearchPhaseController extends AbstractComponent {
final List<CompletionSuggestion.Entry.Option> suggestionOptions = suggestion.getOptions();
for (int scoreDocIndex = currentOffset; scoreDocIndex < currentOffset + suggestionOptions.size(); scoreDocIndex++) {
ScoreDoc shardDoc = sortedDocs[scoreDocIndex];
SearchPhaseResult searchResultProvider = fetchResultsArr.get(shardDoc.shardIndex);
SearchPhaseResult searchResultProvider = resultsLookup.apply(shardDoc.shardIndex);
if (searchResultProvider == null) {
continue;
}
@ -375,8 +374,7 @@ public class SearchPhaseController extends AbstractComponent {
}
private SearchHits getHits(ReducedQueryPhase reducedQueryPhase, boolean ignoreFrom, ScoreDoc[] sortedDocs,
AtomicArray<? extends SearchPhaseResult> fetchResultsArr) {
List<? extends SearchPhaseResult> fetchResults = fetchResultsArr.asList();
Collection<? extends SearchPhaseResult> fetchResults, IntFunction<SearchPhaseResult> resultsLookup) {
boolean sorted = false;
int sortScoreIndex = -1;
if (reducedQueryPhase.oneResult.topDocs() instanceof TopFieldDocs) {
@ -406,7 +404,7 @@ public class SearchPhaseController extends AbstractComponent {
if (!fetchResults.isEmpty()) {
for (int i = 0; i < numSearchHits; i++) {
ScoreDoc shardDoc = sortedDocs[i];
SearchPhaseResult fetchResultProvider = fetchResultsArr.get(shardDoc.shardIndex);
SearchPhaseResult fetchResultProvider = resultsLookup.apply(shardDoc.shardIndex);
if (fetchResultProvider == null) {
continue;
}
@ -435,7 +433,7 @@ public class SearchPhaseController extends AbstractComponent {
* Reduces the given query results and consumes all aggregations and profile results.
* @param queryResults a list of non-null query shard results
*/
public final ReducedQueryPhase reducedQueryPhase(List<? extends SearchPhaseResult> queryResults) {
public ReducedQueryPhase reducedQueryPhase(List<? extends SearchPhaseResult> queryResults) {
return reducedQueryPhase(queryResults, null, 0);
}
@ -448,7 +446,7 @@ public class SearchPhaseController extends AbstractComponent {
* @see QuerySearchResult#consumeAggs()
* @see QuerySearchResult#consumeProfileResult()
*/
private ReducedQueryPhase reducedQueryPhase(List<? extends SearchPhaseResult> queryResults,
private ReducedQueryPhase reducedQueryPhase(Collection<? extends SearchPhaseResult> queryResults,
List<InternalAggregations> bufferdAggs, int numReducePhases) {
assert numReducePhases >= 0 : "num reduce phases must be >= 0 but was: " + numReducePhases;
numReducePhases++; // increment for this phase
@ -461,7 +459,7 @@ public class SearchPhaseController extends AbstractComponent {
return new ReducedQueryPhase(totalHits, fetchHits, maxScore, timedOut, terminatedEarly, null, null, null, null,
numReducePhases);
}
final QuerySearchResult firstResult = queryResults.get(0).queryResult();
final QuerySearchResult firstResult = queryResults.stream().findFirst().get().queryResult();
final boolean hasSuggest = firstResult.suggest() != null;
final boolean hasProfileResults = firstResult.hasProfileResults();
final boolean consumeAggs;
@ -599,7 +597,7 @@ public class SearchPhaseController extends AbstractComponent {
/**
* Creates a new search response from the given merged hits.
* @see #merge(boolean, ScoreDoc[], ReducedQueryPhase, AtomicArray)
* @see #merge(boolean, ScoreDoc[], ReducedQueryPhase, Collection, IntFunction)
*/
public InternalSearchResponse buildResponse(SearchHits hits) {
return new InternalSearchResponse(hits, aggregations, suggest, shardResults, timedOut, terminatedEarly, numReducePhases);

3
core/src/main/java/org/elasticsearch/action/search/SearchScrollQueryAndFetchAsyncAction.java
View File

@ -172,9 +172,10 @@ final class SearchScrollQueryAndFetchAsyncAction extends AbstractAsyncAction {
}
private void innerFinishHim() throws Exception {
List<QueryFetchSearchResult> queryFetchSearchResults = queryFetchResults.asList();
ScoreDoc[] sortedShardDocs = searchPhaseController.sortDocs(true, queryFetchResults.asList(), queryFetchResults.length());
final InternalSearchResponse internalResponse = searchPhaseController.merge(true, sortedShardDocs,
searchPhaseController.reducedQueryPhase(queryFetchResults.asList()), queryFetchResults);
searchPhaseController.reducedQueryPhase(queryFetchSearchResults), queryFetchSearchResults, queryFetchResults::get);
String scrollId = null;
if (request.scroll() != null) {
scrollId = request.scrollId();

3
core/src/main/java/org/elasticsearch/action/search/SearchScrollQueryThenFetchAsyncAction.java
View File

@ -222,7 +222,8 @@ final class SearchScrollQueryThenFetchAsyncAction extends AbstractAsyncAction {
private void finishHim(SearchPhaseController.ReducedQueryPhase queryPhase) {
try {
final InternalSearchResponse internalResponse = searchPhaseController.merge(true, sortedShardDocs, queryPhase, fetchResults);
final InternalSearchResponse internalResponse = searchPhaseController.merge(true, sortedShardDocs, queryPhase,
fetchResults.asList(), fetchResults::get);
String scrollId = null;
if (request.scroll() != null) {
scrollId = request.scrollId();

3
core/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java
View File

@ -119,9 +119,10 @@ public class SearchPhaseControllerTests extends ESTestCase {
}
}
ScoreDoc[] sortedDocs = mergedScoreDocs.toArray(new ScoreDoc[mergedScoreDocs.size()]);
AtomicArray<SearchPhaseResult> searchPhaseResultAtomicArray = generateFetchResults(nShards, mergedSearchDocs, mergedSuggest);
InternalSearchResponse mergedResponse = searchPhaseController.merge(true, sortedDocs,
searchPhaseController.reducedQueryPhase(queryResults.asList()),
generateFetchResults(nShards, mergedSearchDocs, mergedSuggest));
searchPhaseResultAtomicArray.asList(), searchPhaseResultAtomicArray::get);
assertThat(mergedResponse.hits().getHits().length, equalTo(mergedSearchDocs.length));
Suggest suggestResult = mergedResponse.suggest();
for (Suggest.Suggestion<?> suggestion : mergedSuggest) {