Brainy Aggregation Engine

Brainy provides a Rust-accelerated aggregation engine for Brainy, enabling real-time analytics on entity data with incremental updates, parallel rebuilds, and precise statistical operations.

Architecture

Brainy owns the storage and lifecycle. Brainy owns the compute.

┌─────────────────────────────────────────┐
│              Brainy                      │
│  ┌─────────────────────────────────┐    │
│  │   AggregationIndex              │    │
│  │   ├── defineAggregate()         │    │
│  │   ├── removeAggregate()         │    │
│  │   ├── onEntityAdd/Update/Delete │    │
│  │   └── query()                   │    │
│  └──────────────┬──────────────────┘    │
│                 │ provider interface     │
│  ┌──────────────▼──────────────────┐    │
│  │   Brainy AggregationProvider    │    │
│  │   ├── NativeAggregationEngine   │───── Rust (NAPI)
│  │   ├── value multiset (min/max,  │    │
│  │   │   percentile, distinctCount)│    │
│  │   ├── HAVING + array-unnest      │    │
│  │   ├── Rayon parallel rebuild    │    │
│  │   └── Welford's online stddev   │    │
│  └─────────────────────────────────┘    │
└─────────────────────────────────────────┘

When Brainy is installed as a Brainy plugin, the aggregation provider automatically registers. All aggregation computation runs in Rust through NAPI bindings while Brainy handles storage, persistence, and the public API.

Operations

Brainy supports all 9 aggregation operations:

percentile requires a p property in [0, 1] on the metric definition ({ op: 'percentile', field: 'latency', p: 0.95 }). Both percentile and distinctCount are exact — they read the same per-group value multiset that backs precise MIN/MAX, so they are never approximate and never stale after deletes. Percentile uses standard linear interpolation between ranks, verified against numpy's percentiles in src/aggregation/aggregation.test.ts.

Exact value multiset (MIN/MAX, percentile, distinctCount)

Unlike simpler implementations that become stale after deletes, Brainy keeps a per-group Rust BTreeMap<OrderedFloat<f64>, u32> (a sorted value multiset) tracking the exact frequency of every value:

• Add: Insert or increment the count for the value

• Delete: Decrement the count; remove the key if count reaches zero

• MIN / MAX: first / last key of the multiset

• percentile: walk the sorted multiset to the target rank, linear-interpolating between neighbours

• distinctCount: number of keys in the multiset

This single structure guarantees exact MIN, MAX, percentile, and distinct-count after any sequence of add/update/delete operations without requiring a full rescan.

Welford's Online Algorithm

Standard deviation and variance use Welford's numerically-stable online algorithm with M2 tracking (sum of squared differences from the running mean). This computes incrementally without storing all values:

On add(x):
  n += 1
  delta = x - mean
  mean += delta / n
  delta2 = x - mean
  M2 += delta * delta2

Sample variance = M2 / (n - 1)
Sample stddev = sqrt(variance)

Rayon Parallel Rebuild

When rebuilding an aggregate from scratch (e.g., after definition change or cold start), Brainy uses Rayon's parallel iterators to process entities across all CPU cores:

• Entities below 1,000 are processed sequentially (overhead not worth it)

• Above 1,000, Rayon splits the work across threads

• Each thread computes partial aggregation state

• Results are merged with thread-safe combining

For 100K entities with 20 groups, rebuild completes in ~15ms.

Time Window Bucketing

GroupBy dimensions can specify time windows for temporal aggregation. The native engine performs integer-based timestamp bucketing without allocating Date objects:

Keys are computed with pure integer arithmetic (no Date allocation) and are byte-for-byte identical to Brainy's bucketTimestamp() (native/src/aggregation/timestamp.rs).

Combined groupBy dimensions (plain field + time window) produce composite group keys:

brain.defineAggregate({
  name: 'monthly_sales',
  source: { type: 'Event' },
  groupBy: ['region', { field: 'date', window: 'month' }],
  metrics: {
    revenue: { op: 'sum', field: 'amount' },
    count: { op: 'count' }
  }
})TYPESCRIPT

This produces groups like { region: 'US', date: '2024-01' }.

Array-Unnest GroupBy

A groupBy dimension can unnest an array field, so an entity contributes once per distinct array element. The classic use is tag/label frequency:

brain.defineAggregate({
  name: 'tag_frequency',
  source: { type: 'Post' },
  groupBy: [{ field: 'tags', unnest: true }],
  metrics: { count: { op: 'count' } }
})TYPESCRIPT

A post with tags: ['rust', 'db', 'rust'] contributes once to rust and once to db (duplicates within one entity are de-duplicated). An entity with a missing or empty array contributes to no group.

HAVING

queryAggregate can filter groups after metrics are computed via a having clause, using the same Brainy field-operator syntax as where (greaterThan, lessThan, between, anyOf/allOf/not, …). It applies to computed metric values and to count:

brain.queryAggregate('monthly_sales', {
  having: { revenue: { greaterThan: 10000 } }
})TYPESCRIPT

Filtering happens post-aggregation in Rust (O(groups)), so a having clause never re-scans entities.

State Serialization

The engine serializes all internal state (definitions, group states, the per-group value multiset, Welford's M2 values) to JSON for persistence:

// Brainy handles this automatically via the provider interface
const state = engine.serializeState()   // JSON string
engine.restoreState(state)              // Restore on restartTYPESCRIPT

State includes:

• All registered definitions

• Per-aggregate, per-group metric state

• The value multiset backing MIN/MAX, percentile, and distinctCount

• Welford's mean, M2, and count for stddev/variance

Source Filtering

Aggregate definitions can specify source filters to only aggregate entities of a specific type:

{
  "name": "event_stats",
  "source": { "type": "Event" },
  "groupBy": ["category"],
  "metrics": { "count": { "op": "count" } }
}JSON

During incremental updates, entities that don't match the source filter are skipped. During rebuild, the filter is compiled and applied before aggregation.

Aggregate entities themselves (entities with service: 'brainy:aggregation' or metadata.__aggregate: true) are always skipped to prevent infinite feedback loops.

source.where operators

source.where takes the same operators and the same field-addressing law as find()'s where — a set filter selects a set, not nothing:

{
  "name": "revenue",
  "source": { "type": "Event", "where": { "kind": { "oneOf": ["deposit", "fee"] } } },
  "groupBy": ["region"],
  "metrics": { "total": { "op": "sum", "field": "amount" } }
}JSON

Served operators: eq/equals/is, ne/notEquals/isNot, in/oneOf, noneOf, gt/gte/lt/lte (and their long spellings), between, contains, excludes, hasAll, exists, missing — plus the allOf / anyOf / not combinators. Multiple operators on one field are ANDed, an array field matches when any element matches, and ne, noneOf and excludes are satisfied by an entity that has no such field, all exactly as in find().

source.where and find()'s where name one served set. That is a tested law, not a convention: src/native/stringOperatorConformance.test.ts registers an aggregate for each set operator and asserts it counts exactly the rows find() returns for the same clause, so the two halves cannot drift apart again.

Field addressing follows the namespace law: system.<field> reads the record scalar, a bare or metadata.-prefixed name reads the user bag only, and a dotted name walks the nested path.

Any other operator is refused by name at defineAggregate time, with the offending operator in the message. An aggregate that silently sums nothing is the worst failure a number can have, so the engine either serves the filter or says which operator it will not serve — it never registers one it cannot evaluate.

That refusal set is exactly four operators — startsWith, endsWith, length and matches — and they are refused on the find() path for the same reason. See Filter operator conformance for why an index cannot answer them and what to write instead.

Incremental Update Flow

When Brainy calls incrementalUpdate():

1. Source filter check — skip if entity doesn't match

2. Aggregate entity check — skip if entity is itself an aggregate

3. Group key computation — extract groupBy fields, apply time bucketing

4. Metric update — for each metric in the definition:

   • add: Increment sum/count/mean/M2, insert into BTreeMap

   • delete: Decrement sum/count/mean/M2, remove from BTreeMap

   • update: Delete old values, add new values (handles group changes)

Performance

Run the included benchmarks: npm run bench

Troubleshooting

Aggregation not using native engine

Verify Brainy is loaded and the aggregation provider is registered:

const diag = brain.diagnostics()
console.log(diag.providers.aggregation)
// Should show { source: 'plugin' }TYPESCRIPT

Stale MIN/MAX after deletes

This should not happen with Brainy — the BTreeMap guarantees precision. If you see stale values, verify you're running Brainy (not the JS fallback) and that the delete operation includes the correct entity metadata.

Rebuild performance

For datasets over 100K entities, rebuild uses Rayon parallelism automatically. Ensure your system has multiple CPU cores available. Single-core environments still work but won't benefit from parallel rebuild.