Search Execution

A _search request is not "run the query on a shard." It is a scatter-gather across many shards on many nodes, a two-phase (query-then-fetch) protocol, and a reduce on the coordinating node that stitches partial results into a final answer. Understanding this fan-out — who holds what state, what crosses the wire, and where things can go wrong — is the difference between debugging a slow search and guessing.

This chapter follows a search from TransportSearchAction on the coordinating node down to per-shard SearchService phases and back up through SearchPhaseController. It assumes you know the request path (The Action Framework, The Transport Layer) and feeds the Query DSL and Aggregations deep dives, which detail what runs inside a phase.

After this chapter you can:

• Diagram the query-then-fetch fan-out and say what each round trip carries.
• Explain why deep from/size pagination is expensive and what search_after, scroll, and point-in-time (PIT) fix.
• Locate the coordinating-node reduce and the per-shard phase entry points in source.
• Reason about can_match pre-filtering, shard routing, and partial failures.

The cast

grep -n "executeQueryPhase\|executeFetchPhase\|executeDfsPhase\|public.*canMatch" \
  server/src/main/java/org/opensearch/search/SearchService.java
ls server/src/main/java/org/opensearch/action/search/ | grep -i "asyncaction\|phase"

Query-then-fetch: the two-phase model

OpenSearch does not ship full documents from every shard for every candidate hit. That would be wasteful: to return the top 10 of millions, you only need the document IDs and sort values from each shard first, pick the global top 10, then fetch the bodies of exactly those 10. That is query-then-fetch.

sequenceDiagram
    participant C as Client
    participant Co as Coordinating node (TransportSearchAction)
    participant S1 as Shard A (data node)
    participant S2 as Shard B (data node)
    C->>Co: POST /idx/_search {query, from, size, aggs}
    Note over Co: resolve indices, routing -> target shards
    par Query phase (scatter)
        Co->>S1: ShardSearchRequest (QueryPhase)
        Co->>S2: ShardSearchRequest (QueryPhase)
    end
    S1-->>Co: top docIds + sort values + agg slices + maxScore
    S2-->>Co: top docIds + sort values + agg slices + maxScore
    Note over Co: SearchPhaseController.reducedQueryPhase -> global top-K + which shards own them
    par Fetch phase (scatter, only to shards holding winners)
        Co->>S1: ShardFetchRequest (docIds to materialize)
        Co->>S2: ShardFetchRequest
    end
    S1-->>Co: _source + fields + highlights for its winners
    S2-->>Co: _source + fields ...
    Note over Co: merge hits, reduce aggs, build SearchResponse
    Co-->>C: SearchResponse (hits, aggregations, took, _shards)

Phase 1 — Query

On each target shard, SearchService.executeQueryPhase builds a DefaultSearchContext, runs QueryPhase.execute, and returns a QuerySearchResult: the top from + size document IDs for that shard, their sort/score values, maxScore, total hit count (subject to track_total_hits), and any aggregation partials. It does not contain _source.

grep -n "class QuerySearchResult\|topDocs\|aggregations\|totalHits" \
  server/src/main/java/org/opensearch/search/query/QuerySearchResult.java
grep -n "public.*execute" server/src/main/java/org/opensearch/search/query/QueryPhase.java

Reduce — pick the global winners

The coordinating node collects every QuerySearchResult and calls SearchPhaseController.reducedQueryPhase(...): it merges the per-shard top-K into a global top-K (a sorted merge over sort values), reduces aggregations (InternalAggregation.reduce, see Aggregations), and produces a ScoreDoc[] plus the map of which shard owns each winning doc.

grep -n "reducedQueryPhase\|mergeTopDocs\|reduceAggs\|sortDocs" \
  server/src/main/java/org/opensearch/action/search/SearchPhaseController.java

Phase 2 — Fetch

For each shard that owns at least one global winner, the coordinator sends a ShardFetchSearchRequest listing the exact doc IDs. SearchService.executeFetchPhase runs FetchPhase, which materializes _source, stored/doc-value fields, highlights, inner hits, etc., into a FetchSearchResult. The coordinator merges these into the final SearchHits.

grep -n "executeFetchPhase\|FetchPhase\|FetchSearchResult\|fetchSubPhase" \
  server/src/main/java/org/opensearch/search/fetch/FetchPhase.java \
  server/src/main/java/org/opensearch/search/SearchService.java

DFS — when local term stats lie

BM25 scoring (see Query DSL and QueryBuilders) depends on term frequencies and document frequencies. Each shard only knows its own docFreq. With few shards and uniform data this is fine; with skewed data or rare terms across many shards, per-shard scoring can rank inconsistently.

The optional DFS phase fixes this. With ?search_type=dfs_query_then_fetch, the coordinator first runs DfsPhase on each shard to collect global term statistics, aggregates them, and feeds the global stats into the query phase so every shard scores against the same numbers.

flowchart LR
    A[DfsPhase per shard: collect docFreq/termStats] --> B[Coordinator aggregates global term stats]
    B --> C[QueryPhase per shard scores with global stats]
    C --> D[reduce -> FetchPhase]

grep -n "DfsPhase\|AggregatedDfs\|dfs_query_then_fetch\|SearchType.DFS" \
  server/src/main/java/org/opensearch/search/dfs/DfsPhase.java \
  server/src/main/java/org/opensearch/action/search/SearchType.java

Note: DFS adds a network round trip. Use it when scoring consistency matters (e.g., few docs, relevance-sensitive ranking). Default is query_then_fetch.

can_match: don't even ask a shard that can't match

CanMatchPreFilterSearchPhase runs a cheap pre-filter (SearchService.canMatch) against each shard before the real query phase. It uses min/max ranges from segment metadata (e.g., a @timestamp range filter against a time-based index) to skip shards that provably hold no matching documents. This is huge for time-series clusters with hundreds of shards where a query only touches one day.

It also drives field-sort shard ordering: shards are sorted so the most promising are queried first, enabling early termination.

grep -n "canMatch\|CanMatchPreFilter\|MinAndMax\|pre_filter_shard_size" \
  server/src/main/java/org/opensearch/search/SearchService.java \
  server/src/main/java/org/opensearch/action/search/CanMatchPreFilterSearchPhase.java

The pre-filter only kicks in when the number of target shards exceeds pre_filter_shard_size (or for certain frozen/searchable-snapshot indices).

Pagination: from/size vs search_after vs scroll vs PIT

grep -n "max_result_window\|from()\|size()\|searchAfter\|search_after" \
  server/src/main/java/org/opensearch/search/internal/DefaultSearchContext.java \
  server/src/main/java/org/opensearch/index/IndexSettings.java
grep -rn "point_in_time\|PitService\|CreatePit\|SearchContextId" \
  server/src/main/java/org/opensearch/action/search/ | head

Warning: Deep from/size (e.g., from=100000) forces every shard to build a 100k+ priority queue and ship it to the coordinator, which merges all of them. This is why max_result_window exists. The fix is almost always search_after over a PIT, not raising the window.

SearchContext — the per-shard state holder

SearchContext (concrete: DefaultSearchContext) is the per-shard, per-request state: the parsed query, the searcher (a reference into the latest refreshed reader — see Refresh, Flush, and Merge), aggregation collectors, from/size, sort, the QueryShardContext, timeouts, and the fetch sub-phases. For query-then-fetch it is typically created and torn down within a single phase; for scroll/PIT it is kept alive across requests (keyed by a SearchContextId).

grep -n "class DefaultSearchContext\|QueryShardContext\|ContextIndexSearcher\|aggregations(" \
  server/src/main/java/org/opensearch/search/internal/DefaultSearchContext.java

Leaking these (scroll contexts never cleared) is a real production incident: open contexts pin segments and prevent merge reclamation. Watch open_contexts in _nodes/stats/indices/search.

Partial failures and _shards

A search is "successful" even if some shards fail — the response's _shards block reports total, successful, skipped, and failed. The coordinator (AbstractSearchAsyncAction) tolerates per-shard failures up to the point that a phase can still produce a result; failed shards' contributions are simply absent. This is why you can get fewer results than expected without an HTTP error.

grep -n "successfulShards\|skippedShards\|shardFailures\|onShardFailure" \
  server/src/main/java/org/opensearch/action/search/AbstractSearchAsyncAction.java

Note: "Why is my count off?" is frequently a partial shard failure hiding in _shards.failed, not a query bug. Always read _shards before blaming the query.

Reading exercise

# 1. Coordinating-node entry and how it picks the async action (qtf vs dfs)
grep -n "executeSearch\|searchAsyncAction\|SearchQueryThenFetch\|SearchDfsQuery" \
  server/src/main/java/org/opensearch/action/search/TransportSearchAction.java

# 2. The reduce
grep -n "reducedQueryPhase\|reduce\|merge" \
  server/src/main/java/org/opensearch/action/search/SearchPhaseController.java

# 3. Per-shard phases
grep -n "execute" server/src/main/java/org/opensearch/search/query/QueryPhase.java
grep -n "execute\|hitsExecute" server/src/main/java/org/opensearch/search/fetch/FetchPhase.java

# 4. can_match
grep -n "canMatch\|MinAndMax\|FieldSortBuilder" \
  server/src/main/java/org/opensearch/search/SearchService.java

Answer:

1. In the query phase result (QuerySearchResult), is _source present? Where is _source actually loaded, and on which round trip?
2. After reducedQueryPhase, how does the coordinator know which shard to send each fetch request to? What data structure carries that mapping?
3. Trace how TransportSearchAction decides between query_then_fetch and dfs_query_then_fetch. What does DFS buy you, and what does it cost?
4. A user paginates with from=50000&size=20. Quantify (in words) the work each shard does and why search_after over a PIT is strictly cheaper.
5. On a 200-shard time-series index, a query filters @timestamp >= now-1h. Which class skips the irrelevant shards, what segment metadata does it use, and what setting gates whether it runs at all?
6. Find where a per-shard timeout or failure is recorded and explain how the final _shards.failed count is produced.

Common bugs and symptoms

Validation: prove you understand this

1. Draw the full query-then-fetch sequence for a 3-shard index returning the top 10 hits. Label exactly what data crosses on each of the (up to) four network legs and which carries _source.
2. Explain why query-then-fetch sends two rounds instead of one, in terms of bytes-on-the-wire and memory on the coordinator.
3. Given a relevance complaint on a 50-shard index with a rare search term, state the one-flag change you'd make and the new phase that runs, with its source file.
4. Contrast from/size, search_after, scroll, and PIT on three axes: statefulness, deep-page cost, and consistency. Name the class that holds the long-lived state for scroll/PIT.
5. Locate SearchPhaseController.reducedQueryPhase and describe, in two sentences, the merge it performs over per-shard top-docs and the agg reduce it triggers.
6. A user reports a search "missing" documents intermittently. Outline the 3-step diagnosis starting from _shards, then routing, then shard health.