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.

Coding Exercises

You read the collect/reduce split; now you prove it in code. Every exercise here writes a test against the real reduce path so the rules from this lab stop being prose. Locate each class with rg/grep before you write — never trust a remembered line.

1. (warm-up) A test on InternalAvg.reduce. Build two InternalAvg partials by hand (one with sum=10/count=2, one with sum=20/count=3), reduce them with an InternalAggregation.reduceContext, and assert the result is 30/5 = 6.0 — not the average of the two averages (which would be 5.0). This is the reducible-metric rule as an executable assertion. Find the constructor shape:

   grep -rn "class InternalAvg" server/src/main/java/org/opensearch/search/aggregations/metrics/
   grep -rln "InternalAvg" server/src/test/java/org/opensearch/search/aggregations/metrics/ | head
   

2. (core) A partial-reduce assertion. OpenSearch reduces in two stages: per-batch partial reduces (reduceContext.isFinalReduce() == false) then a final reduce. Write a test that runs a two-step reduce on three InternalTerms partials — reduce the first two partially, then reduce that with the third finally — and assert the merged counts equal a single-shot reduce of all three.

   grep -rn "isFinalReduce\|forPartialReduction\|forFinalReduction" \
     server/src/main/java/org/opensearch/search/aggregations/InternalAggregation.java
   

   This pins the invariant that partial reduction must be associative.

3. (core) Prove reduce is not concatenation for terms. Construct two InternalTerms partials that share a bucket key (e.g. product A appears in both) and assert InternalTerms.reduce merges them by key — one bucket A whose doc_count is the sum, not two A buckets. Then add a sub-agg to each and assert the child avg reduced recursively.

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

4. (core) A shard_size truncation test with AggregatorTestCase. Index documents over a few sub-readers (shards-in-miniature), run a terms agg with a tiny shard_size, and assert doc_count_error_upper_bound > 0; raise shard_size and assert it drops to 0. Use the in-memory harness, no cluster:

   grep -rln "extends AggregatorTestCase" server/src/test/java/org/opensearch/search/aggregations/bucket/terms/ | head
   

   The error bound you eyeballed with curl in Step 5 becomes a deterministic test.

5. (advanced) Advanced challenge — a property-based reduce equivalence test. Write a test that generates N random InternalTerms partials, reduces them in two different groupings (e.g. reduce(reduce(a,b), reduce(c,d)) vs reduce(a,b,c,d)), and asserts the two reduced results are equals. This is a fuzz check that reduce is associative and commutative — the property the whole coordinator depends on. Use the randomized framework (randomIntBetween, randomAlphaOfLength) and pin failures with -Dtests.seed. If your custom or modified agg ever fails this, it is a real reduce bug — exactly the class of bug this lab exists to teach you to catch.

Issues to Practice On

Aggregation reduce bugs ("wrong count after merge," "missing bucket," "wrong error bound") live in opensearch-project/OpenSearch under the aggregations area labels (labels move; confirm on the tracker — gh label list --repo opensearch-project/OpenSearch).

Representative patterns. (1) A "doc_count_error_upper_bound is wrong / inconsistent across shard counts" report — reproduce with a skewed multi-shard index, locate the computation via rg "doc_count_error" server/.../bucket/terms/, and fix the per-bucket error accounting in InternalTerms. (2) A "sub-aggregation count is wrong after reduce" report — the classic "reduce-as-concat" trap; reproduce, locate the recursion in reduceBuckets, fix, and add a sub-agg reduce test. Same loop every time: reproduce → locate via rg → fix → test → PR with CHANGELOG + DCO sign-off.

Planted-bug drill. In InternalTerms (or InternalAvg), introduce a reduce bug on purpose: in reduceBuckets, replace the per-key doc_count sum with taking the max of the two counts; or in InternalAvg.reduce, average the two value()s instead of summing sum/count. Run ./gradlew :server:test --tests "*InternalTermsTests*" (or *InternalAvgTests*) and watch which assertion goes red. Revert, then add one assertion that would have caught the exact bug you planted (e.g. "a key present on two shards reduces to the summed count"). Find the seam first:

grep -rn "reduceBuckets\|doc_count\|getDocCount" server/src/main/java/org/opensearch/search/aggregations/bucket/terms/InternalTerms.java | head

Etiquette: Claim an issue before working it, reproduce on main first, and every PR needs a test + a CHANGELOG.md entry + a DCO Signed-off-by (git commit -s). See community interaction and the level-2 PR lab.

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.