import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
public void start() {
    if (expectedSuccessfulOps == 0) {
        // no search shards to search on, bail with empty response (it happens with search across _all with no indices around and consistent with broadcast operations)
        listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, buildTookInMillis(), ShardSearchFailure.EMPTY_ARRAY));
        return;
    }
    int shardIndex = -1;
    for (final ShardIterator shardIt : shardsIts) {
        shardIndex++;
        final ShardRouting shard = shardIt.nextOrNull();
        if (shard != null) {
            performFirstPhase(shardIndex, shardIt, shard);
        } else {
            // really, no shards active in this group
            onFirstPhaseResult(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
        }
    }
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void finishHim() {
    threadPool.executor(ThreadPool.Names.SEARCH).execute(new ActionRunnable<SearchResponse>(listener) {
        @Override
        public void doRun() throws IOException {
            sortedShardList = searchPhaseController.sortDocs(true, queryFetchResults);
            final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryFetchResults,
                queryFetchResults, request);
            String scrollId = null;
            if (request.scroll() != null) {
                scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null);
            }
            listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(),
                buildTookInMillis(), buildShardFailures()));
        }

        @Override
        public void onFailure(Throwable t) {
            ReduceSearchPhaseException failure = new ReduceSearchPhaseException("query_fetch", "", t, buildShardFailures());
            if (logger.isDebugEnabled()) {
                logger.debug("failed to reduce search", failure);
            }
            super.onFailure(t);
        }
    });

}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
protected SearchResponse createResponse(HttpContext<SearchRequest, SearchResponse> httpContext) throws IOException {
    if (httpContext == null) {
        throw new IllegalStateException("no http context");
    }
    HttpResponse httpResponse = httpContext.getHttpResponse();
    logger.info("{}", httpResponse.getContent().toString(CharsetUtil.UTF_8));
    BytesReference ref = new ChannelBufferBytesReference(httpResponse.getContent());
    Map<String, Object> map = JsonXContent.jsonXContent.createParser(ref).map();

    logger.info("{}", map);

    InternalSearchResponse internalSearchResponse = parseInternalSearchResponse(map);
    String scrollId = (String) map.get(SCROLL_ID);
    int totalShards = 0;
    int successfulShards = 0;
    if (map.containsKey(SHARDS)) {
        Map<String, ?> shards = (Map<String, ?>) map.get(SHARDS);
        totalShards = shards.containsKey(TOTAL) ? (Integer) shards.get(TOTAL) : -1;
        successfulShards = shards.containsKey(SUCCESSFUL) ? (Integer) shards.get(SUCCESSFUL) : -1;
    }
    int tookInMillis = map.containsKey(TOOK) ? (Integer) map.get(TOOK) : -1;
    ShardSearchFailure[] shardFailures = null;
    return new SearchResponse(internalSearchResponse, scrollId, totalShards, successfulShards, tookInMillis, shardFailures);
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
protected SearchResponse createResponse(HttpInvocationContext<SearchRequest,SearchResponse> httpInvocationContext) throws IOException {
    if (httpInvocationContext == null) {
        throw new IllegalStateException("no http context");
    }
    HttpResponse httpResponse = httpInvocationContext.getHttpResponse();
    logger.info("{}", httpResponse.getContent().toString(CharsetUtil.UTF_8));
    BytesReference ref = new ChannelBufferBytesReference(httpResponse.getContent());
    Map<String,Object> map = JsonXContent.jsonXContent.createParser(ref).map();

    logger.info("{}", map);

    InternalSearchResponse internalSearchResponse = parseInternalSearchResponse(map);
    String scrollId = (String)map.get(SCROLL_ID);
    int totalShards = 0;
    int successfulShards = 0;
    if (map.containsKey(SHARDS)) {
        Map<String,?> shards = (Map<String,?>)map.get(SHARDS);
        totalShards =  shards.containsKey(TOTAL) ? (Integer)shards.get(TOTAL) : -1;
        successfulShards =  shards.containsKey(SUCCESSFUL) ? (Integer)shards.get(SUCCESSFUL) : -1;
    }
    int tookInMillis = map.containsKey(TOOK) ? (Integer)map.get(TOOK) : -1;
    ShardSearchFailure[] shardFailures = parseShardFailures(map);
    return new SearchResponse(internalSearchResponse, scrollId, totalShards, successfulShards, tookInMillis, shardFailures);
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
@Override
protected void doExecute(SearchRequest request, ActionListener<SearchResponse> listener) {
    listener.onResponse(new SearchResponse(new InternalSearchResponse(
        new SearchHits(
            new SearchHit[0], 0L, 0.0f),
        new InternalAggregations(Collections.emptyList()),
        new Suggest(Collections.emptyList()),
        new SearchProfileShardResults(Collections.emptyMap()), false, false, 1), "", 1, 1, 0, new ShardSearchFailure[0]));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
/**
 * This is the main entry point for a search. This method starts the search execution of the initial phase.
 */
public final void start() {
    if (getNumShards() == 0) {
        //no search shards to search on, bail with empty response
        //(it happens with search across _all with no indices around and consistent with broadcast operations)
        listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, buildTookInMillis(),
            ShardSearchFailure.EMPTY_ARRAY));
        return;
    }
    executePhase(this);
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void moveToNextPhase(SearchPhaseController searchPhaseController, ScoreDoc[] sortedDocs,
                             String scrollId, SearchPhaseController.ReducedQueryPhase reducedQueryPhase,
                             AtomicArray<? extends QuerySearchResultProvider> fetchResultsArr) {
    final InternalSearchResponse internalResponse = searchPhaseController.merge(context.getRequest().scroll() != null,
        sortedDocs, reducedQueryPhase, fetchResultsArr);
    context.executeNextPhase(this, nextPhaseFactory.apply(context.buildSearchResponse(internalResponse, scrollId)));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
public SearchResponse(InternalSearchResponse internalResponse, String scrollId, int totalShards, int successfulShards,
                      long tookInMillis, ShardSearchFailure[] shardFailures) {
    this.internalResponse = internalResponse;
    this.scrollId = scrollId;
    this.totalShards = totalShards;
    this.successfulShards = successfulShards;
    this.tookInMillis = tookInMillis;
    this.shardFailures = shardFailures;
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void innerFinishHim() throws Exception {
    ScoreDoc[] sortedShardDocs = searchPhaseController.sortDocs(true, queryFetchResults);
    final InternalSearchResponse internalResponse = searchPhaseController.merge(true, sortedShardDocs,
        searchPhaseController.reducedQueryPhase(queryFetchResults.asList()), queryFetchResults);
    String scrollId = null;
    if (request.scroll() != null) {
        scrollId = request.scrollId();
    }
    listener.onResponse(new SearchResponse(internalResponse, scrollId, this.scrollId.getContext().length, successfulOps.get(),
        buildTookInMillis(), buildShardFailures()));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void finishHim(SearchPhaseController.ReducedQueryPhase queryPhase) {
    try {
        final InternalSearchResponse internalResponse = searchPhaseController.merge(true, sortedShardDocs, queryPhase, fetchResults);
        String scrollId = null;
        if (request.scroll() != null) {
            scrollId = request.scrollId();
        }
        listener.onResponse(new SearchResponse(internalResponse, scrollId, this.scrollId.getContext().length, successfulOps.get(),
            buildTookInMillis(), buildShardFailures()));
    } catch (Exception e) {
        listener.onFailure(new ReduceSearchPhaseException("fetch", "inner finish failed", e, buildShardFailures()));
    }
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
/**
 * Enriches search hits and completion suggestion hits from <code>sortedDocs</code> using <code>fetchResultsArr</code>,
 * merges suggestions, aggregations and profile results
 *
 * Expects sortedDocs to have top search docs across all shards, optionally followed by top suggest docs for each named
 * completion suggestion ordered by suggestion name
 */
public InternalSearchResponse merge(boolean ignoreFrom, ScoreDoc[] sortedDocs,
                                    ReducedQueryPhase reducedQueryPhase,
                                    AtomicArray<? extends QuerySearchResultProvider> fetchResultsArr) {
    if (reducedQueryPhase.isEmpty()) {
        return InternalSearchResponse.empty();
    }
    List<? extends AtomicArray.Entry<? extends QuerySearchResultProvider>> fetchResults = fetchResultsArr.asList();
    SearchHits hits = getHits(reducedQueryPhase, ignoreFrom, sortedDocs, fetchResultsArr);
    if (reducedQueryPhase.suggest != null) {
        if (!fetchResults.isEmpty()) {
            int currentOffset = hits.getHits().length;
            for (CompletionSuggestion suggestion : reducedQueryPhase.suggest.filter(CompletionSuggestion.class)) {
                final List<CompletionSuggestion.Entry.Option> suggestionOptions = suggestion.getOptions();
                for (int scoreDocIndex = currentOffset; scoreDocIndex < currentOffset + suggestionOptions.size(); scoreDocIndex++) {
                    ScoreDoc shardDoc = sortedDocs[scoreDocIndex];
                    QuerySearchResultProvider searchResultProvider = fetchResultsArr.get(shardDoc.shardIndex);
                    if (searchResultProvider == null) {
                        continue;
                    }
                    FetchSearchResult fetchResult = searchResultProvider.fetchResult();
                    int fetchResultIndex = fetchResult.counterGetAndIncrement();
                    if (fetchResultIndex < fetchResult.hits().internalHits().length) {
                        SearchHit hit = fetchResult.hits().internalHits()[fetchResultIndex];
                        CompletionSuggestion.Entry.Option suggestOption =
                            suggestionOptions.get(scoreDocIndex - currentOffset);
                        hit.score(shardDoc.score);
                        hit.shard(fetchResult.shardTarget());
                        suggestOption.setHit(hit);
                    }
                }
                currentOffset += suggestionOptions.size();
            }
            assert currentOffset == sortedDocs.length : "expected no more score doc slices";
        }
    }
    return reducedQueryPhase.buildResponse(hits);
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
public void testSkipPhase() throws IOException {
    MockSearchPhaseContext mockSearchPhaseContext = new MockSearchPhaseContext(1);
    mockSearchPhaseContext.searchTransport = new SearchTransportService(
        Settings.builder().put("search.remote.connect", false).build(), null, null) {

        @Override
        void sendExecuteMultiSearch(MultiSearchRequest request, SearchTask task, ActionListener<MultiSearchResponse> listener) {
          fail("no collapsing here");
        }
    };

    SearchHits hits = new SearchHits(new SearchHit[]{new SearchHit(1, "ID", new Text("type"),
        Collections.singletonMap("someField", new SearchHitField("someField", Collections.singletonList(null)))),
        new SearchHit(2, "ID2", new Text("type"),
            Collections.singletonMap("someField", new SearchHitField("someField", Collections.singletonList(null))))}, 1, 1.0F);
    InternalSearchResponse internalSearchResponse = new InternalSearchResponse(hits, null, null, null, false, null, 1);
    SearchResponse response = mockSearchPhaseContext.buildSearchResponse(internalSearchResponse, null);
    AtomicReference<SearchResponse> reference = new AtomicReference<>();
    ExpandSearchPhase phase = new ExpandSearchPhase(mockSearchPhaseContext, response, r ->
        new SearchPhase("test") {
            @Override
            public void run() throws IOException {
                reference.set(r);
            }
        }
    );
    phase.run();
    mockSearchPhaseContext.assertNoFailure();
    assertNotNull(reference.get());
    assertEquals(1, mockSearchPhaseContext.phasesExecuted.get());
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
@Override
protected void moveToSecondPhase() throws Exception {
    final InternalSearchResponse internalResponse = searchPhaseController.merge(SearchPhaseController.EMPTY_DOCS, firstResults,
            (AtomicArray<? extends FetchSearchResultProvider>) AtomicArray.empty(), request);
    String scrollId = null;
    if (request.scroll() != null) {
        scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, ImmutableMap.of("total_hits", Long.toString
                (internalResponse.hits().totalHits())));
    }
    listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(), buildTookInMillis(), buildShardFailures()));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
public SearchResponse(InternalSearchResponse internalResponse, String scrollId, int totalShards, int successfulShards, long tookInMillis, ShardSearchFailure[] shardFailures) {
    this.internalResponse = internalResponse;
    this.scrollId = scrollId;
    this.totalShards = totalShards;
    this.successfulShards = successfulShards;
    this.tookInMillis = tookInMillis;
    this.shardFailures = shardFailures;
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void innerFinishHim() throws Exception {
    ScoreDoc[] sortedShardList = searchPhaseController.sortDocs(true, queryFetchResults);
    final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryFetchResults,
        queryFetchResults, request);
    String scrollId = null;
    if (request.scroll() != null) {
        scrollId = request.scrollId();
    }
    listener.onResponse(new SearchResponse(internalResponse, scrollId, this.scrollId.getContext().length, successfulOps.get(),
        buildTookInMillis(), buildShardFailures()));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
@Override
protected void moveToSecondPhase() throws Exception {
    // no need to sort, since we know we have no hits back
    final InternalSearchResponse internalResponse = searchPhaseController.merge(SearchPhaseController.EMPTY_DOCS, firstResults,
            (AtomicArray<? extends FetchSearchResultProvider>) AtomicArray.empty(), request);
    String scrollId = null;
    if (request.scroll() != null) {
        scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null);
    }
    listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(), buildTookInMillis(), buildShardFailures()));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void innerFinishHim() {
    InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults, fetchResults, request);
    String scrollId = null;
    if (request.scroll() != null) {
        scrollId = request.scrollId();
    }
    listener.onResponse(new SearchResponse(internalResponse, scrollId, this.scrollId.getContext().length, successfulOps.get(),
        buildTookInMillis(), buildShardFailures()));
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void finishHim() {
    threadPool.executor(ThreadPool.Names.SEARCH).execute(new ActionRunnable<SearchResponse>(listener) {
        @Override
        public void doRun() throws IOException {
            final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults,
                fetchResults, request);
            String scrollId = null;
            if (request.scroll() != null) {
                scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null);
            }
            listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(),
                buildTookInMillis(), buildShardFailures()));
            releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
        }

        @Override
        public void onFailure(Throwable t) {
            try {
                ReduceSearchPhaseException failure = new ReduceSearchPhaseException("merge", "", t, buildShardFailures());
                if (logger.isDebugEnabled()) {
                    logger.debug("failed to reduce search", failure);
                }
                super.onFailure(failure);
            } finally {
                releaseIrrelevantSearchContexts(queryResults, docIdsToLoad);
            }
        }
    });
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
private void finishHim() {
    threadPool.executor(ThreadPool.Names.SEARCH).execute(new ActionRunnable<SearchResponse>(listener) {
        @Override
        public void doRun() throws IOException {
            final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, firstResults,
                fetchResults, request);
            String scrollId = null;
            if (request.scroll() != null) {
                scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null);
            }
            listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps,
                successfulOps.get(), buildTookInMillis(), buildShardFailures()));
            releaseIrrelevantSearchContexts(firstResults, docIdsToLoad);
        }

        @Override
        public void onFailure(Throwable t) {
            try {
                ReduceSearchPhaseException failure = new ReduceSearchPhaseException("fetch", "", t, buildShardFailures());
                if (logger.isDebugEnabled()) {
                    logger.debug("failed to reduce search", failure);
                }
                super.onFailure(failure);
            } finally {
                releaseIrrelevantSearchContexts(firstResults, docIdsToLoad);
            }
        }
    });
}
 

import org.elasticsearch.search.internal.InternalSearchResponse; //导入依赖的package包/类
@Override
protected void moveToSecondPhase() throws Exception {
    threadPool.executor(ThreadPool.Names.SEARCH).execute(new ActionRunnable<SearchResponse>(listener) {
        @Override
        public void doRun() throws IOException {
            boolean useScroll = request.scroll() != null;
            sortedShardList = searchPhaseController.sortDocs(useScroll, firstResults);
            final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, firstResults,
                firstResults, request);
            String scrollId = null;
            if (request.scroll() != null) {
                scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null);
            }
            listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(),
                buildTookInMillis(), buildShardFailures()));
        }

        @Override
        public void onFailure(Throwable t) {
            ReduceSearchPhaseException failure = new ReduceSearchPhaseException("merge", "", t, buildShardFailures());
            if (logger.isDebugEnabled()) {
                logger.debug("failed to reduce search", failure);
            }
            super.onFailure(failure);
        }
    });
}