ES5.6 search流程与scroll

2018-01-19 本文已影响197人 YG_9013

ES search大致流程

请求转化，由RestSearchAction转为TransportSearchAction，执行其doExecute()方法

1、如果查询请求的索引含有正则表达式和别名，找出具体的索引

indices = indexNameExpressionResolver.concreteIndices(clusterState, searchRequest.indicesOptions(),
            timeProvider.getAbsoluteStartMillis(), localIndices.indices());

2、找出routing，并根据routing找出shard

Map<String, Set<String>> routingMap = indexNameExpressionResolver.resolveSearchRouting(clusterState, searchRequest.routing(),
      searchRequest.indices());
GroupShardsIterator<ShardIterator> localShardsIterator = clusterService.operationRouting().searchShards(clusterState,
        concreteIndices, routingMap, searchRequest.preference());

3、解析请求中的权重boost

Map<String, Float> concreteIndexBoosts = resolveIndexBoosts(searchRequest, clusterState);

4、根据请求的类型选择不同的query类

switch(searchRequest.searchType()) {
    case DFS_QUERY_THEN_FETCH:
        searchAsyncAction = new SearchDfsQueryThenFetchAsyncAction(logger, searchTransportService, connectionLookup,
            aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators,
            timeProvider, clusterStateVersion, task);
        break;
    case QUERY_AND_FETCH:
    case QUERY_THEN_FETCH:
        searchAsyncAction = new SearchQueryThenFetchAsyncAction(logger, searchTransportService, connectionLookup,
            aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators,
            timeProvider, clusterStateVersion, task);
        break;
    default:
        throw new IllegalStateException("Unknown search type: [" + searchRequest.searchType() + "]");
}

插播一条DFS_QUERY_THEN_FETCH与QUERY_THEN_FETCH的区别。

QUERY_THEN_FETCH
请求执行有两个阶段。第一个阶段就是查询所有相关的shards。所有的shards执行请求并根据form和size返回一个排好序的结果。协调节点获取到所有的信息后merge并resort，然后根据form和size取出结果。第二个阶段则是根据第一阶段的结果在相应的shard上取出数据。
DFS_QUERY_THEN_FETCH
与QUERY_THEN_FETCH几乎一样。只有在第一阶段为了更精确的打分，计算的是分布式的term frequencies。

5、SearchQueryThenFetchAsyncAction中则是执行标准的查询流程：

query
fetch
merge

SearchQueryThenFetchAsyncAction标准查询流程

1、Query

首先遍历shard，执行shard查询请求。

for (final SearchShardIterator shardIt : shardsIts) {
    shardIndex++;
    final ShardRouting shard = shardIt.nextOrNull();
    if (shard != null) {
        performPhaseOnShard(shardIndex, shardIt, shard);
    } else {
        // really, no shards active in this group
        onShardFailure(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
    }
}

每个shard通过SearchTransportService发送query请求：

getSearchTransport().sendExecuteQuery(getConnection(shardIt.getClusterAlias(), shard.currentNodeId()),
        buildShardSearchRequest(shardIt), getTask(), listener);

本地则是执行TransportService的sendLocalRequest方法.最终调用的是

handler.messageReceived(request, channel);

此处的channel是DirectResponseChannel，而此处的handler则是TransportSearchAction 初始化的时候写入SearchTransportService中的。

transportService.registerRequestHandler(QUERY_ACTION_NAME, ShardSearchTransportRequest::new, ThreadPool.Names.SEARCH,
  new TaskAwareTransportRequestHandler<ShardSearchTransportRequest>() {
    @Override
    public void messageReceived(ShardSearchTransportRequest request, TransportChannel channel, Task task) throws Exception {
        SearchPhaseResult result = searchService.executeQueryPhase(request, (SearchTask)task);
        channel.sendResponse(result);
    }
});

因此，最终执行的是searchService的executeQueryPhase方法。

真正的query本地查询：

1）生成一个searchContext，然后将request写入context：

final DefaultSearchContext searchContext = new DefaultSearchContext(idGenerator.incrementAndGet(), request, shardTarget,
        engineSearcher, indexService, indexShard, bigArrays, threadPool.estimatedTimeInMillisCounter(), timeout, fetchPhase);

parseSource(context, request.source());

2）queryPhase预处理。比如说如果请求没有query，默认为match_all,如果没有boost，采用默认的值等等

queryPhase.preProcess(context);

3）执行查询

loadOrExecuteQueryPhase(request, context);

loadOrExecuteQueryPhase中则是调用queryPhase执行查询：

queryPhase.execute(context);

在queryPhase中执行真正的query查询：
1）重写query，比如说讲一个prefixQuery转化为多个包含termQuery的BoolQuery
assert query == searcher.rewrite(query); // already rewritten

2）将各个子查询求用Collector包裹，Collector传给lucene，进行真正的lucene查询

if (terminateAfterSet) {
    final Collector child = collector;
    // throws Lucene.EarlyTerminationException when given count is reached
    collector = Lucene.wrapCountBasedEarlyTerminatingCollector(collector, searchContext.terminateAfter());
    if (doProfile) {
        collector = new InternalProfileCollector(collector, CollectorResult.REASON_SEARCH_TERMINATE_AFTER_COUNT,
                Collections.singletonList((InternalProfileCollector) child));
    }
}

if (searchContext.parsedPostFilter() != null) {
    final Collector child = collector;
    // this will only get applied to the actual search collector and not
    // to any scoped collectors, also, it will only be applied to the main collector
    // since that is where the filter should only work
    final Weight filterWeight = searcher.createNormalizedWeight(searchContext.parsedPostFilter().query(), false);
    collector = new FilteredCollector(collector, filterWeight);
    if (doProfile) {
        collector = new InternalProfileCollector(collector, CollectorResult.REASON_SEARCH_POST_FILTER,
                Collections.singletonList((InternalProfileCollector) child));
    }
}
// plug in additional collectors, like aggregations
final List<Collector> subCollectors = new ArrayList<>();
subCollectors.add(collector);
subCollectors.addAll(searchContext.queryCollectors().values());
collector = MultiCollector.wrap(subCollectors);

//最后包裹一个Cancellable
if (collector != null) {
    final Collector child = collector;
    collector = new CancellableCollector(searchContext.getTask()::isCancelled, searchContext.lowLevelCancellation(), collector);
    if (doProfile) {
        collector = new InternalProfileCollector(collector, CollectorResult.REASON_SEARCH_CANCELLED,
            Collections.singletonList((InternalProfileCollector) child));
    }
}

3）执行lucene的查询

searcher.search(query, collector);

4）将查询结果以TopDocs的方式返回

queryResult.topDocs(topDocsCallable.call(), sortValueFormats);

topDocsCallable是根据不同的查询封装的。举个例子：

if (query.getClass() == MatchAllDocsQuery.class) {
    collector = null;
    topDocsCallable = new Callable<TopDocs>() {
        @Override
        public TopDocs call() throws Exception {
            int count = searcher.getIndexReader().numDocs();
            return new TopDocs(count, Lucene.EMPTY_SCORE_DOCS, 0);
        }
    };
} else if (query.getClass() == TermQuery.class && searcher.getIndexReader().hasDeletions() == false) {
    final Term term = ((TermQuery) query).getTerm();
    collector = null;
    topDocsCallable = new Callable<TopDocs>() {
        @Override
        public TopDocs call() throws Exception {
            int count = 0;
            for (LeafReaderContext context : searcher.getIndexReader().leaves()) {
                count += context.reader().docFreq(term);
            }
            return new TopDocs(count, Lucene.EMPTY_SCORE_DOCS, 0);
        }
    };
}

2、fetch阶段（+merge阶段）：

执行的是FetchSearchPhase的dorun方法。
1）首先reduce

final SearchPhaseController.ReducedQueryPhase reducedQueryPhase = resultConsumer.reduce();

resultConsumer 是在构造SearchQueryThenFetchAsyncAction的时候创建的，执行的是SearchPhaseController的reducedQueryPhase方法。

searchPhaseController.newSearchPhaseResults(request, shardsIts.size())

2）查找reduce出的文档id，按照shardId存放

final IntArrayList[] docIdsToLoad = searchPhaseController.fillDocIdsToLoad(numShards, reducedQueryPhase.scoreDocs);

如果是scroll查询，将lastEmittedDocPerShard存入ShardFetchSearchRequest：

final ScoreDoc[] lastEmittedDocPerShard = isScrollSearch ?
               searchPhaseController.getLastEmittedDocPerShard(reducedQueryPhase, numShards)
               : null;
ShardFetchSearchRequest fetchSearchRequest = createFetchRequest(queryResult.queryResult().getRequestId(), i, entry,
                            lastEmittedDocPerShard, searchShardTarget.getOriginalIndices());

3）针对于每个shard，创建fetch请求，并执行

ShardFetchSearchRequest fetchSearchRequest = createFetchRequest(queryResult.queryResult().getRequestId(), i, entry,
    lastEmittedDocPerShard, searchShardTarget.getOriginalIndices());
executeFetch(i, searchShardTarget, counter, fetchSearchRequest, queryResult.queryResult(),
    connection);

本地节点收到请求后执行的是SearchService的executeFetchPhase方法。这主要涉及些lucene的东西，不在详细解析。
4）当从所有的shard完获取完结果之后执行：

final Runnable finishPhase = ()
-> moveToNextPhase(searchPhaseController, scrollId, reducedQueryPhase, queryAndFetchOptimization ?
queryResults : fetchResults);

moveToNextPhase中开始执行merge：

final InternalSearchResponse internalResponse = searchPhaseController.merge(context.getRequest().scroll() != null,
reducedQueryPhase, fetchResultsArr.asList(), fetchResultsArr::get);

merge结束后，通过nextPhaseFactory将结果发送出去:

    (response, scrollId) -> new ExpandSearchPhase(context, response,
                (finalResponse) -> sendResponsePhase(finalResponse, scrollId, context))

  private static SearchPhase sendResponsePhase(InternalSearchResponse response, String scrollId, SearchPhaseContext context) {
    return new SearchPhase("response") {
        @Override
        public void run() throws IOException {
            context.onResponse(context.buildSearchResponse(response, scrollId));
        }
    };
}

context则是SearchQueryThenFetchAsyncAction的父类AbstractSearchAsyncAction，调用buildSearchResponse方法构造response，并回调listener。

public final void onResponse(SearchResponse response) {
    listener.onResponse(response);
}

scroll查询

scroll查询原理

在第一次查询时，记录上一次查询的位置，在接下来的查询中获取到上次查询的位置，接着查询。

比如说将查询order by time offset 0 limit 100，改写成order by time where time>0 limit 100，记录最后一个$time_max,接下来的查询order by time offset 100 limit 100，改写成order by time where time>$time_max limit 100。如此往复，可以看出scroll是一个常量查询延迟和开销。
这个从源码中也可以看出：

if (returnsDocsInOrder(query, searchContext.sort())) {
    if (scrollContext.totalHits == -1) {
        // first round
        assert scrollContext.lastEmittedDoc == null;
        // there is not much that we can optimize here since we want to collect all
        // documents in order to get the total number of hits
    } else {
        // now this gets interesting: since we sort in index-order, we can directly
        // skip to the desired doc and stop collecting after ${size} matches
        if (scrollContext.lastEmittedDoc != null) {
        if (scrollContext.lastEmittedDoc != null) {
            BooleanQuery bq = new BooleanQuery.Builder()
                .add(query, BooleanClause.Occur.MUST)
                .add(new MinDocQuery(after.doc + 1), BooleanClause.Occur.FILTER)
                .build();
            query = bq;
        }
        searchContext.terminateAfter(numDocs);
    }
}

当scrollContext.lastEmittedDoc不为空也就是查询scroll的下一页时，将查询变为一个带偏移量的bool查询。
在本地查出数据后，更新scrollContext的值：

if (scrollContext != null) {
    if (scrollContext.totalHits == -1) {
        // first round
        scrollContext.totalHits = topDocs.totalHits;
        scrollContext.maxScore = topDocs.getMaxScore();
    } else {
        // subsequent round: the total number of hits and
        // the maximum score were computed on the first round
        topDocs.totalHits = scrollContext.totalHits;
        topDocs.setMaxScore(scrollContext.maxScore);
    }
    if (searchContext.request().numberOfShards() == 1) {
        // if we fetch the document in the same roundtrip, we already know the last emitted doc
        if (topDocs.scoreDocs.length > 0) {
            // set the last emitted doc
            scrollContext.lastEmittedDoc = topDocs.scoreDocs[topDocs.scoreDocs.length - 1];
        }
    }
}

我们看到上述只更新了scrollContext的totalHits和maxScore，而lastEmittedDoc则是在fetch阶段中更新的：
先将lastEmittedDocPerShard存入ShardFetchSearchRequest：

final ScoreDoc[] lastEmittedDocPerShard = isScrollSearch ?
               searchPhaseController.getLastEmittedDocPerShard(reducedQueryPhase, numShards)
               : null;
ShardFetchSearchRequest fetchSearchRequest = createFetchRequest(queryResult.queryResult().getRequestId(), i, entry,
                            lastEmittedDocPerShard, searchShardTarget.getOriginalIndices());

然后每个节点在接收到fetch信息后，执行SearchService的executeFetchPhase方法：

 if (request.lastEmittedDoc() != null) {
                context.scrollContext().lastEmittedDoc = request.lastEmittedDoc();
 }

当使用scroll请求时，scrollID是不变的，从SearchScrollAsyncAction代码中可以看出：

protected final void sendResponse(SearchPhaseController.ReducedQueryPhase queryPhase,
                                  final AtomicArray<? extends SearchPhaseResult> fetchResults) {
    try {
        final InternalSearchResponse internalResponse = searchPhaseController.merge(true, queryPhase, fetchResults.asList(),
            fetchResults::get);
        // the scroll ID never changes we always return the same ID. This ID contains all the shards and their context ids
        // such that we can talk to them abgain in the next roundtrip.
        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()));
    }
}