Lab 7.2: Aggregations and the Coordinating-Node Reduce

Background

Aggregations are OpenSearch's analytics engine — terms, date_histogram, avg, cardinality, and arbitrarily nested sub-aggregations are what power dashboards. Like search, aggregations run per shard and then reduce on the coordinating node (formerly the master/now cluster-manager role is unrelated here — the coordinating node is whichever node received the request). Unlike top-K search, aggregations can be approximate by design, and a contributor who does not understand where the approximation comes from will write incorrect aggregations and misread bug reports.

In this lab you run real aggregations with curl, then read the framework code that produced them: AggregatorFactory → Aggregator → InternalAggregation.reduce. You will see exactly which work happens on the shard and which happens on the coordinator, and you will measure shard_size vs size approximation error on a real index.

Why This Lab Matters for Contributors

• Aggregation bugs are subtle: "wrong count," "missing bucket," "wrong total after merge" are all reduce bugs. You cannot fix them without understanding the collect/reduce split.
• The most common new-contributor mistake in this subsystem is implementing reduce as a naive concatenation. Seeing a correct reduce makes the trap obvious.
• Lab 7.3 asks you to build an aggregation. This lab is the reading you need first.

Prerequisites

• A running node: ./gradlew run. Set export OS=http://localhost:9200.
• Lab 7.1 completed (you know where the query phase is).
• Read alongside: Aggregations deep dive and DocValues and Fielddata.

Step-by-Step Tasks

Step 1 — Index data with a deliberate skew (10 min)

To see approximation error you need a term distribution where a globally significant term is not locally significant on every shard. Build that on purpose:

curl -s -XPUT "$OS/sales" -H 'Content-Type: application/json' -d '{
  "settings": { "number_of_shards": 5, "number_of_replicas": 0 },
  "mappings": { "properties": {
    "product": { "type": "keyword" },
    "amount":  { "type": "double" },
    "ts":      { "type": "date" }
  }}
}'

# 26 products A..Z; "A" is dominant overall but spread thin across shards
for i in $(seq 1 2000); do
  R=$((RANDOM%100))
  if [ $R -lt 8 ]; then P="A"; else P=$(printf "\\$(printf '%03o' $((66+RANDOM%25)))"); fi
  curl -s -XPOST "$OS/sales/_doc" -H 'Content-Type: application/json' -d "{
    \"product\":\"$P\",\"amount\":$((RANDOM%500)).$((RANDOM%99)),
    \"ts\":\"2026-06-$(printf '%02d' $((1+RANDOM%28)))T00:00:00Z\" }" >/dev/null
done
curl -s -XPOST "$OS/sales/_refresh" >/dev/null
echo "loaded"

Step 2 — Run a terms + date_histogram and read the response (10 min)

curl -s "$OS/sales/_search?size=0" -H 'Content-Type: application/json' -d '{
  "aggs": {
    "top_products": {
      "terms": { "field": "product", "size": 5 },
      "aggs": { "avg_amount": { "avg": { "field": "amount" } } }
    },
    "over_time": {
      "date_histogram": { "field": "ts", "calendar_interval": "week" },
      "aggs": { "revenue": { "sum": { "field": "amount" } } }
    }
  }
}' | python3 -m json.tool | head -60

Note size=0 — you do not want hits, only aggregations, so the query phase collects buckets but skips top-K. Find the two key fields in the top_products response:

• doc_count_error_upper_bound — the worst-case count error for buckets not returned.
• sum_other_doc_count — docs in terms that fell outside the returned top size.

These two fields are the approximation, made visible.

Step 3 — Read the framework: factory → aggregator → internal (25 min)

Locate the framework root:

ls server/src/main/java/org/opensearch/search/aggregations/

Read these four touchpoints. For each, grep first, then read the named method.

(a) AggregationBuilder — the parsed-from-JSON, wire-serializable request. The terms builder:

grep -rn "class TermsAggregationBuilder" server/src/main/java/org/opensearch/search/aggregations/bucket/terms/

Read how size, shardSize, and field are parsed and serialized (writeTo / doXContentBody). Note: it is both Writeable (crosses to data nodes) and ToXContent (round-trips to JSON).

(b) AggregatorFactory — builds an Aggregator per shard:

grep -rn "class AggregatorFactory" server/src/main/java/org/opensearch/search/aggregations/
grep -rn "class TermsAggregatorFactory" server/src/main/java/org/opensearch/search/aggregations/bucket/terms/

Read createInternal(...) — it decides which concrete Aggregator to use (e.g. GlobalOrdinalsStringTermsAggregator vs MapStringTermsAggregator) based on field type and cardinality. This is a performance decision (global ordinals → fast). See DocValues and Fielddata.

(c) Aggregator — collects per document on a shard:

grep -rn "abstract class Aggregator" server/src/main/java/org/opensearch/search/aggregations/Aggregator.java
grep -n "getLeafCollector\|collect\|buildAggregations\|buildAggregation" \
  server/src/main/java/org/opensearch/search/aggregations/Aggregator.java

The two methods that matter:

• getLeafCollector(...) returns a LeafBucketCollector whose collect(doc, bucket) is called for every matching doc — this is the collect loop, reading DocValues.
• buildAggregations(...) / buildAggregation(...) produces the shard's InternalAggregation partial at the end of the query phase.

(d) InternalAggregation — the partial result and the reduce logic:

grep -rn "abstract class InternalAggregation" server/src/main/java/org/opensearch/search/aggregations/InternalAggregation.java
grep -n "reduce" server/src/main/java/org/opensearch/search/aggregations/InternalAggregation.java | head

reduce(List<InternalAggregation> aggregations, ReduceContext) runs on the coordinator. For terms (InternalTerms.reduce):

grep -rn "class InternalTerms" server/src/main/java/org/opensearch/search/aggregations/bucket/terms/
grep -n "reduce\|reduceBuckets\|merge" server/src/main/java/org/opensearch/search/aggregations/bucket/terms/InternalTerms.java | head

Read reduceBuckets. Observe that reduce is not concatenation: it merges buckets by key, sums their doc_count, recursively reduces each bucket's sub-aggregations, re-sorts, and re-applies size. This is the single most important method in the aggregation framework.

Step 4 — Collect vs reduce: trace one terms request through both (10 min)

Fill in this table from what you read (it is the lab's core artifact):

Step 5 — Measure shard_size vs size approximation error (15 min)

Run the same terms agg three ways and compare the returned count for product A and the doc_count_error_upper_bound:

agg () {
  curl -s "$OS/sales/_search?size=0" -H 'Content-Type: application/json' -d "{
    \"aggs\": { \"p\": { \"terms\": { \"field\": \"product\", \"size\": 5, \"shard_size\": $1 } } } }" \
  | python3 -c "import json,sys;d=json.load(sys.stdin)['aggregations']['p'];print('shard_size=$1','err_bound=',d['doc_count_error_upper_bound'],'sum_other=',d['sum_other_doc_count']);[print('  ',b['key'],b['doc_count']) for b in d['buckets']]"
}
agg 1
agg 5
agg 100

You should observe: tiny shard_size inflates doc_count_error_upper_bound and can drop or mis-rank product A; larger shard_size shrinks the error toward zero (at the cost of more data moved in the reduce). Explain why in your log, referencing where each shard truncates to shard_size in buildAggregations before the coordinator ever sees the buckets.

Note: The default shard_size is roughly size * 1.5 + 10 (read it in source rather than trusting this number — grep -rn "shardSize" server/.../TermsAggregationBuilder.java). The trade-off is correctness vs. data moved/CPU in the reduce. A terms agg can never be exact on a multi-shard index without shard_size = full cardinality.

Step 6 — Sub-aggregations and the reduce recursion (10 min)

Your top_products.avg_amount is a metric sub-aggregation. Confirm in source that reduce recurses: when InternalTerms.reduce merges two buckets with the same key, it must also reduce their child aggregations. Grep:

grep -rn "InternalAggregations.reduce\|reduce(.*reduceContext" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/terms/InternalTerms.java | head

Answer: why can avg be reduced correctly across shards from partials alone? (Because avg's InternalAvg partial carries both the sum and the count, not the average — so the coordinator can recompute sum/count globally.) Confirm:

grep -rn "class InternalAvg" server/src/main/java/org/opensearch/search/aggregations/metrics/
grep -n "getValue\|sum\|count" server/src/main/java/org/opensearch/search/aggregations/metrics/InternalAvg.java | head

This is the design rule for any reducible metric: the partial must carry enough state to reduce losslessly. You will apply this rule directly in Lab 7.3.

Reading Exercises

Answer each in your log with the file + grep:

1. Find where a terms aggregator truncates to shard_size before emitting its partial.
2. Find the pipeline aggregations directory — how do pipeline aggs differ from bucket/metric aggs in when they run (they run on the reduce output, not during collect)?

   ls server/src/main/java/org/opensearch/search/aggregations/pipeline/ | head
   

3. Find where size=0 causes the query phase to skip top-K hit collection but still run aggregators.
4. Why does aggregating on a text field fail by default, while keyword works? (Hint: DocValues — text has none unless fielddata: true.)

Implementation Requirements

Deliverable is a Markdown analysis file containing:

• The completed collect vs reduce table from Step 4.
• The shard_size measurement from Step 5 with your three runs and an explanation of the trend.
• A one-paragraph statement of the reducible-metric rule (partials must carry enough state), citing InternalAvg as the example.

Expected Output

• A sales index, 5 shards, ~2000 docs, product A globally dominant.
• A terms response showing nonzero doc_count_error_upper_bound at shard_size=1 and ~0 at shard_size=100.
• A reading log mapping every framework class you grepped to its role.

Troubleshooting

Stretch Goals

1. Add "show_term_doc_count_error": true to the terms agg and read the per-bucket error in the response. Find where it is computed in InternalTerms.
2. Build a cardinality agg and read InternalCardinality — it carries a HyperLogLog++ sketch. Why is the sketch (not a count) the correct partial to ship for reduce?
3. Nest a terms inside a date_histogram and reason about the reduce cost as the product of bucket counts.

Validation / Self-check

1. Which class runs the per-document collect loop, and which method is called per doc?
2. Which class and method run the reduce, and on which node?
3. Why is InternalTerms.reduce not a simple concatenation? Name two things it does.
4. State the reducible-metric rule and explain why InternalAvg ships sum+count, not the average.
5. What do doc_count_error_upper_bound and sum_other_doc_count mean, and what makes them nonzero?
6. Why can a multi-shard terms agg never be exact without shard_size = full cardinality?
7. Why do aggregations read DocValues rather than the inverted index?

Cross-references: Aggregations deep dive, DocValues and Fielddata, Lab 7.3: Build a Custom Aggregation, Issue roadmap: search and aggregations.