Aggregations

Aggregations are OpenSearch's analytics engine: terms, date_histogram, avg, cardinality, nested sub-aggregations — the things that power dashboards. Like search, they run per shard and then reduce on the coordinating node. But unlike top-K search, aggregations can be approximate by design, and understanding why (and where the error comes from) is essential to using them correctly and to contributing a new aggregation.

This chapter follows the aggregation framework from AggregatorFactory through per-shard collection into InternalAggregation.reduce on the coordinator. It extends Search Execution (aggs ride the query phase) and depends hard on DocValues and Fielddata (aggs read columnar doc values, and global ordinals make terms fast).

After this chapter you can:

• Explain the factory → aggregator → internal-agg → reduce lifecycle.
• Distinguish bucket aggregators from metric aggregators and what each emits.
• Reason precisely about terms approximation error and the size / shard_size trade-off.
• Register a custom aggregation via SearchPlugin.getAggregations().

The four-stage lifecycle

ls server/src/main/java/org/opensearch/search/aggregations/
grep -n "class AggregatorFactories\|class AggregatorFactory\|abstract class Aggregator" \
  server/src/main/java/org/opensearch/search/aggregations/AggregatorFactories.java \
  server/src/main/java/org/opensearch/search/aggregations/AggregatorFactory.java \
  server/src/main/java/org/opensearch/search/aggregations/Aggregator.java

Bucket vs metric aggregators

A bucket aggregator owns child AggregatorFactories; metric aggregators are leaves. This nesting is what makes terms → avg ("average price per category") work: the terms aggregator creates one avg sub-aggregator instance per bucket and routes documents into the right one.

find server/src/main/java/org/opensearch/search/aggregations/bucket -name "TermsAggregator*.java"
find server/src/main/java/org/opensearch/search/aggregations/metrics -name "AvgAggregator*.java"

Collection: CollectorManager and BucketCollector

Aggregation collection plugs into Lucene's Collector machinery. Each Aggregator exposes a LeafBucketCollector (via getLeafCollector(LeafReaderContext)) that Lucene calls once per matching doc with collect(int doc, long bucketOrd). Bucket aggregators allocate bucket ordinals and recursively drive their sub-aggregators with the child bucket ordinal.

For concurrent (segment-parallel) search, aggregators are created through a CollectorManager/AggregationCollectorManager so that per-slice partials can be merged. The values themselves come from doc values — terms over a keyword field reads SortedSetDocValues and uses global ordinals to bucket by ordinal rather than by string, which is dramatically faster (see DocValues and Fielddata).

grep -rn "LeafBucketCollector\|getLeafCollector\|collect(int\|bucketOrd" \
  server/src/main/java/org/opensearch/search/aggregations/
grep -rn "CollectorManager\|AggregationCollectorManager\|globalOrdinals\|SortedSetDocValues" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/terms/

Reduce: where shards become one answer

Each shard returns a partial InternalAggregation. The coordinator collects them and calls InternalAggregation.reduce(List<InternalAggregation>, ReduceContext). For avg: sum the sums, sum the counts, divide once. For terms: merge bucket lists by key, summing doc counts and recursively reducing sub-aggregations, then trim to the requested size. For cardinality: merge HyperLogLog++ sketches.

flowchart TD
    subgraph Shard A
      A1[QueryPhase scan] --> A2[TermsAggregator collect by global ordinal]
      A2 --> A3[InternalTerms partial: top shard_size buckets]
    end
    subgraph Shard B
      B1[QueryPhase scan] --> B2[TermsAggregator collect]
      B2 --> B3[InternalTerms partial]
    end
    A3 --> R[Coordinator: InternalTerms.reduce]
    B3 --> R
    R --> F[Merge buckets by key, sum counts, reduce sub-aggs, trim to size]
    F --> O[Final aggregations in SearchResponse]

grep -rn "public InternalAggregation reduce\|doReduce\|ReduceContext" \
  server/src/main/java/org/opensearch/search/aggregations/InternalAggregation.java
grep -rn "reduce" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/terms/InternalTerms.java

Note: Reduce can happen in stages. For many shards the coordinator does a partial reduce in batches (batched_reduce_size) to bound memory, then a final reduce. A pipeline aggregation's logic runs in the final reduce.

The terms approximation: why counts can be wrong

This is the single most-misunderstood aggregation behavior, and a frequent "bug report" that is actually correct-by-design.

Each shard returns only its top shard_size terms (not all terms). The coordinator merges these. A term that is globally in the top size but ranks just below shard_size on every individual shard can be undercounted or missed entirely, because no shard ever reported it. OpenSearch surfaces this honestly:

• doc_count_error_upper_bound — the maximum a returned bucket's count could be off by.
• sum_other_doc_count — documents in buckets that didn't make the final list.

The mitigation is shard_size (defaults to roughly size * 1.5 + 10): each shard returns more candidates than the final size, shrinking the chance of a miss, at the cost of more memory/transfer.

grep -rn "shard_size\|shardSize\|doc_count_error\|sum_other_doc_count\|showTermDocCountError" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/terms/

Warning: Ordering a terms agg by a sub-aggregation metric (e.g., "top 5 categories by avg price") is inherently more approximate than ordering by doc_count, because the shard can't know which terms will win the metric ranking globally. Raise shard_size substantially or use composite/cardinality strategies for exactness-sensitive cases.

Sub-aggregations and the matrix-stats module

Sub-aggregations are just nested AggregatorFactories: a date_histogram with a child avg produces, per time bucket, the average of a field. The parent routes each doc to a bucket ordinal and calls the child's collect with that ordinal.

Not every aggregation lives in server/. The aggs-matrix-stats module (under modules/) is the canonical example of an aggregation shipped as a bundled module via the plugin SPI, not baked into core. It computes covariance/correlation across multiple fields (matrix_stats). Read it as the reference implementation for "how to add an aggregation through SearchPlugin."

find modules/aggs-matrix-stats -name "*.java" | grep -i "agg\|plugin" | head
grep -rn "getAggregations\|AggregationSpec\|MatrixStatsAggregationBuilder" \
  modules/aggs-matrix-stats/src/main/java/

Registering a custom aggregation

A plugin implements SearchPlugin.getAggregations() and returns AggregationSpecs. Each binds the agg name, the builder's stream reader and fromXContent parser, and registers the InternalAggregation's NamedWriteable reader so partial results can cross the wire and be reduced.

@Override
public List<AggregationSpec> getAggregations() {
    return List.of(
        new AggregationSpec(
            MyAggBuilder.NAME,            // "my_agg"
            MyAggBuilder::new,            // StreamInput ctor (wire)
            MyAggBuilder::fromXContent)   // JSON parser
        .addResultReader(InternalMyAgg::new)); // reduce-side wire reader
}

You implement: MyAggBuilder extends AbstractAggregationBuilder, a MyAggregatorFactory, the MyAggregator (its getLeafCollector/collect), and InternalMyAgg extends InternalAggregation with a correct reduce(...). The reduce implementation is the part reviewers scrutinize most — it must be associative and commutative because shard order is non-deterministic and partial reduces batch arbitrarily.

grep -n "getAggregations\|AggregationSpec\|addResultReader" \
  server/src/main/java/org/opensearch/plugins/SearchPlugin.java

See Plugin Architecture for wiring and the Level-7 plugin lab for a build-it walkthrough.

Reading exercise

# 1. Factory -> aggregator creation
grep -n "createInternal\|doCreateInternal\|class TermsAggregatorFactory" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregatorFactory.java

# 2. Collection
grep -n "getLeafCollector\|collect\|collectBucket\|collectExistingBucket" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/BucketsAggregator.java

# 3. Reduce + error accounting
grep -n "reduce\|doc_count_error\|sum_other\|shardSize" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/terms/InternalTerms.java

# 4. The bundled module pattern
grep -rn "getAggregations" modules/aggs-matrix-stats/src/main/java/

Answer:

1. When a terms agg has an avg sub-aggregation, how many avg aggregator instances effectively exist, and how does a document get routed to the right one? (Hint: bucket ordinals.)
2. Why must InternalAggregation.reduce be associative and commutative? Tie your answer to partial reduces and non-deterministic shard ordering.
3. Walk through how doc_count_error_upper_bound is computed for a terms agg. Under what order does the error become unbounded/harder to compute?
4. Where does a terms aggregator read its values from, and what makes bucketing by global ordinal faster than bucketing by string? (Cross-check DocValues and Fielddata.)
5. In aggs-matrix-stats, find getAggregations() and list everything the AggregationSpec registers. Which registration is specifically for the reduce side?
6. What does batched_reduce_size control, and on which node does it matter?

Common bugs and symptoms

Validation: prove you understand this

1. Draw the four-stage lifecycle (build, create, collect, reduce), labeling which stage runs on the coordinator vs the data nodes, and what object crosses the wire between collect and reduce.
2. Explain, with a concrete 3-shard example, how a globally-popular term can be under-reported by a terms agg, and how shard_size reduces the risk.
3. Define doc_count_error_upper_bound and sum_other_doc_count and say what a non-zero doc_count_error_upper_bound tells a user about trusting the counts.
4. For a custom InternalAgg.reduce, write (in pseudocode) the reduce for a sum-style metric and argue it is associative and commutative.
5. Explain why bucketing a terms agg by global ordinal beats bucketing by raw string, and what data structure supplies the ordinals.
6. Using aggs-matrix-stats as the template, list the four classes you'd create to add a new aggregation and the one method whose correctness the reviewer will care about most.