flowchart TB
    subgraph Client["客户端层"]
        RESTClient[REST Client]
        JavaClient[Java Client]
        HTTPRequest[HTTP Request<br/>POST /index/_search]
    end

    subgraph RESTLayer["REST接口层 (协调节点)"]
        RestSearchAction[RestSearchAction<br/>解析HTTP请求]
        SearchRequestBuilder[SearchRequest构建<br/>解析Query DSL]
    end

    subgraph ActionLayer["Action传输层 (协调节点)"]
        TransportSearchAction[TransportSearchAction<br/>分布式协调器]
        SearchPhaseProvider[SearchPhaseProvider<br/>阶段提供者]
        SearchAsyncAction[SearchQueryThenFetch<br/>AsyncAction<br/>异步执行器]
    end

    subgraph Coordinator["协调阶段执行器 (协调节点)"]
        CanMatchPhase[CanMatchPreFilterPhase<br/>预过滤阶段]
        QueryPhaseCoord[QueryPhase协调<br/>并发查询所有分片]
        ReducePhase[ReducePhase<br/>SearchPhaseController<br/>结果归并]
        FetchPhaseCoord[FetchPhase协调<br/>并发获取文档]
    end

    subgraph TransportLayer["传输服务层"]
        SearchTransportService[SearchTransportService<br/>RPC通信]
        ConnectionMgr[Connection Manager<br/>节点连接管理]
    end

    subgraph DataNode1["数据节点 1"]
        SService1[SearchService]
        SContext1[SearchContext]
        QPExec1[QueryPhase.execute]
        FPExec1[FetchPhase.execute]
        Shard1[IndexShard]
        Lucene1[Lucene Searcher]
    end

    subgraph DataNode2["数据节点 2"]
        SService2[SearchService]
        SContext2[SearchContext]
        QPExec2[QueryPhase.execute]
        FPExec2[FetchPhase.execute]
        Shard2[IndexShard]
        Lucene2[Lucene Searcher]
    end

    subgraph DataNodeN["数据节点 N"]
        SServiceN[SearchService]
        SContextN[SearchContext]
        QPExecN[QueryPhase.execute]
        FPExecN[FetchPhase.execute]
        ShardN[IndexShard]
        LuceneN[Lucene Searcher]
    end

    RESTClient --> HTTPRequest
    JavaClient --> HTTPRequest
    HTTPRequest --> RestSearchAction

    RestSearchAction --> SearchRequestBuilder
    SearchRequestBuilder --> TransportSearchAction

    TransportSearchAction --> SearchPhaseProvider
    SearchPhaseProvider --> SearchAsyncAction
    SearchAsyncAction --> CanMatchPhase

    CanMatchPhase --> |过滤后的分片列表| QueryPhaseCoord

    QueryPhaseCoord --> SearchTransportService
    SearchTransportService --> ConnectionMgr

    ConnectionMgr --> |ShardSearchRequest| SService1
    ConnectionMgr --> |ShardSearchRequest| SService2
    ConnectionMgr --> |ShardSearchRequest| SServiceN

    SService1 --> SContext1 --> QPExec1 --> Shard1 --> Lucene1
    SService2 --> SContext2 --> QPExec2 --> Shard2 --> Lucene2
    SServiceN --> SContextN --> QPExecN --> ShardN --> LuceneN

    Lucene1 --> |QuerySearchResult| ReducePhase
    Lucene2 --> |QuerySearchResult| ReducePhase
    LuceneN --> |QuerySearchResult| ReducePhase

    ReducePhase --> |Top K docIds| FetchPhaseCoord
    FetchPhaseCoord --> SearchTransportService

    SearchTransportService --> |ShardFetchRequest| SService1
    SearchTransportService --> |ShardFetchRequest| SService2
    SearchTransportService --> |ShardFetchRequest| SServiceN

    SService1 --> FPExec1 --> Lucene1
    SService2 --> FPExec2 --> Lucene2
    SServiceN --> FPExecN --> LuceneN

    Lucene1 --> |FetchSearchResult| SearchAsyncAction
    Lucene2 --> |FetchSearchResult| SearchAsyncAction
    LuceneN --> |FetchSearchResult| SearchAsyncAction

    SearchAsyncAction --> |SearchResponse| RestSearchAction
    RestSearchAction --> |JSON Response| Client

    style RESTLayer fill:#E1F5FF
    style ActionLayer fill:#FFE1E1
    style Coordinator fill:#E1FFE1
    style TransportLayer fill:#F5E1FF
    style DataNode1 fill:#FFF4E1
    style DataNode2 fill:#FFF4E1
    style DataNodeN fill:#FFF4E1

public class SearchService extends AbstractLifecycleComponent {
    // 线程池
    private final ThreadPool threadPool;

    // 搜索上下文管理
    private final ConcurrentMapLong<ReaderContext> activeReaders;

    // Fetch 阶段处理
    private final FetchPhase fetchPhase;

    // 执行 Query Phase
    public void executeQueryPhase(
        ShardSearchRequest request,
        SearchShardTask task,
        ActionListener<SearchPhaseResult> listener
    );

    // 执行 Fetch Phase
    public void executeFetchPhase(
        ShardFetchRequest request,
        SearchShardTask task,
        ActionListener<FetchSearchResult> listener
    );

    // 创建搜索上下文
    private SearchContext createContext(
        ReaderContext reader,
        ShardSearchRequest request,
        SearchShardTask task
    );
}

public interface SearchContext extends Releasable {
    // 查询信息
    SearchShardTarget shardTarget();
    ShardSearchRequest request();
    Query query();

    // 搜索器
    ContextIndexSearcher searcher();

    // 结果
    QuerySearchResult queryResult();
    FetchSearchResult fetchResult();

    // 聚合上下文
    SearchContextAggregations aggregations();

    // 排序
    SortAndFormats sort();

    // 分页
    int from();
    int size();

    // 超时
    TimeValue timeout();
    boolean isTimedOut();
}

public class QueryPhase {

    public static void execute(SearchContext searchContext) {
        // 1. 预处理聚合
        AggregationPhase.preProcess(searchContext);

        // 2. 构建 Collector
        List<Collector> collectors = new ArrayList<>();
        collectors.add(new TopDocsCollector());
        if (searchContext.aggregations() != null) {
            collectors.add(searchContext.aggregations().collector());
        }

        // 3. 执行搜索
        MultiCollector multiCollector = MultiCollector.wrap(collectors);
        searchContext.searcher().search(
            searchContext.query(),
            multiCollector
        );

        // 4. 后处理
        RescorePhase.execute(searchContext);  // Rescore
        SuggestPhase.execute(searchContext);  // Suggest
    }
}

public final class FetchPhase {

    public void execute(
        SearchContext context,
        int[] docIdsToLoad
    ) {
        // 1. 加载文档
        SearchHits hits = buildSearchHits(
            context,
            docIdsToLoad
        );

        // 2. 执行 SubPhases
        for (FetchSubPhase subPhase : fetchSubPhases) {
            subPhase.process(context, hits);
        }

        // 3. 设置结果
        context.fetchResult().shardResult(hits);
    }

    private SearchHits buildSearchHits(
        SearchContext context,
        int[] docIdsToLoad
    ) {
        SearchHit[] hits = new SearchHit[docIdsToLoad.length];

        for (int i = 0; i < docIdsToLoad.length; i++) {
            int docId = docIdsToLoad[i];

            // 读取文档
            Document doc = context.searcher().doc(docId);

            // 构建 SearchHit
            hits[i] = new SearchHit(docId);
            hits[i].sourceRef(doc.getSource());
            hits[i].score(scores[i]);
        }

        return new SearchHits(hits, totalHits, maxScore);
    }
}

1. Query Phase: 所有分片并行查询,返回 Top K docIds
2. Reduce: 协调节点归并,选出全局 Top K
3. Fetch Phase: 获取 Top K 文档内容

1. DFS Phase: 收集所有分片的词频统计
2. Query Phase: 基于全局词频计算分数
3. Reduce: 归并结果
4. Fetch Phase: 获取文档

classDiagram
    class SearchContext {
        <<abstract>>
        +ShardSearchContextId id
        +SearchType searchType
        +SearchShardTarget shardTarget
        +ShardSearchRequest request
        +ReaderContext readerContext
        +Query query
        +ParsedQuery postFilter
        +SortAndFormats sort
        +int from
        +int size
        +int trackTotalHitsUpTo
        +int terminateAfter
        +TimeValue timeout
        +SearchContextAggregations aggregations
        +SearchHighlightContext highlight
        +FetchSourceContext fetchSourceContext
        +StoredFieldsContext storedFieldsContext
        +QuerySearchResult queryResult
        +FetchSearchResult fetchResult
        +preProcess()
        +buildFilteredQuery(Query)
        +close()
    }

    class ShardSearchContextImpl {
        -ReaderContext readerContext
        -ShardSearchRequest request
        -SearchShardTarget shardTarget
        -ContextIndexSearcher searcher
        -SearchExecutionContext searchExecutionContext
        -QuerySearchResult queryResult
        -FetchSearchResult fetchResult
        -ScrollContext scrollContext
        -int from
        -int size
        -Query query
        -SortAndFormats sort
        +execute()
    }

    class ReaderContext {
        -ShardSearchContextId id
        -IndexService indexService
        -IndexShard indexShard
        -Engine.SearcherSupplier searcherSupplier
        -AtomicLong keepAlive
        -AtomicLong lastAccessTime
        -boolean singleSession
        -List~Releasable~ onCloses
        -Map~String,Object~ context
        +incRef()
        +tryIncRef()
        +decRef()
        +updateLastAccessTime()
        +close()
    }

    class ContextIndexSearcher {
        -Query cancellable
        -Runnable checkCancelled
        -CircuitBreaker circuitBreaker
        +search(Query, Collector)
        +searchAfter(ScoreDoc, Query, int)
        +searchLeaf(LeafReaderContext, Weight, Collector)
    }

    SearchContext <|-- ShardSearchContextImpl
    ShardSearchContextImpl --> ReaderContext : holds
    ShardSearchContextImpl --> ContextIndexSearcher : uses
    ReaderContext --> IndexShard : references

创建 → preProcess() → Query Phase → Fetch Phase → close()

// 增加引用
boolean success = readerContext.tryIncRef();

// 使用
try {
    Engine.Searcher searcher = readerContext.acquireSearcher("search");
    // ... 执行搜索
} finally {
    // 减少引用
    readerContext.decRef();
}

classDiagram
    class SearchRequest {
        -String[] indices
        -SearchType searchType
        -SearchSourceBuilder source
        -String routing
        -String preference
        -Boolean requestCache
        -Boolean allowPartialSearchResults
        -TimeValue scrollKeepAlive
        -int maxConcurrentShardRequests
        -Integer preFilterShardSize
        -int batchedReduceSize
        -IndicesOptions indicesOptions
        +indices(String[])
        +source(SearchSourceBuilder)
        +validate()
    }

    class SearchSourceBuilder {
        -QueryBuilder query
        -int from
        -int size
        -List~SortBuilder~ sorts
        -List~AggregationBuilder~ aggregations
        -HighlightBuilder highlight
        -FetchSourceContext fetchSource
        -SearchAfterBuilder searchAfter
        -TimeValue timeout
        -boolean explain
        -boolean version
        -Integer trackTotalHitsUpTo
        -Integer terminateAfter
        +query(QueryBuilder)
        +from(int)
        +size(int)
        +sort(SortBuilder)
        +aggregation(AggregationBuilder)
        +toXContent(XContentBuilder)
    }

    class SearchResponse {
        -SearchHits hits
        -InternalAggregations aggregations
        -Suggest suggest
        -SearchProfileResults profileResults
        -boolean timedOut
        -Boolean terminatedEarly
        -int numReducePhases
        -String scrollId
        -BytesReference pointInTimeId
        -int totalShards
        -int successfulShards
        -int skippedShards
        -ShardSearchFailure[] shardFailures
        -Clusters clusters
        -long tookInMillis
        +getHits()
        +getAggregations()
        +getTook()
    }

    class SearchHits {
        -TotalHits totalHits
        -float maxScore
        -SearchHit[] hits
        +iterator()
    }

    class SearchHit {
        -String id
        -float score
        -Map~String,Object~ sourceAsMap
        -String index
        -long version
        -long seqNo
        -long primaryTerm
        -Object[] sortValues
        -Map~String,HighlightField~ highlightFields
        -Map~String,SearchHitField~ fields
        +getId()
        +getScore()
        +getSourceAsMap()
    }

    SearchRequest --> SearchSourceBuilder : contains
    SearchResponse --> SearchHits : contains
    SearchHits --> SearchHit : contains many

public class SearchRequest extends LegacyActionRequest {
    // 目标索引
    private String[] indices = Strings.EMPTY_ARRAY;

    // 搜索类型 (QUERY_THEN_FETCH, DFS_QUERY_THEN_FETCH)
    private SearchType searchType = SearchType.DEFAULT;

    // 搜索查询体
    private SearchSourceBuilder source;

    // 路由控制
    private String routing;

    // 偏好设置（_local, _primary, _replica, custom string）
    private String preference;

    // 是否使用请求缓存
    private Boolean requestCache;

    // 是否允许部分结果
    private Boolean allowPartialSearchResults;

    // Scroll 保持时间
    private TimeValue scrollKeepAlive;

    // 并发分片请求数量
    private int maxConcurrentShardRequests = 5;

    // 预过滤分片大小阈值
    private Integer preFilterShardSize = 128;

    // 批量归并大小
    private int batchedReduceSize = 512;

    // 索引选项
    private IndicesOptions indicesOptions;
}

public class SearchSourceBuilder {
    private QueryBuilder query;              // 查询条件
    private int from = 0;                     // 分页起始位置
    private int size = 10;                    // 返回结果数量
    private List<SortBuilder<?>> sorts;       // 排序规则
    private List<AggregationBuilder> aggregations; // 聚合
    private List<String> storedFields;        // 返回的 stored 字段
    private List<String> docValueFields;      // 返回的 doc_value 字段
    private FetchSourceContext fetchSource;   // _source 过滤
    private HighlightBuilder highlight;       // 高亮
    private SearchAfterBuilder searchAfter;   // search_after 分页
    private TimeValue timeout;                // 查询超时时间
    private boolean explain;                  // 是否返回评分解释
    private boolean version;                  // 是否返回版本号
    private boolean trackScores;              // 是否计算分数
    private Integer trackTotalHitsUpTo;       // 统计总数上限
    private Integer terminateAfter;           // 提前终止文档数
}

public class SearchResponse extends ActionResponse {
    // 搜索结果
    private final SearchHits hits;

    // 聚合结果
    private final InternalAggregations aggregations;

    // 搜索建议结果
    private final Suggest suggest;

    // 性能分析结果
    private final SearchProfileResults profileResults;

    // 是否超时
    private final boolean timedOut;

    // 是否提前终止
    private final Boolean terminatedEarly;

    // 归并阶段数量
    private final int numReducePhases;

    // Scroll ID
    private final String scrollId;

    // Point-in-Time ID
    private final BytesReference pointInTimeId;

    // 分片统计
    private final int totalShards;
    private final int successfulShards;
    private final int skippedShards;
    private final ShardSearchFailure[] shardFailures;

    // 集群统计 (跨集群搜索)
    private final Clusters clusters;

    // 耗时
    private final long tookInMillis;
}

// TransportSearchAction - 搜索操作的传输层动作
public class TransportSearchAction extends HandledTransportAction<SearchRequest, SearchResponse> {

    @Override
    protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<SearchResponse> listener) {
        // 1. 解析和重写请求
        final long relativeStartNanos = System.nanoTime();
        final SearchTimeProvider timeProvider = new SearchTimeProvider(
            searchRequest.getOrCreateAbsoluteStartMillis(),
            relativeStartNanos,
            System::nanoTime
        );

        // 2. 解析目标索引
        final ClusterState clusterState = clusterService.state();
        final Index[] concreteIndices = indexNameExpressionResolver.concreteIndices(
            clusterState,
            searchRequest.indicesOptions(),
            searchRequest.indices()
        );

        // 3. 计算目标分片
        final Map<String, AliasFilter> aliasFilter = buildPerIndexAliasFilter(
            searchRequest,
            clusterState,
            concreteIndices,
            remoteClusterIndices
        );

        // 4. 创建搜索任务上下文
        final SearchTask searchTask = (SearchTask) task;
        final SearchShardIterator[] shardIterators = buildShardIterators(
            clusterState,
            concreteIndices,
            aliasFilter,
            searchRequest.routing(),
            searchRequest.preference()
        );

        // 5. 执行搜索
        searchAsyncAction(
            searchTask,
            searchRequest,
            shardIterators,
            timeProvider,
            listener
        ).start();
    }

    private void searchAsyncAction(
        SearchTask task,
        SearchRequest searchRequest,
        SearchShardIterator[] shardIterators,
        SearchTimeProvider timeProvider,
        ActionListener<SearchResponse> listener
    ) {
        // 创建异步搜索执行器
        SearchQueryThenFetchAsyncAction action = new SearchQueryThenFetchAsyncAction(
            task,
            searchRequest,
            shardIterators,
            timeProvider,
            searchPhaseController,
            executor,
            listener
        );
        return action;
    }
}

// 1. 高效分页
{
  "size": 10,
  "sort": [{"timestamp": "desc"}, {"_id": "asc"}],
  "search_after": [1609459200000, "doc#10"]
}

// 2. _source 过滤
{
  "_source": ["title", "author", "publish_date"],
  "query": {"match": {"title": "elasticsearch"}}
}

// 3. 仅统计数量
{
  "size": 0,
  "track_total_hits": true,
  "query": {"match_all": {}}
}

// 4. 使用 Filter (可缓存)
{
  "query": {
    "bool": {
      "filter": [
        {"term": {"status": "published"}},
        {"range": {"date": {"gte": "2024-01-01"}}}
      ]
    }
  }
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant REST as RestSearchAction<br/>(REST层)
    participant Transport as TransportSearchAction<br/>(Action层)
    participant AsyncAction as SearchQueryThenFetch<br/>AsyncAction
    participant TransportSvc as SearchTransportService<br/>(Transport层)
    participant DataNode as SearchService<br/>(数据节点)
    participant Context as SearchContext
    participant QueryPhase as QueryPhase.execute
    participant Lucene as Lucene IndexSearcher
    participant Controller as SearchPhaseController<br/>(归并)
    participant FetchPhase as FetchPhase.execute

    Note over Client,Lucene: REST层处理
    Client->>REST: HTTP POST /index/_search<br/>{query, size, from}
    REST->>REST: 解析URL参数 (index, routing)
    REST->>REST: 解析JSON Body → SearchSourceBuilder
    REST->>REST: 构建SearchRequest对象
    REST->>Transport: client.execute(TransportSearchAction.TYPE, request)

    Note over Transport,AsyncAction: Action层协调
    Transport->>Transport: doExecute(task, searchRequest, listener)
    Transport->>Transport: 解析目标索引 → concreteIndices[]
    Transport->>Transport: 计算分片路由 → SearchShardIterator[]
    Transport->>Transport: 构建别名过滤器 → aliasFilter
    Transport->>AsyncAction: new SearchQueryThenFetchAsyncAction(...)<br/>.start()

    Note over AsyncAction,TransportSvc: 异步执行器启动
    AsyncAction->>AsyncAction: start() → doRun()
    AsyncAction->>AsyncAction: 构建分片索引映射 shardIndexMap

    loop 每个分片
        AsyncAction->>AsyncAction: buildShardSearchRequest(shardIt)
        AsyncAction->>AsyncAction: 应用节点级并发控制 (maxConcurrent)
        AsyncAction->>TransportSvc: sendExecuteQuery(connection, request)
    end

    Note over TransportSvc,DataNode: 传输层 RPC 调用
    TransportSvc->>DataNode: [RPC] ShardSearchRequest

    Note over DataNode,Lucene: 数据节点 Query Phase
    DataNode->>DataNode: executeQueryPhase(request, task, listener)
    DataNode->>DataNode: 获取 IndexShard
    DataNode->>DataNode: rewriteAndFetchShardRequest()<br/>(查询重写)
    DataNode->>DataNode: 提交到 search 线程池

    DataNode->>Context: createContext(readerContext, request)
    Context->>Context: 创建 DefaultSearchContext
    Context->>Context: parseSource() → 解析查询/聚合/排序
    Context->>Context: preProcess() → 前置处理

    DataNode->>QueryPhase: QueryPhase.execute(searchContext)
    QueryPhase->>QueryPhase: AggregationPhase.preProcess()
    QueryPhase->>QueryPhase: 构建 QueryPhaseCollectorManager
    QueryPhase->>QueryPhase: 创建 TopDocsCollector + AggsCollector
    QueryPhase->>Lucene: searcher.search(query, collectorManager)
    Lucene->>Lucene: 执行 Lucene 查询
    Lucene->>Lucene: 收集 Top (size+from) docs
    Lucene-->>QueryPhase: QueryPhaseResult<br/>(docIds, scores, sortValues)
    QueryPhase->>QueryPhase: RescorePhase.execute() (如果有)
    QueryPhase->>QueryPhase: SuggestPhase.execute() (如果有)
    QueryPhase-->>DataNode: queryResult
    DataNode-->>TransportSvc: QuerySearchResult
    TransportSvc-->>AsyncAction: QuerySearchResult

    Note over AsyncAction,Controller: 协调节点归并
    AsyncAction->>AsyncAction: 收集所有分片的 QuerySearchResult
    AsyncAction->>Controller: getNextPhase() → 触发 Reduce
    Controller->>Controller: reducedQueryPhase(queryResults)
    Controller->>Controller: 归并排序所有分片 Top K
    Controller->>Controller: 归并聚合结果 (Aggregations.reduce)
    Controller->>Controller: 选出全局 Top size
    Controller-->>AsyncAction: ReducedQueryPhase<br/>(全局 Top K docIds by shard)

    Note over AsyncAction,FetchPhase: Fetch Phase
    AsyncAction->>AsyncAction: 构建 FetchSearchPhase
    AsyncAction->>AsyncAction: 按分片分组 docIdsToLoad[]

    loop 每个需要fetch的分片
        AsyncAction->>TransportSvc: sendExecuteFetch(connection, fetchRequest)
        TransportSvc->>DataNode: [RPC] ShardFetchRequest<br/>(contextId, docIds[])
        DataNode->>DataNode: executeFetchPhase(request, task, listener)
        DataNode->>DataNode: 查找现有 ReaderContext
        DataNode->>FetchPhase: FetchPhase.execute(context, docIds)
        FetchPhase->>FetchPhase: buildSearchHits(context, docIds)

        loop 每个docId
            FetchPhase->>Lucene: searcher.doc(docId) → 读取 StoredFields
            FetchPhase->>FetchPhase: 提取 _source
            FetchPhase->>FetchPhase: 执行 FetchSubPhases<br/>(highlight, fields, etc)
        end

        FetchPhase-->>DataNode: FetchSearchResult<br/>(SearchHit[])
        DataNode-->>TransportSvc: FetchSearchResult
        TransportSvc-->>AsyncAction: FetchSearchResult
    end

    Note over AsyncAction,REST: 构建最终响应
    AsyncAction->>Controller: merge(reducedQueryPhase, fetchResults)
    Controller->>Controller: 合并文档内容和聚合结果
    Controller-->>AsyncAction: SearchResponseSections
    AsyncAction->>AsyncAction: 构建 SearchResponse
    AsyncAction-->>Transport: SearchResponse
    Transport-->>REST: SearchResponse
    REST->>REST: 格式化为 JSON
    REST-->>Client: HTTP 200 OK<br/>{hits, aggregations, took}

关键步骤:
1. 解析 URL 参数: indices, routing, preference, scroll
2. 解析请求体: withContentOrSourceParamParserOrNull()
3. 调用 parseSearchRequest() 解析 Query DSL
4. 构建 SearchRequest 对象
5. 执行: client.execute(TransportSearchAction.TYPE, searchRequest, listener)

RestSearchAction.prepareRequest()
  → parseSearchRequest(searchRequest, request, parser)
    → searchRequest.source().parseXContent(parser) // 解析Query DSL
  → RestCancellableNodeClient.execute(TransportSearchAction.TYPE, searchRequest)
    → TransportSearchAction.doExecute()

关键步骤:
1. 解析目标索引: indexNameExpressionResolver.concreteIndices()
2. 计算分片路由: operationRouting.searchShards()
3. 构建别名过滤器: buildPerIndexAliasFilter()
4. 创建SearchShardIterator[]: buildShardIterators()
5. 决定搜索类型并创建AsyncAction:
   - QUERY_THEN_FETCH → SearchQueryThenFetchAsyncAction
   - DFS_QUERY_THEN_FETCH → SearchDfsQueryThenFetchAsyncAction
6. 启动异步执行: searchAsyncAction.start()

operationRouting.searchShards():
- 根据 routing 参数选择分片
- 根据 preference 选择主分片或副本分片
  - _primary: 总是使用主分片
  - _replica: 总是使用副本分片
  - _local: 优先使用本地分片
  - _only_local: 仅使用本地分片
- 构建 SearchShardIterator: 包含主分片和所有副本的迭代器

关键步骤:
1. 检查是否需要 CanMatch 预过滤:
   - 条件: preFilterSearchShards && 分片数 > threshold
   - 执行 CanMatchPreFilterSearchPhase
2. 如果不需要预过滤,直接执行 Query Phase:
   - 遍历所有 SearchShardIterator
   - 对每个分片调用 executePhaseOnShard()
3. 并发控制:
   - 使用 pendingExecutionsPerNode 限制每节点并发
   - 默认 maxConcurrentShardRequests = 5

// 每个节点维护一个待执行队列和 Semaphore
class PendingExecutions {
    private final Semaphore permits; // 最多maxConcurrent个许可
    private final Queue<Runnable> queue; // 待执行队列

    void tryExecute(Runnable task) {
        if (permits.tryAcquire()) {
            task.run(); // 立即执行
        } else {
            queue.add(task); // 加入队列
        }
    }

    void finishOperation() {
        permits.release();
        Runnable next = queue.poll();
        if (next != null) next.run(); // 执行下一个
    }
}

executePhaseOnShard():
1. 构建 ShardSearchRequest
2. 通过 SearchTransportService 发送 RPC 请求
3. 注册异步回调监听器 SearchActionListener

关键步骤:
1. 获取或创建 ReaderContext:
   - 如果是首次查询: 创建新的 ReaderContext
   - 如果是 Scroll/PIT: 复用现有 ReaderContext
2. 重写查询: rewriteAndFetchShardRequest()
   - 在分片上下文重写 Query
   - 优化查询结构
3. 提交到 search 线程池:
   - 使用 runAsync(executor, task)
   - executor = threadPool.executor(Names.SEARCH)
4. 在线程池中执行:
   - 创建 SearchContext
   - 调用 QueryPhase.execute(searchContext)
   - 返回 QuerySearchResult

createContext():
1. 创建 DefaultSearchContext
2. 解析 SearchSource:
   - parseSource(context, request.source())
   - 解析 query, aggregations, sort, highlight
3. 设置默认值 from=0, size=10
4. 执行 preProcess():
   - 构建 Lucene Query
   - 初始化聚合上下文

关键步骤:
1. 预处理聚合: AggregationPhase.preProcess()
2. 构建 CollectorManager:
   - TopDocsCollector: 收集 Top K 文档
   - AggregationCollector: 执行聚合计算
   - 使用 MultiCollector 组合多个 Collector
3. 执行搜索: searcher.search(query, collectorManager)
4. 后处理:
   - RescorePhase.execute(): 重新打分
   - SuggestPhase.execute(): 搜索建议
5. 设置结果: queryResult.topDocs()

QueryPhaseCollectorManager.newCollector():
1. 如果启用 profile: 包装 InternalProfileCollector
2. 创建 TopDocsCollector:
   - 如果有排序: TopFieldCollector
   - 否则: TopScoreDocCollector
3. 创建 AggregationCollector (如果有聚合)
4. 组合成 QueryPhaseCollector

关键步骤:
1. 收集所有非空的 QuerySearchResult
2. 归并排序:
   - 使用 sortDocs() 进行 K 路归并
   - 创建 ScoreDoc[] 优先队列
   - 按分数/排序字段排序
3. 归并聚合:
   - InternalAggregations.reduce()
   - 对每种聚合类型执行归并逻辑
4. 构建 ReducedQueryPhase:
   - 包含全局 Top K 的 (shardIndex, docId)
   - 归并后的聚合结果

算法流程:
1. 创建 PriorityQueue<ScoreDoc>
2. 从每个分片的 TopDocs 取第一个文档放入队列
3. 循环:
   - 从队列取出最高分/最优排序值的文档
   - 从该分片取下一个文档放入队列
   - 重复直到收集够 size 个文档
4. 时间复杂度: O(N * K * log K)
   - N = 分片数
   - K = size + from

关键步骤:
1. 如果 docIds 为空: 返回空结果
2. 调用 buildSearchHits(context, docIds):
   - 遍历所有 docId
   - 对每个 docId:
     a. 读取 StoredFields: searcher.doc(docId)
     b. 提取 _source 字段
     c. 执行 FetchSubPhases
3. FetchSubPhases 包括:
   - FetchSourcePhase: 过滤 _source
   - HighlightPhase: 生成高亮片段
   - FetchFieldsPhase: 提取指定字段
   - FetchDocValuesPhase: 提取 doc_values

getProcessors():
1. 遍历所有注册的 FetchSubPhase
2. 调用 fsp.getProcessor(context)
3. 如果返回非空 processor:
   - 包装成 ProfileProcessor (如果启用 profile)
   - 添加到 processors 列表
4. 返回 processors 列表

1. FetchSourcePhase: 提取和过滤 _source
2. FetchFieldsPhase: 提取 fields
3. FetchDocValuesPhase: 提取 docvalue_fields
4. HighlightPhase: 生成高亮
5. ExplainPhase: 生成评分解释 (如果 explain=true)
6. InnerHitsPhase: 处理嵌套/父子文档

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Coord as 协调节点<br/>(TransportSearchAction)
    participant CanMatch as CanMatch Phase
    participant Shard1 as Shard 1
    participant Shard2 as Shard 2
    participant ShardN as Shard N
    participant Reducer as SearchPhaseController<br/>(结果归并)

    Client->>Coord: POST /index/_search<br/>{query, size, from}

    Coord->>Coord: 1. 解析请求<br/>构建 SearchRequest
    Coord->>Coord: 2. 计算目标分片<br/>(所有主分片或副本)

    Note over Coord,ShardN: CanMatch Phase - 预过滤

    Coord->>CanMatch: 3. 启动 CanMatch Phase

    par 并发检查所有分片
        CanMatch->>Shard1: CanMatchRequest<br/>(min/max range check)
        Shard1->>Shard1: 检查分片范围<br/>(MinMax统计)
        Shard1-->>CanMatch: canMatch=true
    and
        CanMatch->>Shard2: CanMatchRequest
        Shard2->>Shard2: 检查分片范围
        Shard2-->>CanMatch: canMatch=false<br/>(跳过此分片)
    and
        CanMatch->>ShardN: CanMatchRequest
        ShardN->>ShardN: 检查分片范围
        ShardN-->>CanMatch: canMatch=true
    end

    CanMatch->>Coord: 过滤后的分片列表<br/>[Shard1, ShardN]

    Note over Coord,ShardN: Query Phase - 查询阶段

    Coord->>Coord: 4. 启动 Query Phase

    par 并发查询所有匹配的分片
        Coord->>Shard1: ShardSearchRequest<br/>(query, size+from, aggs)

        Shard1->>Shard1: 5. 创建 SearchContext
        Shard1->>Shard1: 6. 构建 Lucene Query
        Shard1->>Shard1: 7. 执行搜索<br/>searcher.search(query, collector)
        Shard1->>Shard1: 8. 收集 Top (size+from) 文档
        Shard1->>Shard1: 9. 执行聚合<br/>(如果有)

        Shard1-->>Coord: QuerySearchResult<br/>(docIds[], scores[], aggs)
    and
        Coord->>ShardN: ShardSearchRequest

        ShardN->>ShardN: 创建 SearchContext
        ShardN->>ShardN: 构建 Query
        ShardN->>ShardN: 执行搜索
        ShardN->>ShardN: 收集 Top docs
        ShardN->>ShardN: 执行聚合

        ShardN-->>Coord: QuerySearchResult<br/>(docIds[], scores[], aggs)
    end

    Note over Coord,Reducer: Reduce Phase - 归并阶段

    Coord->>Reducer: 10. 归并所有分片结果

    Reducer->>Reducer: 11. 归并排序<br/>所有 QueryResult
    Reducer->>Reducer: 12. 选择全局 Top (size)<br/>跳过 from
    Reducer->>Reducer: 13. 归并聚合结果<br/>(如果有)

    Reducer-->>Coord: ReducedQueryPhase<br/>(Top K docIds by shard)

    Note over Coord,ShardN: Fetch Phase - 获取阶段

    Coord->>Coord: 14. 构建 Fetch 请求<br/>按分片分组 docIds

    par 并发获取文档内容
        Coord->>Shard1: ShardFetchRequest<br/>(docIds: [3, 7, 15])

        Shard1->>Shard1: 15. 读取文档内容<br/>searcher.doc(docId)
        Shard1->>Shard1: 16. 生成高亮<br/>(如果需要)
        Shard1->>Shard1: 17. 提取 _source

        Shard1-->>Coord: FetchSearchResult<br/>(documents[])
    and
        Coord->>ShardN: ShardFetchRequest<br/>(docIds: [2, 9])

        ShardN->>ShardN: 读取文档
        ShardN->>ShardN: 生成高亮
        ShardN->>ShardN: 提取 _source

        ShardN-->>Coord: FetchSearchResult<br/>(documents[])
    end

    Coord->>Coord: 18. 合并 Fetch 结果<br/>按分数顺序排列

    Coord-->>Client: SearchResponse<br/>{hits[], aggregations, took, total}

sequenceDiagram
    autonumber
    participant Coord as 协调节点<br/>CanMatchPreFilterPhase
    participant Rewrite as CoordinatorRewriteContext<br/>查询重写
    participant Transport as SearchTransportService
    participant Shard as 数据节点分片
    participant MinMax as PointValues<br/>MinMax统计

    Note over Coord,MinMax: 阶段0: 协调节点重写

    Coord->>Coord: runCoordinatorRewritePhase()

    loop 每个分片
        Coord->>Rewrite: queryStillMatchesAfterRewrite()
        Rewrite->>Rewrite: 重写查询<br/>(范围查询优化)
        Rewrite->>Rewrite: 检查查询是否可能匹配<br/>(基于索引元数据)
        Rewrite-->>Coord: canMatch (true/false)

        alt canMatch=false (协调节点就能判断)
            Coord->>Coord: 标记分片skip=true
            Coord->>Coord: numPossibleMatches不增加
        else canMatch=true
            Coord->>Coord: 加入matchedShardLevelRequests
        end
    end

    Note over Coord,MinMax: 阶段1: 向数据节点发送CanMatch请求

    Coord->>Coord: 按节点分组请求<br/>groupByNode()

    loop 每个目标节点
        Coord->>Transport: sendCanMatch(nodeId, shards[])
        Transport->>Shard: [RPC] CanMatchNodeRequest

        loop 该节点的每个分片
            Shard->>Shard: 获取 IndexReader
            Shard->>MinMax: 获取字段的min/max值<br/>PointValues.getMinPackedValue()<br/>PointValues.getMaxPackedValue()
            MinMax-->>Shard: min/max BytesRef

            Shard->>Shard: 检查查询范围与分片范围是否重叠
            Note over Shard: 示例: 查询 date >= 2024-01-01<br/>分片 max(date) = 2023-12-31<br/>→ canMatch = false

            Shard->>Shard: 如果有排序,同时返回minSortValue/maxSortValue
            Shard-->>Transport: CanMatchShardResponse<br/>(canMatch, minMax, estimatedCount)
        end

        Transport-->>Coord: CanMatchNodeResponse[]
    end

    Note over Coord: 阶段2: 处理响应并排序分片

    loop 每个分片响应
        alt canMatch=true
            Coord->>Coord: possibleMatches.set(shardIndex)
            Coord->>Coord: numPossibleMatches++
            Coord->>Coord: 记录minAndMaxes[shardIndex]
        else canMatch=false
            Coord->>Coord: 跳过该分片
        end
    end

    Coord->>Coord: 如果有排序,按minSortValue排序分片<br/>(确保包含早期数据的分片先查询)

    Coord->>Coord: 构建过滤后的 SearchShardIterator[]
    Coord-->>Coord: 返回可能匹配的分片列表<br/>numPossibleMatches / totalShards

    Note over Coord: 优化效果示例:<br/>原始128个分片 → 过滤后仅64个分片<br/>减少50%的Query Phase工作量

sequenceDiagram
    autonumber
    participant Coord as 协调节点<br/>AsyncAction
    participant Transport as SearchTransportService
    participant Service as SearchService<br/>(数据节点)
    participant Context as DefaultSearchContext
    participant QueryPhase as QueryPhase
    participant Collector as CollectorManager
    participant Lucene as Lucene IndexSearcher
    participant Aggs as AggregationPhase

    Note over Coord,Lucene: 协调节点发起Query请求

    Coord->>Coord: executePhaseOnShard(shardIt)
    Coord->>Coord: buildShardSearchRequest(shardIt)<br/>构造ShardSearchRequest
    Coord->>Coord: 应用并发控制<br/>(每节点最多5个并发)
    Coord->>Transport: sendExecuteQuery(connection, request)
    Transport->>Service: [RPC] ShardSearchRequest

    Note over Service,Context: 数据节点接收并处理

    Service->>Service: executeQueryPhase(request, task, listener)
    Service->>Service: 获取/创建 ReaderContext<br/>(持有IndexReader引用)
    Service->>Service: rewriteAndFetchShardRequest()<br/>重写查询(分片级优化)
    Service->>Service: 提交到search线程池<br/>executor.execute()

    Note over Service,Context: search线程池中执行

    Service->>Context: createContext(readerContext, request)
    Context->>Context: new DefaultSearchContext()
    Context->>Context: parseSource(request.source())<br/>解析查询/聚合/排序

    Context->>Context: 构建Lucene Query对象
    Note over Context: QueryBuilder → Query转换:<br/>MatchQueryBuilder → TermQuery<br/>RangeQueryBuilder → PointRangeQuery<br/>BoolQueryBuilder → BooleanQuery

    Context->>Context: 构建Sort对象<br/>(如果有排序)
    Context->>Context: 解析Aggregations<br/>构建AggregatorFactories
    Context->>Context: preProcess()<br/>查询预处理和验证

    Service->>QueryPhase: QueryPhase.execute(searchContext)

    Note over QueryPhase,Aggs: Query Phase 执行

    QueryPhase->>Aggs: AggregationPhase.preProcess(searchContext)
    Aggs->>Aggs: 为每个聚合创建Aggregator
    Aggs->>Aggs: 构建聚合树(嵌套聚合)
    Aggs-->>QueryPhase: AggregationContext

    QueryPhase->>Collector: createQueryPhaseCollectorManager()

    alt 有排序
        Collector->>Collector: new TopFieldCollectorManager<br/>(排序字段, size+from)
    else 无排序
        Collector->>Collector: new TopScoreDocCollectorManager<br/>(size+from)
    end

    alt 有聚合
        Collector->>Collector: aggsCollectorManager.newCollector()
        Collector->>Collector: MultiCollector.wrap(<br/>  topDocsCollector,<br/>  aggsCollector<br/>)
    end

    alt 有postFilter
        Collector->>Collector: 创建FilteredCollector<br/>(在聚合后过滤)
    end

    Collector-->>QueryPhase: QueryPhaseCollectorManager

    QueryPhase->>Lucene: searcher.search(query, collectorManager)

    Note over Lucene: Lucene 搜索执行

    Lucene->>Lucene: 1. Weight weight = query.createWeight()
    Lucene->>Lucene: 2. 遍历所有Segment

    loop 每个Segment
        Lucene->>Lucene: LeafReaderContext leafCtx = segment
        Lucene->>Lucene: Scorer scorer = weight.scorer(leafCtx)
        Lucene->>Lucene: Collector collector = collectorManager.newCollector()
        Lucene->>Lucene: collector.setScorer(scorer)

        loop 匹配的文档
            Lucene->>Lucene: int docId = scorer.nextDoc()
            Lucene->>Lucene: float score = scorer.score()
            Lucene->>Lucene: collector.collect(docId)

            alt TopDocsCollector
                Lucene->>Lucene: 如果分数足够高<br/>加入PriorityQueue
            end

            alt AggsCollector
                Lucene->>Lucene: 执行聚合计算<br/>aggregator.collect(docId)
            end
        end

        Lucene->>Lucene: collector.finishLeaf()
    end

    Lucene->>Lucene: collectorManager.reduce(collectors)<br/>合并所有Segment的结果
    Lucene-->>QueryPhase: QueryPhaseResult<br/>(TopDocs, totalHits)

    Note over QueryPhase: 后处理阶段

    QueryPhase->>QueryPhase: 设置queryResult.topDocs(topDocs)
    QueryPhase->>QueryPhase: 设置queryResult.aggregations(aggs)

    alt 有Rescore
        QueryPhase->>QueryPhase: RescorePhase.execute(searchContext)<br/>重新打分Top K文档
    end

    alt 有Suggest
        QueryPhase->>QueryPhase: SuggestPhase.execute(searchContext)<br/>生成搜索建议
    end

    QueryPhase-->>Service: QuerySearchResult

    Note over Service,Transport: 返回结果给协调节点

    Service->>Service: 检查是否单次查询<br/>(non-scroll)

    alt 单次查询且无结果
        Service->>Service: freeReaderContext()<br/>立即释放资源
    else Scroll/PIT
        Service->>Service: 保持ReaderContext<br/>等待后续请求
    end

    Service-->>Transport: QuerySearchResult<br/>(docIds[], scores[], aggs)
    Transport-->>Coord: QuerySearchResult

    Note over Coord: 收集所有分片结果<br/>准备进入Reduce Phase

// 重写示例
原始查询: MatchQuery("title", "elasticsearch")
  ↓ 重写1: 分析器处理
TermQuery("title", "elasticsearch")
  ↓ 重写2: 前缀优化
如果title字段只有一个Term匹配,可能重写为ConstantScoreQuery
  ↓ 重写3: 分片级优化
如果该分片该Term的docFreq=0,重写为MatchNoDocsQuery

// 聚合执行伪代码
for (Document doc : matchedDocs) {
    topDocsCollector.collect(doc); // 收集到TopDocs
    aggsCollector.collect(doc);     // 同时执行聚合

    // 对于terms聚合
    String value = doc.getField("category");
    termsBucket.get(value).count++;

    // 对于metrics聚合
    double price = doc.getField("price");
    sumAggregator.sum += price;
    avgAggregator.sum += price;
    avgAggregator.count++;
}

sequenceDiagram
    autonumber
    participant AsyncAction as SearchQueryThenFetch<br/>AsyncAction
    participant Consumer as QueryPhaseResult<br/>Consumer
    participant Controller as SearchPhaseController
    participant MergeQueue as PriorityQueue<br/>归并队列
    participant AggsReduce as InternalAggregations<br/>聚合归并

    Note over AsyncAction,AggsReduce: 收集所有分片结果

    loop 每个分片
        AsyncAction->>Consumer: onShardResult(QuerySearchResult)
        Consumer->>Consumer: 添加到results数组<br/>results.set(shardIndex, result)
        Consumer->>Consumer: successfulOps++

        alt 启用批量归并 && 达到batchSize
            Consumer->>Consumer: 执行批量部分归并<br/>减少内存占用
        end
    end

    AsyncAction->>AsyncAction: 等待所有分片完成<br/>countDown.await()
    AsyncAction->>AsyncAction: getNextPhase() → 触发Reduce

    Note over Controller,MergeQueue: Reduce Phase 开始

    AsyncAction->>Controller: reducedQueryPhase(queryResults)

    Controller->>Controller: 1. 收集所有非空 QuerySearchResult
    Controller->>Controller: 2. 验证排序字段格式一致性

    Note over Controller: 归并TopDocs

    Controller->>Controller: sortDocs(results[], from, size)
    Controller->>MergeQueue: 创建 ScoreDocQueue<br/>优先队列(容量=size)

    loop 每个分片的TopDocs
        MergeQueue->>MergeQueue: 取第一个ScoreDoc加入队列<br/>queue.insertWithOverflow(scoreDoc)
    end

    loop size 次
        MergeQueue->>MergeQueue: 从队列取出最高分文档<br/>topDoc = queue.pop()
        MergeQueue->>MergeQueue: 记录 (shardIndex, docId, score)

        alt 该分片还有更多文档
            MergeQueue->>MergeQueue: 从该分片取下一个ScoreDoc<br/>加入队列
        end
    end

    MergeQueue-->>Controller: 全局TopDocs[]<br/>已按分数排序

    Controller->>Controller: 跳过前 from 个文档
    Controller->>Controller: 选出 size 个文档

    Note over Controller,AggsReduce: 归并聚合结果

    alt 有聚合
        Controller->>AggsReduce: reduce(List<InternalAggregations>)

        loop 每个聚合
            alt Terms聚合
                AggsReduce->>AggsReduce: 1. 合并所有分片的buckets
                AggsReduce->>AggsReduce: 2. 按count降序排序
                AggsReduce->>AggsReduce: 3. 选出TopN buckets
                AggsReduce->>AggsReduce: 4. 递归归并子聚合
            else Metrics聚合 (sum/avg/max/min)
                AggsReduce->>AggsReduce: 直接合并指标值
                Note over AggsReduce: sum: 累加所有分片的sum<br/>avg: sum(所有sum) / sum(所有count)<br/>max: max(所有max)<br/>min: min(所有min)
            else Cardinality聚合
                AggsReduce->>AggsReduce: 合并HyperLogLog草图
            else Percentiles聚合
                AggsReduce->>AggsReduce: 合并TDigest草图
            end
        end

        AggsReduce-->>Controller: 归并后的InternalAggregations
    end

    Note over Controller: 归并Suggest结果

    alt 有Suggest
        Controller->>Controller: Suggest.reduce(groupedSuggestions)

        loop 每个Suggestion
            Controller->>Controller: 合并所有分片的options
            Controller->>Controller: 按score排序
            Controller->>Controller: 选出TopN options
        end
    end

    Note over Controller: 构建ReducedQueryPhase

    Controller->>Controller: new ReducedQueryPhase(<br/>  topDocs: 全局Top K的(shard,docId),<br/>  aggregations: 归并后的聚合,<br/>  totalHits: 总命中数,<br/>  maxScore: 最高分,<br/>  suggest: 归并后的建议,<br/>  numReducePhases: 归并阶段数<br/>)

    Controller-->>AsyncAction: ReducedQueryPhase

    AsyncAction->>AsyncAction: 构建docIdsToLoad[]<br/>按分片分组TopDocs

    Note over AsyncAction: 准备进入Fetch Phase

// 批量归并参数
int batchedReduceSize = 512; // 默认每批512个分片

// 归并策略
if (numShards <= batchedReduceSize) {
    // 少量分片: 一次性归并
    reduce(allResults);
} else {
    // 大量分片: 分批归并
    while (results.size() > batchedReduceSize) {
        List<QuerySearchResult> batch = results.take(batchedReduceSize);
        ReducedQueryPhase partial = reduce(batch);
        results.add(partial); // 部分结果作为新的"分片"
    }
    reduce(results); // 最终归并
}

// 归并排序伪代码
PriorityQueue<ScoreDoc> queue = new PriorityQueue<>(
    size,
    (doc1, doc2) -> {
        // 比较规则
        if (doc1.score != doc2.score) {
            return Float.compare(doc2.score, doc1.score); // 分数降序
        }
        // 分数相同,按shardIndex排序(保证稳定性)
        return Integer.compare(doc1.shardIndex, doc2.shardIndex);
    }
);

// 初始化: 每个分片的第一个文档加入队列
for (int i = 0; i < numShards; i++) {
    TopDocs shardTopDocs = results[i].topDocs();
    if (shardTopDocs.scoreDocs.length > 0) {
        ScoreDoc doc = shardTopDocs.scoreDocs[0];
        doc.shardIndex = i;
        doc.docIndex = 0;
        queue.add(doc);
    }
}

// 归并: 取size个最高分文档
ScoreDoc[] topDocs = new ScoreDoc[size];
for (int i = 0; i < size && queue.isNotEmpty(); i++) {
    ScoreDoc top = queue.poll(); // 取出当前最高分
    topDocs[i] = top;

    // 从同一分片取下一个文档
    TopDocs shardTopDocs = results[top.shardIndex].topDocs();
    int nextDocIndex = top.docIndex + 1;
    if (nextDocIndex < shardTopDocs.scoreDocs.length) {
        ScoreDoc nextDoc = shardTopDocs.scoreDocs[nextDocIndex];
        nextDoc.shardIndex = top.shardIndex;
        nextDoc.docIndex = nextDocIndex;
        queue.add(nextDoc);
    }
}