Level 7: Search Path and Aggregations

This level is where you learn how OpenSearch actually answers a _search. Up to now you have read the write path, cluster state, allocation, and recovery. Now you turn to the read path: the two-phase, scatter-gather protocol that fans a query out to many shards on many nodes and reduces the partial answers into one result. You will trace a search through QueryPhase and FetchPhase, understand how aggregations collect per shard and reduce on the coordinating node, and finally build and register a custom aggregation in a SearchPlugin.

A contributor who does not understand query-then-fetch cannot reason about pagination cost, partial failures, aggregation approximation error, or why a sort needs DocValues. This level closes that gap and ends with you shipping real Java into the search subsystem.

Learning Objectives

By the end of Level 7 you must be able to:

1. Diagram the query-then-fetch fan-out: what the coordinating node sends, what each shard returns, and what the second round trip fetches.
2. Locate the per-shard phase entry points (SearchService.executeQueryPhase, executeFetchPhase) and the coordinating-node reduce (SearchPhaseController) in source.
3. Explain why deep from/size pagination is expensive and what search_after, scroll, and point-in-time (PIT) do about it.
4. Explain the DfsPhase and when it changes scoring; explain the can_match pre-filter.
5. Walk the aggregation lifecycle: AggregatorFactory → Aggregator (per-shard collect + buildAggregation) → InternalAggregation.reduce(...) on the coordinator.
6. Explain shard_size vs size for a terms aggregation and where approximation error comes from.
7. Trace a QueryBuilder to a Lucene Query through QueryShardContext, and explain why sort and aggs read columnar DocValues.
8. Register a custom aggregation (or query) via SearchPlugin, build it, install it, and exercise it through _search.

The Two-Phase Query-Then-Fetch Model

A _search is not "run the query on a shard." A search index is split into shards, each shard is an independent Lucene index, and no single shard has the global answer. So the coordinating node (the node that received the REST request) runs a scatter-gather:

flowchart TD
    C[Client] -->|HTTP GET _search| R[RestSearchAction]
    R --> T[TransportSearchAction<br/>coordinating node]
    T -->|can_match pre-filter<br/>optional| CM[per-shard can_match]
    T -->|QUERY: fan out| S1[Shard 0<br/>SearchService.executeQueryPhase]
    T -->|QUERY: fan out| S2[Shard 1<br/>QueryPhase]
    T -->|QUERY: fan out| S3[Shard N<br/>QueryPhase]
    S1 -->|top docIds + scores + agg slices| RED[SearchPhaseController<br/>reduce on coordinator]
    S2 --> RED
    S3 --> RED
    RED -->|which docs to fetch| F1[Shard 0<br/>executeFetchPhase]
    RED --> F2[Shard 1<br/>FetchPhase]
    F1 -->|_source + fields| MERGE[merge hits + aggs]
    F2 --> MERGE
    MERGE -->|SearchResponse| C

Phase 1 — Query. The coordinating node sends the parsed request to every target shard. Each shard runs the query locally (QueryPhase), computes its top from + size doc IDs plus their scores (and any aggregation partial results), and returns only IDs, scores, and sort values — not the documents themselves. The coordinator collects these slices in SearchPhaseController and computes the global top size across all shards. This is the reduce.

Phase 2 — Fetch. Now the coordinator knows exactly which doc IDs (on which shards) belong in the final page. It sends a second, much smaller fan-out — only to the shards that own winning docs — asking each to load the actual _source, highlighted fields, etc. (FetchPhase). The coordinator stitches the fetched documents into the final SearchResponse.

Why two phases? Because fetching _source for every candidate on every shard would move enormous amounts of data over the wire just to throw most of it away. Query-then-fetch moves only the documents that survive the global merge.

Note: This is why deep pagination is expensive. To serve from=10000&size=10, every shard must produce its top 10010 docs so the coordinator can find the global top 10010 and slice the last 10. Cost grows with from + size on every shard. Use search_after (cursor by sort values) or point-in-time (PIT) instead — see Search Execution.

Who Does What — Coordinating Node vs Data Node

A single node is often both coordinator and data node for a request — but the roles are distinct, and keeping them straight is the key to reading the code.

DFS, can_match, and Scoring

• DfsPhase (DFS = Distributed Frequency Search). BM25 scoring depends on term statistics (document frequency, total term frequency). Each shard only knows its local statistics, so the same term can score differently on different shards. With search_type=dfs_query_then_fetch, an extra round trip (DfsPhase) gathers global term statistics first, so scoring is consistent across shards. The default query_then_fetch skips it (cheaper, slightly less precise scores). Find it:

  grep -rn "class DfsPhase" server/src/main/java/org/opensearch/search/dfs/
  

• can_match pre-filter. Before the query phase, the coordinator can ask each shard a cheap "could this shard possibly contain a matching doc?" question (using min/max ranges in the shard's segment metadata). Shards that cannot match are skipped entirely. This is why a time-range query over hundreds of time-based indices stays fast — most shards are pruned.

  grep -rn "canMatch" server/src/main/java/org/opensearch/search/SearchService.java
  

Aggregations in One Paragraph

Aggregations ride the query phase. As the query collects matching docs on a shard, each Aggregator also collects into buckets/metrics. At the end of the shard's query phase, each aggregator emits an InternalAggregation (its partial result). The coordinator then calls InternalAggregation.reduce(...) to combine all shards' partials into the final aggregation.

flowchart LR
    AB[AggregationBuilder<br/>from JSON] --> AF[AggregatorFactory<br/>per shard]
    AF --> AG[Aggregator<br/>collect per doc]
    AG -->|buildAggregation| IA[InternalAggregation<br/>shard partial]
    IA -->|wire| RED[InternalAggregation.reduce<br/>coordinator]
    RED --> FINAL[final aggregation in SearchResponse]

The crucial subtlety: aggregations can be approximate by design. A terms aggregation asking for the top 10 terms cannot be exact unless every shard returns every term — instead each shard returns its top shard_size terms, and the coordinator merges. A term that is #11 on two shards but globally #5 can be missed. That trade-off, and the doc_count_error_upper_bound field that quantifies it, are covered in Lab 7.2 and the Aggregations deep dive.

QueryBuilders → Lucene Queries, and DocValues

The JSON query DSL is parsed into a tree of QueryBuilder objects (AbstractQueryBuilder subclasses: MatchQueryBuilder, BoolQueryBuilder, RangeQueryBuilder, …). Each builder is Writeable (it crosses the wire to data nodes) and ToXContent (it round-trips to JSON). On the shard, QueryBuilder.toQuery(QueryShardContext) turns the builder into an actual Lucene Query that Lucene's IndexSearcher can execute. See Query DSL and QueryBuilders.

Scoring and matching read the inverted index. But sorting, aggregations, and script field access read columnar DocValues — a per-field, per-document column store that is fast to iterate by document ID. You cannot sort or aggregate on a field with doc_values: false. This is the single most common "why is my aggregation failing?" root cause. See DocValues and Fielddata.

Key Classes Quick Reference

Exact package paths and class names vary slightly by branch. Confirm with grep, e.g. grep -rn "class SearchPhaseController" server/src/main/java/.

Source Areas to Read

# The coordinating-node search action and reduce
ls server/src/main/java/org/opensearch/action/search/

# Per-shard search service and phases
ls server/src/main/java/org/opensearch/search/
ls server/src/main/java/org/opensearch/search/query/
ls server/src/main/java/org/opensearch/search/fetch/
ls server/src/main/java/org/opensearch/search/dfs/

# The aggregation framework
ls server/src/main/java/org/opensearch/search/aggregations/
ls server/src/main/java/org/opensearch/search/aggregations/bucket/terms/
ls server/src/main/java/org/opensearch/search/aggregations/metrics/

# Query DSL builders
ls server/src/main/java/org/opensearch/index/query/

Labs

Deliverables

You must demonstrate all of the following before advancing to Level 8:

• A completed reading log tracing RestSearchAction → TransportSearchAction → SearchService.executeQueryPhase/executeFetchPhase → SearchPhaseController, with the grep you used at each hop (Lab 7.1).
• A Profile API (_search?profile=true) output correlated to the phases you traced.
• A written explanation of shard_size vs size and where doc_count_error_upper_bound comes from (Lab 7.2).
• A custom aggregation registered via SearchPlugin, built, installed, and exercised via _search, with a passing AggregatorTestCase (Lab 7.3).
• A two-sentence explanation of why sort/aggregations require DocValues.

Common Mistakes

Where This Level Points Next

• The deep dives this level depends on: Search Execution, Aggregations, Query DSL and QueryBuilders, DocValues and Fielddata.
• Level 8 takes you from curated labs to real GitHub issues.
• The issue roadmap stage on search and aggregations lists the kinds of search/agg issues you are now equipped to take.