Lab AG2: Build a Custom Aggregation

Background

You are going to build a brand-new metric aggregation — weighted_avg_sum, a weighted sum Σ(value_i · weight_i) over two numeric fields — and ship it as an OpenSearch SearchPlugin. It is small enough to fit in one lab and big enough to exercise all five classes the framework demands: the AggregationBuilder (wire + XContent), the AggregatorFactory, the Aggregator (collect + buildAggregation), the InternalAggregation (reduce), and the SearchPlugin.getAggregations() registration. You will write a test against AggregatorTestCase, build the plugin, install it, and call it via _search.

This is the apprenticeship piece of the Aggregations intensive: you already traced an existing agg in Lab AG1; now you produce one whose reduce a reviewer would accept.

Why a weighted sum? It is a pure metric (no buckets — keeps the example focused), it reads two value sources (shows the values-source wiring), and its reduce is a plain sum — associative and commutative, which is exactly the property the two-level reduce demands. You will reason about that property explicitly.

Why This Matters for Contributors

Most "I want OpenSearch to compute X" features land as a custom aggregation, and the PR lives or dies on the reduce method. Reviewers scrutinize reduce for associativity/commutativity (does it survive partial reduces and concurrent slice reduces?), the NamedWriteable registration (does the partial cross the wire?), and the values-source plumbing (does it read doc values correctly, including missing/multi-valued?). Doing this once, end to end, with a test, is how you stop being scared of the aggregation framework.

Prerequisites

• OpenSearch source checkout at ~/src/OpenSearch; ./gradlew works.
• JDK matching the checkout (./gradlew -version to confirm).
• Read Aggregations intensive (the collection lifecycle and the reduce section) and skim modules/aggs-matrix-stats as the in-tree template for "an aggregation shipped via the plugin SPI."
• Read Plugin Architecture for how SearchPlugin is loaded.

Step-by-Step Tasks

Step 1 — Scaffold a plugin module

Create a plugin skeleton. You can copy the layout of an existing simple plugin or hand-roll it. Minimal layout:

plugins/agg-weighted/
  build.gradle
  src/main/java/org/opensearch/agg/weighted/
    WeightedAvgSumAggregationBuilder.java
    WeightedAvgSumAggregatorFactory.java
    WeightedAvgSumAggregator.java
    InternalWeightedAvgSum.java
    WeightedAggPlugin.java
  src/test/java/org/opensearch/agg/weighted/
    WeightedAvgSumAggregatorTests.java

build.gradle (the essentials — match versions to your checkout):

apply plugin: 'opensearch.opensearchplugin'
apply plugin: 'opensearch.yaml-rest-test'

opensearchplugin {
  name        'agg-weighted'
  description 'A weighted_avg_sum metric aggregation'
  classname   'org.opensearch.agg.weighted.WeightedAggPlugin'
}

cd ~/src/OpenSearch
# Confirm the plugin gradle plugin id and a real example to copy from:
grep -rn "opensearch.opensearchplugin\|classname" plugins/*/build.gradle | head
ls plugins/   # pick a small one (e.g. analysis-*) to mirror structure

Step 2 — The AggregationBuilder (wire + XContent + tree)

It parses the JSON, holds the two field configs, serializes over the wire, and creates the factory. It extends a multi-values-source builder so it can carry the value and weight fields.

package org.opensearch.agg.weighted;

import org.opensearch.core.ParseField;
import org.opensearch.core.common.io.stream.StreamInput;
import org.opensearch.core.common.io.stream.StreamOutput;
import org.opensearch.core.xcontent.XContentBuilder;
import org.opensearch.index.query.QueryShardContext;
import org.opensearch.search.aggregations.AggregationBuilder;
import org.opensearch.search.aggregations.AggregatorFactories;
import org.opensearch.search.aggregations.AggregatorFactory;
import org.opensearch.search.aggregations.support.*;

import java.io.IOException;
import java.util.Map;
import java.util.Objects;

public class WeightedAvgSumAggregationBuilder
        extends MultiValuesSourceAggregationBuilder.LeafOnly<WeightedAvgSumAggregationBuilder> {

    public static final String NAME = "weighted_avg_sum";
    public static final ParseField VALUE_FIELD  = new ParseField("value");
    public static final ParseField WEIGHT_FIELD = new ParseField("weight");

    // XContent parser: registered in the plugin (Step 6).
    public static final MultiValuesSourceParseHelper.ParserSupplier<WeightedAvgSumAggregationBuilder> PARSER_NAME = null; // see Step 6

    public WeightedAvgSumAggregationBuilder(String name) {
        super(name);
    }

    // ---- wire ----
    public WeightedAvgSumAggregationBuilder(StreamInput in) throws IOException {
        super(in);
    }

    @Override
    protected void innerWriteTo(StreamOutput out) { /* no extra fields beyond the values-sources */ }

    @Override
    protected ValuesSourceType defaultValueSourceType() {
        return CoreValuesSourceType.NUMERIC;
    }

    @Override
    public String getType() { return NAME; }

    @Override
    public BucketCardinality bucketCardinality() { return BucketCardinality.NONE; } // metric

    // ---- create factory ----
    @Override
    protected MultiValuesSourceAggregatorFactory innerBuild(
            QueryShardContext queryShardContext,
            Map<String, ValuesSourceConfig> configs,
            DocValueFormat format,
            AggregatorFactory parent,
            AggregatorFactories.Builder subFactoriesBuilder) throws IOException {
        return new WeightedAvgSumAggregatorFactory(
            name, configs, format, queryShardContext, parent, subFactoriesBuilder, metadata);
    }

    @Override
    public XContentBuilder doXContentBody(XContentBuilder builder, Params params) {
        return builder; // the values-sources serialize themselves in the superclass
    }

    @Override
    protected AggregationBuilder shallowCopy(AggregatorFactories.Builder f, Map<String, Object> m) {
        WeightedAvgSumAggregationBuilder copy = new WeightedAvgSumAggregationBuilder(name);
        copy.metadata = m;
        return copy;
    }

    @Override public boolean equals(Object o) { return o instanceof WeightedAvgSumAggregationBuilder && super.equals(o); }
    @Override public int hashCode() { return Objects.hash(super.hashCode()); }
}

Note: Class/method names on the multi-values-source base (MultiValuesSourceAggregationBuilder.LeafOnly, innerBuild, MultiValuesSourceParseHelper) vary slightly by version. Confirm against the in-tree weighted_avg (single-value-pair) and matrix_stats:

grep -rln "MultiValuesSourceAggregationBuilder\|MultiValuesSourceParseHelper" \
  server/src/main/java/org/opensearch/search/aggregations/ \
  modules/aggs-matrix-stats/src/main/java/
grep -rn "class WeightedAvgAggregationBuilder" server/src/main/java/org/opensearch/search/aggregations/metrics/

Step 3 — The AggregatorFactory

It validates and creates the Aggregator per shard. It carries the resolved ValuesSourceConfigs for value and weight.

package org.opensearch.agg.weighted;

import org.opensearch.index.query.QueryShardContext;
import org.opensearch.search.DocValueFormat;
import org.opensearch.search.aggregations.*;
import org.opensearch.search.aggregations.support.*;

import java.io.IOException;
import java.util.Map;

public class WeightedAvgSumAggregatorFactory extends MultiValuesSourceAggregatorFactory {

    public WeightedAvgSumAggregatorFactory(
            String name,
            Map<String, ValuesSourceConfig> configs,
            DocValueFormat format,
            QueryShardContext queryShardContext,
            AggregatorFactory parent,
            AggregatorFactories.Builder subFactoriesBuilder,
            Map<String, Object> metadata) throws IOException {
        super(name, configs, format, queryShardContext, parent, subFactoriesBuilder, metadata);
    }

    @Override
    protected Aggregator createUnmapped(SearchContext searchContext, Aggregator parent, Map<String, Object> metadata) throws IOException {
        // No mapping for the fields on this shard -> emit an empty result.
        return new WeightedAvgSumAggregator(name, null, null, format, searchContext, parent, metadata);
    }

    @Override
    protected Aggregator doCreateInternal(
            SearchContext searchContext,
            Map<String, ValuesSourceConfig> configs,
            DocValueFormat format,
            Aggregator parent,
            CardinalityUpperBound cardinality,   // <-- the sizing hint
            Map<String, Object> metadata) throws IOException {
        ValuesSource.Numeric valueVS  = (ValuesSource.Numeric) configs.get("value").getValuesSource();
        ValuesSource.Numeric weightVS = (ValuesSource.Numeric) configs.get("weight").getValuesSource();
        return new WeightedAvgSumAggregator(name, valueVS, weightVS, format, searchContext, parent, metadata);
    }
}

grep -rn "class MultiValuesSourceAggregatorFactory\|doCreateInternal\|createUnmapped\|CardinalityUpperBound" \
  server/src/main/java/org/opensearch/search/aggregations/support/MultiValuesSourceAggregatorFactory.java

Step 4 — The Aggregator (collect + buildAggregation)

The heart. It keeps one DoubleArray of running weighted sums, indexed by owning bucket ordinal so it works as a sub-agg too. Note the grow/addRequestCircuitBreakerBytes handling — that's the byte guard from the intensive.

package org.opensearch.agg.weighted;

import org.apache.lucene.index.LeafReaderContext;
import org.opensearch.common.util.BigArrays;
import org.opensearch.common.util.DoubleArray;
import org.opensearch.index.fielddata.SortedNumericDoubleValues;
import org.opensearch.search.DocValueFormat;
import org.opensearch.search.aggregations.*;
import org.opensearch.search.aggregations.metrics.NumericMetricsAggregator;
import org.opensearch.search.aggregations.support.ValuesSource;
import org.opensearch.search.internal.SearchContext;

import java.io.IOException;
import java.util.Map;

public class WeightedAvgSumAggregator extends NumericMetricsAggregator.SingleValue {

    private final ValuesSource.Numeric valueVS;
    private final ValuesSource.Numeric weightVS;
    private final DocValueFormat format;
    private DoubleArray sums;   // indexed by owningBucketOrd

    WeightedAvgSumAggregator(String name, ValuesSource.Numeric valueVS, ValuesSource.Numeric weightVS,
                             DocValueFormat format, SearchContext context,
                             Aggregator parent, Map<String, Object> metadata) throws IOException {
        super(name, context, parent, metadata);
        this.valueVS = valueVS;
        this.weightVS = weightVS;
        this.format = format;
        if (valueVS != null) {
            this.sums = context.bigArrays().newDoubleArray(1, true);  // accounted on the request breaker
        }
    }

    @Override
    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, LeafBucketCollector sub) throws IOException {
        if (valueVS == null || weightVS == null) {
            return LeafBucketCollector.NO_OP_COLLECTOR;   // unmapped shard
        }
        final BigArrays bigArrays = context.bigArrays();
        final SortedNumericDoubleValues values  = valueVS.doubleValues(ctx);
        final SortedNumericDoubleValues weights = weightVS.doubleValues(ctx);
        return new LeafBucketCollectorBase(sub, values) {
            @Override
            public void collect(int doc, long owningBucketOrd) throws IOException {
                sums = bigArrays.grow(sums, owningBucketOrd + 1);     // size by owning ord
                if (values.advanceExact(doc) && weights.advanceExact(doc)) {
                    // first value/weight only (keep the example single-valued)
                    double v = values.nextValue();
                    double w = weights.nextValue();
                    sums.increment(owningBucketOrd, v * w);
                }
            }
        };
    }

    @Override
    public double metric(long owningBucketOrd) {
        return (sums == null || owningBucketOrd >= sums.size()) ? 0.0 : sums.get(owningBucketOrd);
    }

    @Override
    public InternalAggregation buildAggregation(long owningBucketOrd) throws IOException {
        if (sums == null || owningBucketOrd >= sums.size()) {
            return buildEmptyAggregation();
        }
        return new InternalWeightedAvgSum(name, sums.get(owningBucketOrd), format, metadata());
    }

    @Override
    public InternalAggregation buildEmptyAggregation() {
        return new InternalWeightedAvgSum(name, 0.0, format, metadata());
    }

    @Override
    public void doClose() {
        org.opensearch.common.lease.Releasables.close(sums);   // free the BigArray
    }
}

Warning: Three things reviewers will flag if missing: (1) bigArrays.grow sizing by owningBucketOrd (so the agg is correct as a sub-agg, not just at top level); (2) Releasables.close(sums) in doClose (otherwise you leak request breaker bytes); (3) the unmapped/NO_OP_COLLECTOR path (so a shard missing the field doesn't NPE). All three come straight from the intensive's memory-safety and owningBucketOrd sections.

grep -rn "class NumericMetricsAggregator\|abstract.*metric(long\|newDoubleArray\|Releasables.close" \
  server/src/main/java/org/opensearch/search/aggregations/metrics/NumericMetricsAggregator.java
grep -rn "class SumAggregator\|increment\|compensation" \
  server/src/main/java/org/opensearch/search/aggregations/metrics/SumAggregator.java

Step 5 — The InternalAggregation (reduce)

The wire-and-reduce object. reduce is a plain sum — write it and then prove it's associative and commutative.

package org.opensearch.agg.weighted;

import org.opensearch.core.common.io.stream.StreamInput;
import org.opensearch.core.common.io.stream.StreamOutput;
import org.opensearch.core.xcontent.XContentBuilder;
import org.opensearch.search.DocValueFormat;
import org.opensearch.search.aggregations.InternalAggregation;
import org.opensearch.search.aggregations.metrics.InternalNumericMetricsAggregation;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;

public class InternalWeightedAvgSum extends InternalNumericMetricsAggregation.SingleValue {

    private final double sum;

    InternalWeightedAvgSum(String name, double sum, DocValueFormat format, Map<String, Object> metadata) {
        super(name, metadata);
        this.sum = sum;
        this.format = format;
    }

    // ---- wire (the result reader registered in the plugin) ----
    public InternalWeightedAvgSum(StreamInput in) throws IOException {
        super(in);
        format = in.readNamedWriteable(DocValueFormat.class);
        sum = in.readDouble();
    }

    @Override
    protected void doWriteTo(StreamOutput out) throws IOException {
        out.writeNamedWriteable(format);
        out.writeDouble(sum);
    }

    @Override public String getWriteableName() { return WeightedAvgSumAggregationBuilder.NAME; }

    @Override public double value() { return sum; }

    // ---- THE reduce: sum of sums. Associative + commutative. ----
    @Override
    public InternalWeightedAvgSum reduce(List<InternalAggregation> aggregations, ReduceContext reduceContext) {
        double total = 0;
        for (InternalAggregation a : aggregations) {
            total += ((InternalWeightedAvgSum) a).sum;   // order-independent
        }
        return new InternalWeightedAvgSum(name, total, format, getMetadata());
    }

    @Override
    public XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
        builder.field("value", sum);
        if (format != DocValueFormat.RAW) {
            builder.field("value_as_string", format.format(sum).toString());
        }
        return builder;
    }

    @Override public int hashCode() { return Objects.hash(super.hashCode(), sum); }
    @Override public boolean equals(Object o) {
        return o instanceof InternalWeightedAvgSum && super.equals(o)
            && Double.compare(((InternalWeightedAvgSum) o).sum, sum) == 0;
    }
}

The associativity/commutativity argument (write this in your deliverable): reduce is total = Σ sum_i. Real addition is associative ((a+b)+c = a+(b+c)) and commutative (a+b = b+a), so the result is independent of (a) the order shards/slices arrive and (b) how partial reduces batch them. Therefore the agg is correct at both the slice reduce and the coordinator reduce. Contrast: if reduce returned sum_0 (first wins) or a floating average of per-shard averages, it would be order-dependent and wobble under concurrency — the exact bug class the intensive warns about.

grep -rn "class InternalNumericMetricsAggregation\|SingleValue\|abstract.*reduce(" \
  server/src/main/java/org/opensearch/search/aggregations/metrics/InternalNumericMetricsAggregation.java

Step 6 — Register via SearchPlugin.getAggregations()

package org.opensearch.agg.weighted;

import org.opensearch.plugins.Plugin;
import org.opensearch.plugins.SearchPlugin;
import org.opensearch.search.aggregations.support.MultiValuesSourceParseHelper;

import java.util.List;

public class WeightedAggPlugin extends Plugin implements SearchPlugin {

    @Override
    public List<AggregationSpec> getAggregations() {
        return List.of(
            new AggregationSpec(
                    WeightedAvgSumAggregationBuilder.NAME,           // "weighted_avg_sum"
                    WeightedAvgSumAggregationBuilder::new,           // StreamInput ctor (wire)
                    (parser, name) -> PARSER.parse(parser, name))    // JSON parser (below)
                .addResultReader(InternalWeightedAvgSum::new)        // <-- reduce-side wire reader
        );
    }

    // Build the XContent parser for the two value sources.
    private static final org.opensearch.common.xcontent.ContextParser<String, WeightedAvgSumAggregationBuilder> PARSER;
    static {
        org.opensearch.core.xcontent.ObjectParser<WeightedAvgSumAggregationBuilder, String> p =
            new org.opensearch.core.xcontent.ObjectParser<>(
                WeightedAvgSumAggregationBuilder.NAME, WeightedAvgSumAggregationBuilder::new);
        MultiValuesSourceParseHelper.declareCommon(p, true, org.opensearch.search.aggregations.support.CoreValuesSourceType.NUMERIC);
        MultiValuesSourceParseHelper.declareField("value",  p, true, false, false, false);
        MultiValuesSourceParseHelper.declareField("weight", p, true, false, false, false);
        PARSER = p::parse;
    }
}

Note: addResultReader(InternalWeightedAvgSum::new) is the single most commonly-forgotten line, and its omission is invisible on one shard and fatal on many — the partial can't be deserialized for reduce. The deep-dive bugs table flags exactly this. Confirm the AggregationSpec API:

grep -n "getAggregations\|class AggregationSpec\|addResultReader\|MultiValuesSourceParseHelper" \
  server/src/main/java/org/opensearch/plugins/SearchPlugin.java \
  server/src/main/java/org/opensearch/search/aggregations/support/MultiValuesSourceParseHelper.java

Step 7 — Test with AggregatorTestCase

This runs the aggregator over an in-memory Lucene index — no cluster needed. It is the fastest correctness loop and what CI runs.

package org.opensearch.agg.weighted;

import org.apache.lucene.document.Document;
import org.apache.lucene.document.SortedNumericDocValuesField;
import org.apache.lucene.tests.index.RandomIndexWriter;
import org.opensearch.index.mapper.MappedFieldType;
import org.opensearch.index.mapper.NumberFieldMapper;
import org.opensearch.search.aggregations.AggregatorTestCase;

import static org.apache.lucene.document.NumericDocValuesField.*;

public class WeightedAvgSumAggregatorTests extends AggregatorTestCase {

    public void testWeightedSum() throws Exception {
        MappedFieldType valueFt  = new NumberFieldMapper.NumberFieldType("v", NumberFieldMapper.NumberType.DOUBLE);
        MappedFieldType weightFt = new NumberFieldMapper.NumberFieldType("w", NumberFieldMapper.NumberType.DOUBLE);

        WeightedAvgSumAggregationBuilder agg = new WeightedAvgSumAggregationBuilder("ws");
        // value=v, weight=w  (use the multi-values-source setters; confirm names by grep)
        // agg.value(new MultiValuesSourceFieldConfig.Builder().setFieldName("v").build());
        // agg.weight(...);

        testCase(agg, new org.apache.lucene.search.MatchAllDocsQuery(), iw -> {
            iw.addDocument(doc(2.0, 3.0));   // 6
            iw.addDocument(doc(5.0, 1.0));   // 5
            iw.addDocument(doc(1.0, 4.0));   // 4
        }, result -> {
            InternalWeightedAvgSum ws = (InternalWeightedAvgSum) result;
            assertEquals(15.0, ws.value(), 0.0);   // 6 + 5 + 4
        }, valueFt, weightFt);
    }

    private static Iterable<org.apache.lucene.index.IndexableField> doc(double v, double w) {
        Document d = new Document();
        d.add(new SortedNumericDocValuesField("v", Double.doubleToRawLongBits(v)));
        d.add(new SortedNumericDocValuesField("w", Double.doubleToRawLongBits(w)));
        return d;
    }
}

# Confirm AggregatorTestCase's testCase(...) overloads and the multi-values config:
grep -rn "protected .* testCase\|class AggregatorTestCase\|MultiValuesSourceFieldConfig" \
  test/framework/src/main/java/org/opensearch/search/aggregations/AggregatorTestCase.java \
  server/src/main/java/org/opensearch/search/aggregations/support/MultiValuesSourceFieldConfig.java
# Run just your test:
./gradlew :plugins:agg-weighted:test --tests "*WeightedAvgSumAggregatorTests*"

Step 8 — Build, install, exercise

cd ~/src/OpenSearch
./gradlew :plugins:agg-weighted:assemble
# the zip lands under plugins/agg-weighted/build/distributions/

# Option A: run a dev node with the plugin (fastest):
./gradlew :plugins:agg-weighted:run

# Option B: install into a tarball:
bin/opensearch-plugin install \
  file:///$HOME/src/OpenSearch/plugins/agg-weighted/build/distributions/agg-weighted-*.zip

Then exercise it:

curl -s -XPUT 'localhost:9200/sales' -H 'Content-Type: application/json' -d '{
  "mappings": { "properties": { "v": {"type":"double"}, "w": {"type":"double"} } } }'
curl -s -XPOST 'localhost:9200/sales/_bulk' -H 'Content-Type: application/json' -d '
{"index":{}}
{"v":2,"w":3}
{"index":{}}
{"v":5,"w":1}
{"index":{}}
{"v":1,"w":4}
'
curl -s -XPOST 'localhost:9200/sales/_refresh' >/dev/null

curl -s 'localhost:9200/sales/_search' -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": { "ws": { "weighted_avg_sum": {
    "value":  { "field": "v" },
    "weight": { "field": "w" }
  } } }
}' | jq '.aggregations.ws'

Expected: { "value": 15.0 } (2·3 + 5·1 + 1·4).

• Now nest it under a terms to prove the owningBucketOrd sizing works: add a keyword field, group by it, and confirm each bucket gets its own weighted sum.

Deliverables

• Five compiling classes: builder, factory, aggregator, internal, plugin.
• A passing AggregatorTestCase test.
• _search returns the correct weighted sum, including nested under terms (proves per-owning-ord sizing).
• A written paragraph proving reduce is associative and commutative, and naming the three reviewer-checklist items you handled (grow by owning ord, Releasables.close, addResultReader).

Troubleshooting

Expected Output

• :plugins:agg-weighted:test green.
• _search returns {"value": 15.0} top-level and per-bucket values when nested.
• The plugin shows in GET /_cat/plugins.

Stretch Goals

• Multi-valued handling. Decide and document what value/weight do when a field has multiple values per doc (sum all? error?). Read how weighted_avg and sum handle it.
• Compensated summation. Replace the naive += with Kahan/Neumaier compensation like SumAggregator to reduce float error; add a test with values of very different magnitudes.
• A pipeline-friendly output. Make sure value() is readable by a bucket_script pipeline agg (used in Lab AG3).
• Concurrency test. Add a test that forces multiple slices and asserts the result equals the sequential result — the concrete check from the concurrent search masterclass.

Coding Exercises

You already wrote one aggregation end to end. These exercises harden it the way a reviewer would — every one produces code (a new test, a new method, a variant agg). Confirm base-class signatures with rg before you write (rg -n "doCreateInternal|createUnmapped" server/.../support/MultiValuesSourceAggregatorFactory.java).

1. (warm-up) Finish the multi-values wiring in the test. The Step 7 test leaves the value(...)/weight(...) setters commented. Find the real setter names and MultiValuesSourceFieldConfig builder (rg -n "class MultiValuesSourceFieldConfig|public .* value\(|public .* weight\(" server/), wire them, and make testWeightedSum assert 15.0 for real. Run ./gradlew :plugins:agg-weighted:test --tests "*WeightedAvgSumAggregatorTests*".

2. (core) A standalone reduce test. Independent of the cluster, write an OpenSearchTestCase that builds three InternalWeightedAvgSum partials with known sums and asserts reduce(...) returns their total. Then split them into two groups, reduce each group, reduce the two partials, and assert the result equals the one-shot reduce — your executable proof that the reduce is associative. (This is the test that would catch a future refactor breaking partial reduce.)

3. (core) An empty/unmapped-shard test. Add a test that runs the aggregation over a segment where value/weight are absent, asserting it returns the empty result (0.0) and does not NPE — exercising the createUnmapped / NO_OP_COLLECTOR path. Reviewers always check this; prove it with a test rather than by eyeballing.

4. (core) Add a count so it becomes a true weighted average. Extend the agg to carry a second DoubleArray of summed weights (Σweight_i) and emit Σ(v·w) / Σw as a new weighted_average field — keeping the existing weighted_avg_sum value too. Size both arrays by owningBucketOrd, release both in doClose, and update InternalWeightedAvgSum's wire format + reduce (now you reduce two sums). Add a test asserting the weighted average, nested under terms, matches a hand computation per bucket.

5. (advanced) Advanced challenge — make it survive concurrency and refactor it into a pipeline-friendly metric. (a) Apply Kahan/Neumaier compensated summation like SumAggregator (rg -n "compensation|Kahan" server/.../metrics/SumAggregator.java) so high-magnitude inputs don't lose precision; add a test mixing 1e16 and 1.0 that fails on naive += and passes after. (b) Write an OpenSearchIntegTestCase that forces multiple segments and runs with search.concurrent_segment_search.mode: all, asserting the slice-reduced result equals the single-slice result. (c) Make value() consumable by a bucket_script pipeline agg (the bridge to Lab AG3) and add a yaml-rest-test using it. Deliverable: a green compensated, concurrency-proof, pipeline-consumable agg with all three tests.

Issues to Practice On

The aggregation framework you just touched is one of the most active areas on opensearch-project/OpenSearch. Practice on real work there.

Labels move; confirm with gh label list --repo opensearch-project/OpenSearch before relying on one.

Representative patterns. (1) "Add a new metric/pipeline aggregation X" — this is precisely the five-class shape you built; the work is the reduce correctness, the addResultReader registration, and a thorough AggregatorTestCase. (2) "Custom/plugin aggregation breaks on multi-shard reduce." — nearly always a missing or wrong NamedWriteable/result-reader, or a non-associative reduce. Reproduce on a 3-shard index, locate via rg, fix, add a partial-reduce test.

Planted-bug drill. Delete the .addResultReader(InternalWeightedAvgSum::new) line from WeightedAggPlugin.getAggregations(). Run your test on one shard (green) then on a multi-shard OpenSearchIntegTestCase (red — the partial can't be deserialized for reduce). Restore the line, then add an integration test that runs the agg across ≥2 shards so this regression can never silently return. This is the single most common custom-aggregation bug; now you have a test that catches it.

Etiquette: claim the issue with a comment first, reproduce before you fix, 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. List the five classes and the one method whose correctness a reviewer cares about most. Why that one?
2. State the associativity/commutativity argument for your reduce in one or two sentences, and give a reduce that would violate it.
3. Why must the aggregator size its DoubleArray by owningBucketOrd rather than keep a single scalar? What breaks if it doesn't?
4. What does addResultReader register, and what symptom appears on a multi-shard cluster if you omit it?
5. Where are the BigArrays you allocate accounted, and what must doClose do?

Next: Lab AG3 — Composite, Pipeline, and Memory. You built an agg; next you page, pipeline, and break it on memory limits.