Adding an Aggregator

Embedding Pulse, not contributing to it? This recipe adds a built-in in internal/processing/, which embedders cannot import. To add your own operator, implement the matching contract in the public extend package instead — see Extension Points.

Audience: Pulse internals contributors adding a new AGG_* operator.

This page is a step-by-step recipe. The binding list of companions is CLAUDE.md’s Update Demand table (aggregator row) and its long form, .claude/reference/update-demand.md; this is the human-readable walkthrough.

From CLAUDE.md, The Update Demand (registered aggregator row) and .claude/reference/update-demand.md.

1. Declare the type constant

Add the new constant to types/types.go and the slice returned by types.AllAggregationTypes(). Example, for a hypothetical AGG_GINI:

const (
    // ... existing constants ...
    AGG_GINI AggregationType = "AGG_GINI"
)

func AllAggregationTypes() []AggregationType {
    return []AggregationType{
        // ... existing entries, alphabetised ...
        AGG_GINI,
    }
}

The exhaustiveness tests (TestStreamability_AggregationsKnown and friends) will fail until you add the streamability case in step 4.

2. Implement the aggregator and register it

The operator implementation lives in internal/processing/. Write the factory function (newGini(...) returning the aggregator interface) and register it in aggregatorRegistry in internal/processing/registry.go.

If the aggregator can update one row at a time, also implement the OnlineAggregator interface so it joins the streaming Process path. Sort-based or sum-of-deviation aggregators (like AGG_MEDIAN, AGG_ZSCORE) skip this interface and run in the buffered path.

3. Tests

Tests come first: write them in internal/processing/aggregator_test.go before the implementation, run the suite, confirm they fail informatively, then port the implementation until green. See Testing Conventions.

4. Declare streamability

Add a case for the new type in types/streamability.go:

func (t AggregationType) Streamable() bool {
    switch t {
    // ...
    case AGG_GINI:
        return false // sort-based
    }
}

Add the same row to the table in types/streamability_test.go.

If the aggregator is online, also expect TestRegistryStreamabilityMatchesTypes to compare your OnlineAggregator implementation against the AggregationType.Streamable() return value — they must agree.

5. Update the skill pack

Add a section for the new aggregator in skills/aggregation-design.md. Cover when to use it, what its inputs and outputs look like, and any caveats (sort cost, memory, supported field types).

The CI gate TestSkillsCoverAllComponents parses the skill body for the operator name; the section can live anywhere in the file as long as the name appears.

6. Declare the capability metadata

Add a row to internal/descriptor/capabilities_aggregators.go describing the operator’s params, accepted field types, emitted type, and the streamable hint. TestManifestOperatorsComplete enforces that every registered aggregator has a capability row.

The capability row also carries the operator’s ComponentSchema and Mergeability — both required for every registered aggregator since v0.20.0. The next section walks through both.

6a. Declare the ComponentSchema (Response.Components contract)

Every registered aggregator MUST declare a ComponentSchema on its capability row in internal/descriptor/capabilities_aggregators.go. The schema enumerates the operator-specific keys the aggregator emits into Response.Components.Aggregations[i].Operator and tags the operator with one of three mergeability classes:

The aggSchema helper in capabilities_aggregators.go prepends the universal floor ({"n", "n_null"}) automatically — do not list those keys in your extra slice, the helper adds them:

{
    Name:          string(types.AGG_GINI),
    Category:      "aggregator",
    Description:   "Gini coefficient of the field across the input set.",
    AcceptsTypes:  numericFieldTypesAnalyticsNoDecimal,
    EmitsTypeNote: "scalar float64 in [0, 1]",
    Streamable:    false,
    ComponentSchema: aggSchema(None,
        ComponentKey{Name: "sorted_n", Type: "int", Description: "Number of non-null values seen before the final sort."},
    ),
},

A floor-only aggregator (operator-specific keys empty) is valid — pass no extra arguments and the schema declares only {n, n_null} under the chosen mergeability class. AGG_COUNT is the canonical floor-only operator.

6b. Emit per-operator component values at runtime

Two equivalent paths exist for emitting the operator-specific keys at runtime; pick whichever fits your aggregator type.

Sibling interface (preferred for built-in operators). Implement MetaAggregator (internal/processing) on your aggregator type and add a compile-time assertion in internal/processing/aggregator.go:

type giniAggregator struct {
    // ... operator state ...
}

func (g *giniAggregator) Components() (map[string]any, error) {
    return map[string]any{
        "sorted_n": len(g.values),
    }, nil
}

// In internal/processing/aggregator.go's compile-time assertion block:
var _ MetaAggregator = (*giniAggregator)(nil)

The assertion list near the bottom of internal/processing/aggregator.go is the grep-discoverable record of which operators emit components. Add the new entry so interface drift is caught at build time.

ComponentsFunc closure. Used primarily by pulse.Options.Extensions registrations where you cannot extend the aggregator type. The closure receives the constructed aggregator instance and returns the operator-specific keys map. See Extension Points for the full extension recipe and the probe-validation contract (PULSE_EXTENSION_MISSING_COMPONENT_SCHEMA, PULSE_EXTENSION_COMPONENT_SCHEMA_MISMATCH).

In both paths the orchestrator owns the universal floor {n, n_null} unconditionally — your Components() MUST NOT re-emit those keys. Return (nil, nil) to signal “no operator-specific keys; let the orchestrator fill the floor only” — the canonical AGG_COUNT shape.

The full Response.Components contract (per-family typed shape, streaming behaviour by mergeability class, overlay parity reads) lives in the response-components skill — that file remains the authoritative contract document.

6c. Mergeable-aggregator rule

If your aggregator is mergeable, implement the engine’s MergeableAggregator.MergeOnline(other) and declare it as Mergeable() in AggregationType.Mergeable() (types/streamability.go) so it composes correctly under parallel decode (internal/service/parallel_reduce.go) and shard reduce (internal/service/shard_reduce.go). Both surfaces fold per-worker / per-shard partials in deterministic index order via mergeShardPartials + finalizeMergedPartial.

An aggregator registered but not Mergeable() silently forces the request down the serial scanIter / shardIter path; both parallel paths gate on internal/processing’s CanMergeRequestWithExtensions (the built-in-only CanMergeRequest is its nil-registry case) and fall through cleanly when an entry is not flagged.

An extension aggregator takes the same paths through the public contract instead: it implements extend.MergeableAggregator.Merge(other) and its pulse.AggregatorRegistration declares Streamable: true and Mergeable: true; probe-validation at pulse.New refuses a declaration the value cannot honour (PULSE_EXTENSION_MERGEABLE_MISMATCH). See Extension points — Aggregator.

Crosstab margin class. A built-in cell aggregator is admitted to the fused crosstab arm by AggregationType.MarginReducibility() (types/streamability.go) — classify every new aggregator there (the default is MarginRecompute, which keeps its crosstabs buffered). An extension aggregator declares the same class on its registration’s MarginReducibility field (needs Mergeable; refused otherwise with PULSE_EXTENSION_MARGIN_REDUCIBILITY_MISMATCH); undeclared, its crosstab cells run buffered.

Associative + commutative aggregators (count, sum, min, max, frequency, distinct_count, mode) produce byte-equal merge output; Welford-Pébaÿ aggregators (mean, variance, stddev) use Chan-Welford and stay within ULP of serial. If your aggregator’s online state cannot be folded associatively, leave Mergeable() returning false. See Cohort schema design — Parallel decode for the gate composition and observed perf characteristics.

6d. Vector-input and array-valued aggregators

A built-in that reads a member set instead of one field (the AGG_VEC_* shape) joins vectors.VectorAggregationTypes (internal/vectors/aggregations.go); one whose value is one figure per member also returns true from AggregationType.ArrayValued() (types/streamability.go) and implements Rich (the []float64) and ExpandedElements (params.expand). Every consumer — predict’s field references, the runtime slot rewrite, the crosstab / sort / overlay / chain refusals, the expanded output columns and the recommend planner — asks the one classifier, vectors.AggregationKinds (IsVector, IsArrayValued), so nothing else needs a list. An extension declares the same two facts on its registration (VectorInput, ArrayValued) and is classified identically — see Extension Points — Vector inputs.

A bit-exact merge keys its partials by the row’s GLOBAL record ordinal (Record.MergeBlock: block, shard, head) and combines them in the fixed MergeTree shape, so a shard archive answers its single-file twin’s bits (aggregator_vector.go, block_merge.go).

7. CLAUDE.md and registered-component lists

There is no registered-aggregator count or list in CLAUDE.md to update — CLAUDE.md never hardcodes registered counts; the manifest is the source of truth. The Update Demand operator row (.claude/reference/update-demand.md) requires instead the atomic skill skills/op-agg-<kebab>.md, the capability declaration in internal/descriptor/capabilities_aggregators.go — and an internal/examples/<dir>/*.json example whose _meta.operators names the operator (TestEveryOperatorHasAnExampleTag). Edit CLAUDE.md only if the operator introduces a contract it states directly, and mind TestClaudeMdSizeBudget — long-form prose belongs in .claude/reference/.

If the operator interacts with categorical fields in a special way, also update internal/descriptor/predict.go’s numericAggregations map.

8. Run the gates

go test ./internal/skills/ -run TestSkillsCoverAllComponents
go test ./descriptor/ ./internal/descriptor/ -run 'TestManifest|TestPredict'
go test ./internal/processing/ -run TestRegistryStreamability
go test ./...

The full Update Demand row for aggregators says: atomic skill + capability declaration + an example tag + the existing test coverage. All of them ride in the same PR. See The Update Demand.