Lab AG3: Composite, Pipeline, and Memory

Background

Three things separate someone who uses aggregations from someone who can operate and contribute to them: pagination (composite — the only paginating agg), reduce-time post-processing (pipeline aggs), and the memory guards that stop a runaway agg from taking down a node. This lab does all three by hand, with real curl, so you have felt the after_key loop, watched a derivative compute on the reduce, and deliberately tripped both search.max_buckets and the request circuit breaker.

This is the operational capstone of the Aggregations intensive — its "composite," "pipeline," and "three layers of memory safety" sections become keystrokes here. It leans on the circuit breakers and memory deep-dive for the byte guard.

Why This Matters for Contributors

The aggregation issues that reach the tracker are rarely "wrong math." They are "my dashboard query OOMs the cluster," "my composite scroll never terminates," "my pipeline returns null." Each has a precise cause — bucket explosion, a missing loop-termination on a short page, a wrong bucketsPath or gap_policy. You learn to recognize all three by causing them on purpose and reading the exception the node actually returns.

Prerequisites

• A running OpenSearch (3.x). ./gradlew run or a tarball/Docker node.
• jq for parsing responses.
• OpenSearch source checkout for the grep steps.
• Read the Aggregations intensive sections on composite, pipeline aggs, and memory safety, plus the circuit breakers deep-dive.

Part A — Composite pagination with after_key

Step 1 — Index data with two dimensions

curl -s -XDELETE 'localhost:9200/events' >/dev/null
curl -s -XPUT 'localhost:9200/events' -H 'Content-Type: application/json' -d '{
  "settings": { "number_of_shards": 1, "number_of_replicas": 0 },
  "mappings": { "properties": {
    "user":      { "type": "keyword" },
    "action":    { "type": "keyword" },
    "ts":        { "type": "date" },
    "bytes":     { "type": "long" }
  }}
}'

# 12 events across 4 users x 3 actions
curl -s -XPOST 'localhost:9200/events/_bulk' -H 'Content-Type: application/json' -d '
{"index":{}}
{"user":"u1","action":"view","ts":"2026-01-01T00:00:00Z","bytes":100}
{"index":{}}
{"user":"u1","action":"click","ts":"2026-01-02T00:00:00Z","bytes":200}
{"index":{}}
{"user":"u2","action":"view","ts":"2026-01-01T00:00:00Z","bytes":150}
{"index":{}}
{"user":"u2","action":"buy","ts":"2026-01-03T00:00:00Z","bytes":300}
{"index":{}}
{"user":"u3","action":"view","ts":"2026-01-02T00:00:00Z","bytes":120}
{"index":{}}
{"user":"u3","action":"click","ts":"2026-01-02T00:00:00Z","bytes":220}
{"index":{}}
{"user":"u4","action":"buy","ts":"2026-01-03T00:00:00Z","bytes":500}
{"index":{}}
{"user":"u1","action":"view","ts":"2026-01-04T00:00:00Z","bytes":110}
{"index":{}}
{"user":"u2","action":"click","ts":"2026-01-05T00:00:00Z","bytes":210}
{"index":{}}
{"user":"u3","action":"buy","ts":"2026-01-06T00:00:00Z","bytes":330}
{"index":{}}
{"user":"u4","action":"view","ts":"2026-01-01T00:00:00Z","bytes":140}
{"index":{}}
{"user":"u4","action":"click","ts":"2026-01-07T00:00:00Z","bytes":230}
'
curl -s -XPOST 'localhost:9200/events/_refresh' >/dev/null

Step 2 — Page through every (user, action) with size: 2

A tiny size forces multiple pages so you actually exercise the loop.

# Page 1 — no "after"
curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": { "combos": { "composite": {
    "size": 2,
    "sources": [
      { "u": { "terms": { "field": "user"   } } },
      { "a": { "terms": { "field": "action" } } }
    ]
  } } }
}' | jq '.aggregations.combos | {after_key, buckets: [.buckets[].key]}'

You get 2 buckets and an after_key. Feed it back:

# Page 2 — pass the after_key from page 1
curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": { "combos": { "composite": {
    "size": 2,
    "after": { "u": "u1", "a": "view" },   <-- replace with page 1 after_key
    "sources": [
      { "u": { "terms": { "field": "user"   } } },
      { "a": { "terms": { "field": "action" } } }
    ]
  } } }
}' | jq '.aggregations.combos | {after_key, buckets: [.buckets[].key]}'

Step 3 — Script the full loop (the real pattern)

Doing it by hand is error-prone; this is how you'd actually drain a composite:

#!/usr/bin/env bash
after='null'
page=0
while : ; do
  page=$((page+1))
  if [ "$after" = "null" ]; then after_clause=''; else after_clause="\"after\": $after,"; fi
  resp=$(curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d "{
    \"size\": 0,
    \"aggs\": { \"combos\": { \"composite\": {
      \"size\": 2, $after_clause
      \"sources\": [
        { \"u\": { \"terms\": { \"field\": \"user\"   } } },
        { \"a\": { \"terms\": { \"field\": \"action\" } } }
      ]
    } } }
  }")
  n=$(echo "$resp" | jq '.aggregations.combos.buckets | length')
  echo "page $page: $n buckets -> $(echo "$resp" | jq -c '[.aggregations.combos.buckets[].key]')"
  after=$(echo "$resp" | jq -c '.aggregations.combos.after_key')
  # TERMINATION: stop when a page returns fewer than size buckets
  [ "$n" -lt 2 ] && break
done

• Run it. Confirm it terminates and the pages, concatenated, enumerate every distinct (user, action) combo exactly once, in sorted order.
• The termination rule is load-bearing: stop when a page returns fewer than size buckets. If you only stop on an empty page you do one extra request; if you never check, the loop never ends. This is the intensive's "loops forever" bug, live.

Step 4 — See the paging machine in the source

cd ~/src/OpenSearch
grep -n "afterKey\|CompositeValuesCollectorQueue\|class CompositeAggregator\|size\|class CompositeKey" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/composite/CompositeAggregator.java
grep -n "afterKey\|class InternalComposite\|buckets" \
  server/src/main/java/org/opensearch/search/aggregations/bucket/composite/InternalComposite.java

• Find the bounded CompositeValuesCollectorQueue (capacity = size) and the after_key it emits. Confirm memory is O(size), independent of total cardinality — the whole reason composite exists.

Note: composite must be top-level and cannot order by a sub-metric. Try nesting it under a terms and read the rejection. That constraint is the price of bounded-memory pagination.

Part B — A pipeline aggregation on the reduce

Step 5 — derivative over a date_histogram

A pipeline agg runs at reduce over the output buckets of a sibling, not over documents. derivative differences consecutive buckets.

curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": {
    "per_day": {
      "date_histogram": { "field": "ts", "calendar_interval": "day" },
      "aggs": {
        "daily_bytes": { "sum": { "field": "bytes" } },
        "bytes_delta": { "derivative": { "buckets_path": "daily_bytes" } }
      }
    }
  }
}' | jq '.aggregations.per_day.buckets[] | {key_as_string, daily_bytes: .daily_bytes.value, bytes_delta: .bytes_delta.value}'

• The first bucket has no bytes_delta (nothing to difference against). Each later bucket's bytes_delta is daily_bytes[i] - daily_bytes[i-1].
• Note: derivative is nested inside the date_histogram and reads its sibling daily_bytes via buckets_path. It never saw a document.

Step 6 — bucket_script combining two metrics

bucket_script evaluates a Painless expression over named sibling metrics per bucket — also at reduce time.

curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": {
    "per_user": {
      "terms": { "field": "user", "size": 10 },
      "aggs": {
        "total_bytes": { "sum": { "field": "bytes" } },
        "event_count": { "value_count": { "field": "bytes" } },
        "bytes_per_event": {
          "bucket_script": {
            "buckets_path": { "tb": "total_bytes", "ec": "event_count" },
            "script": "params.tb / params.ec"
          }
        }
      }
    }
  }
}' | jq '.aggregations.per_user.buckets[] | {key, total_bytes: .total_bytes.value, event_count: .event_count.value, bytes_per_event: .bytes_per_event.value}'

• Confirm bytes_per_event == total_bytes / event_count per user.
• Break it on purpose: change "tb" to a path that doesn't exist and read the error. Then set "gap_policy": "insert_zeros" vs the default skip and see how a missing value is handled.

Step 7 — See pipelines run only on the final reduce

cd ~/src/OpenSearch
grep -n "reducePipelines\|class DerivativePipelineAggregator\|class BucketScriptPipelineAggregator\|buckets_path\|gapPolicy" \
  server/src/main/java/org/opensearch/search/aggregations/pipeline/*.java | head
grep -n "reducePipelines\|isFinalReduce" \
  server/src/main/java/org/opensearch/search/aggregations/InternalAggregation.java

• Confirm pipeline logic runs in reducePipelines, gated on the final reduce — it needs the globally reduced bucket values, so it cannot run per-shard or per-slice. This is why a pipeline agg is invisible to the slice-reduce level of concurrent search.

Part C — Trip the memory guards

Step 8 — Trip search.max_buckets (the count guard)

Lower the cap so you can hit it without a billion docs, then ask for a huge bucket tree.

# Lower the bucket cap for this lab (remember to restore it!)
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": { "search.max_buckets": 20 }
}' | jq

# A composite that enumerates more combos than the cap, plus a sub-agg per bucket:
curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": { "explode": {
    "terms": { "field": "user", "size": 100 },
    "aggs": { "by_action": {
      "terms": { "field": "action", "size": 100 },
      "aggs": { "by_day": { "date_histogram": { "field": "ts", "calendar_interval": "day" } } }
    } }
  } }
}' | jq '.error.type, .error.reason'

• You should get "too_many_buckets_exception" with a message naming search.max_buckets. That is MultiBucketConsumer.accept throwing.
• This is a count guard. It fires on the number of buckets, regardless of how much heap is free.

# Confirm the guard in source:
grep -rn "class MultiBucketConsumer\|MAX_BUCKET_SETTING\|search.max_buckets\|TooManyBucketsException" \
  server/src/main/java/org/opensearch/search/aggregations/MultiBucketConsumerService.java

# Restore the default when done:
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": { "search.max_buckets": null }
}' | jq

Step 9 — Trip the request circuit breaker (the byte guard)

The byte guard is orthogonal — it fires on heap bytes, not bucket count. Lower the request-breaker limit, then run an agg that allocates real BigArrays (a high-cardinality terms or a cardinality).

# Lower the request breaker limit dramatically (lab only — RESTORE after!)
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": { "indices.breaker.request.limit": "1%" }
}' | jq

# Force allocation: many distinct keys + a cardinality sketch per bucket.
curl -s 'localhost:9200/events/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": { "hi_card": {
    "terms": { "field": "user", "size": 10000 },
    "aggs": { "uniq": { "cardinality": { "field": "bytes", "precision_threshold": 40000 } } }
  } }
}' | jq '.error.type, .error.reason' 2>/dev/null

• On a small dataset you may need to scale up docs to actually trip it; the point is to see circuit_breaking_exception with a <request> breaker label and a byte budget in the message, distinct from the bucket-count error.
• Restore the limit:

curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": { "indices.breaker.request.limit": null }
}' | jq

# Confirm where aggs charge the breaker:
grep -rn "addRequestCircuitBreakerBytes\|bigArrays\|REQUEST\|CircuitBreakingException" \
  server/src/main/java/org/opensearch/search/aggregations/AggregatorBase.java | head

Step 10 — Reason about the memory: cardinality + global ordinals

No new commands — reason and write. Answer in your deliverable:

• Where does the heap go in hi_card above? (the terms bucket-ord hash sized by user-cardinality; one HyperLogLogPlusPlus sketch per bucket; the global-ordinal map for the user field.)
• Which guard fires first as you scale: many tiny buckets → max_buckets (count); a few buckets each holding a big HLL++ at precision_threshold: 40000 → the request breaker (bytes). Tie each to its class.
• The global-ordinal cost: loading global ordinals for a very high-cardinality keyword is itself a fielddata/BigArrays allocation — it can trip the breaker before a single bucket forms. Cross-check docvalues/fielddata.

Deliverables

• The composite paging loop output: every (user, action) combo enumerated once, in order, with correct termination.
• derivative and bucket_script outputs, with a one-line note that pipeline aggs run on the final reduce and never see documents.
• Captured too_many_buckets_exception and circuit_breaking_exception responses, each labeled count-guard vs byte-guard with the class that throws.
• A short memory write-up answering the Step 10 questions.

Troubleshooting

Warning: Steps 8–9 lower persistent cluster settings. Restore both to null before you leave, or every subsequent aggregation on this node inherits a tiny cap and breaks confusingly.

Expected Output

• The paging loop prints N pages and terminates on the short final page, covering all distinct combos.
• derivative/bucket_script produce the arithmetic you predict by hand.
• Two distinct exceptions: too_many_buckets_exception (count) and circuit_breaking_exception with a <request> label (bytes).

Stretch Goals

• Composite with a date_histogram source and a missing_bucket: true term source; observe how nulls become their own bucket and how after encodes them.
• cumulative_sum + moving_fn pipeline aggs over per_day; chain a pipeline that reads another pipeline's output (buckets_path to a sibling pipeline).
• Quantify the HLL++ memory by varying precision_threshold (10 → 40000) and watching GET /_nodes/stats/breaker request-breaker estimated_size move. Tie it to the 1.04/sqrt(2^p) error formula from the intensive.
• Reproduce on concurrency. Turn on search.concurrent_segment_search.mode: all, re-run the pipeline query, and confirm the pipeline still computes only once (final reduce) — link the concurrent search masterclass.

Coding Exercises

You drove composite paging, pipelines, and the memory guards by hand with curl. Now turn each behavior into code that asserts it. Locate every class with rg first (rg -l CompositeAggregator server/; rg -l MultiBucketConsumerService server/) — no remembered paths or line numbers.

1. (warm-up) Port the composite paging loop to a test. Write an OpenSearchIntegTestCase that indexes the events docs, drains the composite with size: 2 (loop until a short page), and asserts the concatenated keys enumerate every distinct (user, action) combo exactly once, in sorted order. The test must encode the termination rule (stop on buckets.size() < size), so a regression that loops forever fails fast instead of hanging.

2. (core) Unit-test a pipeline agg's reduce-time arithmetic. Using AggregatorTestCase / the pipeline test base (rg -l "DerivativeTests|BucketScriptTests|class.*PipelineAggregator.*Tests" server/), write a test for derivative over a date_histogram+sum: index a few docs across days and assert each later bucket's delta is value[i] - value[i-1] and the first bucket has none. This pins the Step 5 behavior under JUnit.

3. (core) Assert the count guard throws. Write a test that sets search.max_buckets low (via index/cluster setting or the consumer directly — rg -n "MAX_BUCKET_SETTING|TooManyBucketsException" server/.../MultiBucketConsumerService.java) and runs an aggregation that exceeds it, asserting a TooManyBucketsException/too_many_buckets_exception with the bucket count in the message. This is the executable form of Step 8.

4. (core) Instrument where pipelines run. Add a temporary logger.info in the reducePipelines path gated on isFinalReduce (the line you found in Step 7: rg -n "reducePipelines|isFinalReduce" server/.../InternalAggregation.java). Run the derivative query under both default and search.concurrent_segment_search.mode: all and capture the logs to prove the pipeline executes once, on the final reduce only — never per-slice. Revert the probe after.

5. (advanced) Advanced challenge — a memory-pressure regression test. Write an OpenSearchIntegTestCase that (a) lowers indices.breaker.request.limit, (b) runs a high-cardinality terms + per-bucket cardinality (high precision_threshold) and asserts a CircuitBreakingException with a <request> label, then (c) restores the limit and asserts the same query succeeds. Add a third assertion that reads GET /_nodes/stats/breaker estimated_size before/after and shows it returns to baseline — i.e. the agg released its BigArrays (no leak). Deliverable: one self-contained test distinguishing the byte guard from the count guard and proving clean release, cross-linked to circuit breakers & memory.

Issues to Practice On

The operational failure modes you triggered here (OOM, runaway composite, null pipeline) are exactly what reaches the tracker. Practice on opensearch-project/OpenSearch.

Labels drift — confirm with gh label list --repo opensearch-project/OpenSearch.

Representative patterns. (1) "Composite aggregation never terminates / misses buckets across pages." — reproduce by draining with a tiny size, then trace CompositeAggregator's bounded queue and after_key emission (Step 4). (2) "Query OOMs / trips the request breaker on a heavy aggregation." — reproduce by lowering the breaker limit, confirm it's the byte guard not the bucket-count guard, and check whether the offending agg sizes/releases its BigArrays correctly. Approach: reproduce → locate with rg → fix → test → PR with CHANGELOG + DCO.

Planted-bug drill. In MultiBucketConsumerService (the accept/count method), change the comparison from > to >= (or bump the limit by one) so the guard fires one bucket too late or too early. Rebuild and run the bucket-consumer tests (rg -l "MultiBucketConsumer.*Tests|TooManyBuckets" server/); watch the off-by-one test go red. Revert, then add an assertion that pins the exact boundary: a request producing exactly max_buckets succeeds and max_buckets + 1 throws. That boundary test is what a reviewer wants for any change to the guard.

Etiquette: claim the issue first, reproduce before fixing, and every PR needs a test + CHANGELOG entry + DCO Signed-off-by (git commit -s). See community interaction and the prepare-a-PR lab.

Validation / Self-check

1. Why is composite the only paginating agg? Describe the bounded queue and the exact loop-termination rule, and why memory is O(size).
2. On which reduce does a pipeline agg run, and why can't it run per-shard or per-slice? What does buckets_path name?
3. Name the two memory guards, the exception each throws, and whether it counts buckets or bytes. Give an agg shape that trips each.
4. In a high-cardinality terms + per-bucket cardinality, list every place the heap goes and which guard fires first as you scale.
5. Why can loading global ordinals trip the request breaker before a single bucket is formed?

This closes the Aggregations masterclass. Back to the intensive; related: the circuit breakers and memory deep-dive, the concurrent segment search masterclass (the slice reduce), and the star-tree aggregations fast path.