David Snelling
bf4a333f9b
The 3.0 native engine ships as @soulcraft/cor (brainy 8.x <-> cor 3.x are a version-matched pair); the old @soulcraft/cortex package stays on the 2.x/7.x line. Public docs and .d.ts-visible comments now name the correct package. Historical 2.x contract references (e.g. the 2.3.1 read-side fallback) keep the old name deliberately.
242 lines
10 KiB
Markdown
242 lines
10 KiB
Markdown
# Aggregation Architecture
|
|
|
|
> Write-time incremental aggregation with O(1) reads
|
|
|
|
## Design Principles
|
|
|
|
1. **Write-time computation** — aggregates update on every `add()`, `update()`, and `delete()`, not as batch jobs
|
|
2. **Incremental state** — running totals maintained per group, never rescanning the dataset
|
|
3. **Provider interface** — TypeScript engine is the default; plugins can replace it with native implementations
|
|
4. **Zero-allocation reads** — query results are computed from pre-aggregated state
|
|
|
|
## Component Overview
|
|
|
|
```
|
|
┌──────────────────────────────────────────────────────────┐
|
|
│ Brainy │
|
|
│ │
|
|
│ add() / update() / delete() │
|
|
│ │ │
|
|
│ ▼ │
|
|
│ ┌──────────────────────┐ ┌────────────────────────┐ │
|
|
│ │ AggregationIndex │ │ AggregateMaterializer │ │
|
|
│ │ │───▶│ (debounced writes) │ │
|
|
│ │ ├─ definitions Map │ └────────────────────────┘ │
|
|
│ │ ├─ states Map │ │
|
|
│ │ └─ staleMinMax Set │ ┌────────────────────────┐ │
|
|
│ │ │ │ timeWindows.ts │ │
|
|
│ │ Source filter ──────│───▶│ bucketTimestamp() │ │
|
|
│ │ Group key ──────────│───▶│ parseBucketRange() │ │
|
|
│ └──────────┬───────────┘ └────────────────────────┘ │
|
|
│ │ │
|
|
│ │ provider interface │
|
|
│ ▼ │
|
|
│ ┌──────────────────────┐ │
|
|
│ │ AggregationProvider │ (optional, registered by │
|
|
│ │ ├─ incrementalUpdate│ plugin like @soulcraft/cor) │
|
|
│ │ ├─ rebuildAggregate │ │
|
|
│ │ ├─ queryAggregate │ │
|
|
│ │ └─ serialize/restore│ │
|
|
│ └──────────────────────┘ │
|
|
└──────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
## State Management
|
|
|
|
### Definitions
|
|
|
|
Registered via `brain.defineAggregate(def)`. Stored in a `Map<string, AggregateDefinition>` keyed by aggregate name. Persisted to storage under `__aggregation_definitions__` on flush.
|
|
|
|
### Group State
|
|
|
|
Each aggregate maintains a `Map<string, AggregateGroupState>` where keys are serialized group key values (e.g., `category=food|date=2024-01`). Each group holds per-metric `MetricState`:
|
|
|
|
```typescript
|
|
interface MetricState {
|
|
sum: number // Running total
|
|
count: number // Entity count
|
|
min: number // Minimum (Infinity if empty)
|
|
max: number // Maximum (-Infinity if empty)
|
|
m2?: number // Welford's M2 for stddev/variance
|
|
}
|
|
```
|
|
|
|
### Change Detection
|
|
|
|
On restart, definition hashes (FNV-1a 32-bit) are compared with the persisted hash. If a definition changed (different groupBy, metrics, or source), the aggregate state is reset and must be rebuilt.
|
|
|
|
## Write-Time Update Flow
|
|
|
|
When `brain.add(entity)` is called:
|
|
|
|
```
|
|
1. For each registered aggregate:
|
|
├─ Source filter check (type, service, where)
|
|
│ └─ Skip if entity doesn't match
|
|
├─ Aggregate entity check
|
|
│ └─ Skip if entity.service === 'brainy:aggregation'
|
|
│ or entity.metadata.__aggregate is set
|
|
├─ Group key computation
|
|
│ └─ Extract groupBy fields from metadata
|
|
│ Apply time bucketing for windowed dimensions
|
|
└─ Metric update
|
|
└─ For each metric in the definition:
|
|
├─ count: increment count
|
|
├─ sum/avg: add value to sum, increment count
|
|
├─ min/max: compare and update
|
|
└─ stddev/variance: Welford's online update
|
|
```
|
|
|
|
### Update Handling
|
|
|
|
On `brain.update(entity)`, the engine reverses the old entity's contribution and applies the new entity's contribution. This correctly handles:
|
|
|
|
- **Value changes**: old amount=10, new amount=20 — sum adjusts by +10
|
|
- **Group key changes**: entity moves from category "food" to "drink" — both groups update
|
|
- **Source filter changes**: entity type changes from Event to Document — removed from matching aggregates
|
|
|
|
### Delete Handling
|
|
|
|
On `brain.remove(id)`, the engine reverses the entity's contribution:
|
|
|
|
- `count` and `sum` are decremented
|
|
- `min`/`max` may become stale (marked in `staleMinMax` for lazy recompute)
|
|
- Welford's M2 is updated with the inverse formula
|
|
- Empty groups (all metric counts at zero) are removed
|
|
|
|
## Algorithms
|
|
|
|
### Welford's Online Algorithm
|
|
|
|
Standard deviation and variance use Welford's numerically stable online algorithm with M2 tracking. This computes incrementally without storing individual values:
|
|
|
|
```
|
|
On add(x):
|
|
count += 1
|
|
oldMean = (sum - x) / (count - 1) // mean before this value
|
|
sum += x
|
|
mean = sum / count // mean after this value
|
|
M2 += (x - oldMean) * (x - mean)
|
|
|
|
On remove(x):
|
|
oldMean = sum / count
|
|
sum -= x
|
|
count -= 1
|
|
newMean = sum / count
|
|
M2 = max(0, M2 - (x - oldMean) * (x - newMean))
|
|
|
|
Sample variance = M2 / (count - 1)
|
|
Sample stddev = sqrt(variance)
|
|
```
|
|
|
|
M2 is clamped to zero on remove to prevent floating-point drift from producing negative values.
|
|
|
|
### MIN/MAX Handling
|
|
|
|
The TypeScript engine uses simple comparison for add operations and marks MIN/MAX as potentially stale on delete (since removing the current min/max value requires a rescan). Stale values are lazily recomputed on the next query.
|
|
|
|
The Cor native engine uses a `BTreeMap<OrderedFloat<f64>, u64>` that tracks the exact frequency of every value, providing precise MIN/MAX after any sequence of operations without rescanning.
|
|
|
|
### Time Window Bucketing
|
|
|
|
Timestamps (Unix milliseconds) are bucketed using UTC-based formatting:
|
|
|
|
| Granularity | Bucket Key | Algorithm |
|
|
|------------|-----------|-----------|
|
|
| `hour` | `2024-01-15T14` | UTC year-month-day-hour |
|
|
| `day` | `2024-01-15` | UTC year-month-day |
|
|
| `week` | `2024-W03` | ISO 8601 week (Monday start, week 1 contains first Thursday) |
|
|
| `month` | `2024-01` | UTC year-month |
|
|
| `quarter` | `2024-Q1` | `ceil((month) / 3)` |
|
|
| `year` | `2024` | UTC year |
|
|
| `{ seconds: N }` | ISO timestamp | `floor(timestamp / interval) * interval` |
|
|
|
|
Bucket keys can be parsed back into `{ start, end }` timestamp ranges via `parseBucketRange()`.
|
|
|
|
## Provider Interface
|
|
|
|
The `AggregationProvider` interface defines the contract between Brainy's `AggregationIndex` and plugin-provided native implementations:
|
|
|
|
```typescript
|
|
interface AggregationProvider {
|
|
defineAggregate?(def: AggregateDefinition): void
|
|
removeAggregate?(name: string): void
|
|
|
|
incrementalUpdate(
|
|
name: string,
|
|
def: AggregateDefinition,
|
|
entity: Record<string, unknown>,
|
|
op: 'add' | 'update' | 'delete',
|
|
prev?: Record<string, unknown>
|
|
): AggregateGroupState[]
|
|
|
|
computeGroupKey(
|
|
entity: Record<string, unknown>,
|
|
groupBy: GroupByDimension[]
|
|
): Record<string, string | number>
|
|
|
|
rebuildAggregate(
|
|
def: AggregateDefinition,
|
|
entities: Array<Record<string, unknown>>
|
|
): Map<string, AggregateGroupState>
|
|
|
|
queryAggregate(
|
|
state: Map<string, AggregateGroupState>,
|
|
params: AggregateQueryParams
|
|
): AggregateResult[]
|
|
|
|
restoreState?(data: string): void
|
|
serializeState?(): string
|
|
}
|
|
```
|
|
|
|
When a native provider is registered:
|
|
|
|
1. `AggregationIndex` delegates `incrementalUpdate()` to the provider instead of running TypeScript logic
|
|
2. Provider returns updated `AggregateGroupState[]` which are applied back into the state maps
|
|
3. Query execution is delegated via `queryAggregate()`
|
|
4. State serialization is delegated via `serializeState()`/`restoreState()`
|
|
|
|
Brainy retains ownership of the state maps and persistence. The provider handles computation.
|
|
|
|
## Materialization
|
|
|
|
The `AggregateMaterializer` converts aggregate group states into `NounType.Measurement` entities:
|
|
|
|
1. When an aggregate group is updated and `materialize` is enabled, `scheduleMaterialize()` is called
|
|
2. Materialization is debounced (default: 1000ms) to batch rapid updates during ingestion
|
|
3. On trigger, the materializer either creates or updates a `NounType.Measurement` entity
|
|
4. Materialized entities include `service: 'brainy:aggregation'` and `metadata.__aggregate` to prevent infinite loops
|
|
|
|
Materialized entities are automatically visible through:
|
|
- OData endpoints
|
|
- Google Sheets integration
|
|
- Server-Sent Events (SSE)
|
|
- Webhook notifications
|
|
|
|
## Persistence
|
|
|
|
### Storage Keys
|
|
|
|
| Key | Content |
|
|
|-----|---------|
|
|
| `__aggregation_definitions__` | Array of all definitions with FNV-1a hashes |
|
|
| `__aggregation_state_{name}__` | Per-aggregate group states (array of `AggregateGroupState`) |
|
|
| `__aggregation_native_state__` | Serialized native provider state (JSON string) |
|
|
|
|
### Lifecycle
|
|
|
|
1. **`init()`** — Load definitions, compare hashes, load matching state, restore native provider state
|
|
2. **Write operations** — Mark modified aggregates as dirty
|
|
3. **`flush()`** — Persist all dirty aggregate states and native provider state
|
|
4. **`close()`** — Flush and release resources
|
|
|
|
## Source Files
|
|
|
|
| File | Purpose |
|
|
|------|---------|
|
|
| `src/aggregation/AggregationIndex.ts` | Core engine: definitions, state, write hooks, query |
|
|
| `src/aggregation/materializer.ts` | Debounced materialization of results as entities |
|
|
| `src/aggregation/timeWindows.ts` | Time bucketing and bucket range parsing |
|
|
| `src/aggregation/index.ts` | Module exports |
|
|
| `src/types/brainy.types.ts` | Type definitions for all aggregation interfaces |
|