From fe4f5df8c907a235f861cb66bcb13eb9c98704d9 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Tue, 26 May 2026 16:18:47 -0700 Subject: [PATCH] feat: exact percentile and distinctCount aggregation ops Add 'percentile' (with a 'p' fraction in [0,1]) and 'distinctCount' to the aggregation engine. Both are exact, computed from a per-metric value multiset (MetricState.valueCounts) maintained incrementally and delete-safe; percentile uses numpy-linear interpolation. The multiset is JSON-serializable so results survive persistence. 35 aggregation unit tests pass. --- src/aggregation/AggregationIndex.ts | 56 ++++++++++++++++++ src/types/brainy.types.ts | 23 +++++++- .../unit/aggregation/AggregationIndex.test.ts | 58 +++++++++++++++++++ 3 files changed, 135 insertions(+), 2 deletions(-) diff --git a/src/aggregation/AggregationIndex.ts b/src/aggregation/AggregationIndex.ts index 08bb74e0..36cc6645 100644 --- a/src/aggregation/AggregationIndex.ts +++ b/src/aggregation/AggregationIndex.ts @@ -198,6 +198,13 @@ function updateMetricAdd(state: MetricState, val: number, op: AggregationOp): vo const oldMean = state.count > 1 ? (state.sum - val) / (state.count - 1) : 0 state.m2 = (state.m2 ?? 0) + (val - oldMean) * (val - mean) } + + // Percentile/distinctCount: track the value multiset (matches Cortex's value_counts). + if (op === 'percentile' || op === 'distinctCount') { + if (!state.valueCounts) state.valueCounts = {} + const key = String(val) + state.valueCounts[key] = (state.valueCounts[key] ?? 0) + 1 + } } /** @@ -205,6 +212,16 @@ function updateMetricAdd(state: MetricState, val: number, op: AggregationOp): vo * Note: removing from Welford's is the inverse update. */ function updateMetricRemove(state: MetricState, val: number, op: AggregationOp): void { + // Percentile/distinctCount: decrement the value multiset (drop the key at zero). + if ((op === 'percentile' || op === 'distinctCount') && state.valueCounts) { + const key = String(val) + const c = state.valueCounts[key] + if (c !== undefined) { + if (c <= 1) delete state.valueCounts[key] + else state.valueCounts[key] = c - 1 + } + } + if (state.count <= 1) { state.sum = 0 state.count = 0 @@ -221,6 +238,39 @@ function updateMetricRemove(state: MetricState, val: number, op: AggregationOp): } } +/** + * Exact percentile over a value multiset, using linear interpolation between closest ranks + * (numpy 'linear' / type-7): `rank = p·(n−1)`, interpolating between the floor and ceil + * positions of the sorted expansion. Mirrors Cortex's `MetricState::percentile` bit-for-bit. + */ +function computePercentile(valueCounts: Record, count: number, p: number): number { + if (count === 0) return 0 + const pp = Math.max(0, Math.min(1, p)) + const entries = Object.entries(valueCounts) + .map(([v, c]) => [parseFloat(v), c] as [number, number]) + .sort((a, b) => a[0] - b[0]) + if (entries.length === 0) return 0 + if (count === 1) return entries[0][0] + + const rank = pp * (count - 1) + const lo = Math.floor(rank) + const hi = Math.ceil(rank) + const vLo = valueAtRank(entries, lo) + if (lo === hi) return vLo + const vHi = valueAtRank(entries, hi) + return vLo + (vHi - vLo) * (rank - lo) +} + +/** Value at 0-based rank `idx` in the sorted expansion of the multiset. */ +function valueAtRank(entries: Array<[number, number]>, idx: number): number { + let cum = 0 + for (const [v, c] of entries) { + cum += c + if (idx < cum) return v + } + return entries.length ? entries[entries.length - 1][0] : 0 +} + export class AggregationIndex { private storage: StorageAdapter private nativeProvider?: AggregationProvider @@ -595,6 +645,12 @@ export class AggregationIndex { case 'stddev': metrics[metricName] = state.count > 1 ? Math.sqrt((state.m2 ?? 0) / (state.count - 1)) : 0 break + case 'percentile': + metrics[metricName] = computePercentile(state.valueCounts ?? {}, state.count, metricDef.p ?? 0.5) + break + case 'distinctCount': + metrics[metricName] = state.valueCounts ? Object.keys(state.valueCounts).length : 0 + break } if (metricDef.op === 'count') { diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index 8451455c..11755c0e 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -696,7 +696,18 @@ export interface TraverseParams { /** * Supported aggregation operations */ -export type AggregationOp = 'sum' | 'count' | 'avg' | 'min' | 'max' | 'stddev' | 'variance' +export type AggregationOp = + | 'sum' + | 'count' + | 'avg' + | 'min' + | 'max' + | 'stddev' + | 'variance' + /** Exact percentile (requires `p` in `[0,1]` on the metric def) — value-multiset, delete-safe. */ + | 'percentile' + /** Exact count of distinct values — value-multiset, delete-safe. */ + | 'distinctCount' /** * Time window granularity for GROUP BY time dimensions @@ -752,8 +763,10 @@ export interface AggregateDefinition { export interface AggregateMetricDef { /** Aggregation operation */ op: AggregationOp - /** Metadata field to aggregate (required for sum/avg/min/max; optional for count) */ + /** Metadata field to aggregate (required for all ops except count) */ field?: string + /** Percentile fraction in `[0, 1]` — required when `op === 'percentile'`, ignored otherwise. */ + p?: number } /** @@ -767,6 +780,12 @@ export interface MetricState { max: number /** Running M2 for Welford's online variance (sum of squared differences from mean) */ m2?: number + /** + * Value multiset (String(value) → occurrence count) for exact percentile + distinctCount. + * Maintained only for `percentile`/`distinctCount` metrics. JSON-serializable; mirrors + * Cortex's `value_counts` so JS and native agree bit-for-bit. + */ + valueCounts?: Record } /** diff --git a/tests/unit/aggregation/AggregationIndex.test.ts b/tests/unit/aggregation/AggregationIndex.test.ts index 54c71d42..05f4a62a 100644 --- a/tests/unit/aggregation/AggregationIndex.test.ts +++ b/tests/unit/aggregation/AggregationIndex.test.ts @@ -617,4 +617,62 @@ describe('AggregationIndex', () => { } }) }) + + // ============= Percentile + distinctCount (exact, parity with Cortex native) ============= + + describe('percentile + distinctCount', () => { + beforeEach(() => { + index.defineAggregate({ + name: 'lat', + source: { type: NounType.Event }, + groupBy: ['svc'], + metrics: { + p0: { op: 'percentile', field: 'ms', p: 0 }, + p50: { op: 'percentile', field: 'ms', p: 0.5 }, + p90: { op: 'percentile', field: 'ms', p: 0.9 }, + p100: { op: 'percentile', field: 'ms', p: 1 }, + uniq: { op: 'distinctCount', field: 'ms' } + } + }) + }) + + const addLat = (id: string, ms: number) => + index.onEntityAdded(id, { type: NounType.Event, metadata: { svc: 'api', ms } }) + + it('computes exact numpy-linear percentiles (same values as Cortex)', () => { + for (let v = 1; v <= 10; v++) addLat(`e${v}`, v) + const [row] = index.queryAggregate({ name: 'lat' }) + expect(row.metrics.p0).toBe(1) + expect(row.metrics.p50).toBeCloseTo(5.5, 10) + expect(row.metrics.p90).toBeCloseTo(9.1, 10) + expect(row.metrics.p100).toBe(10) + expect(row.metrics.uniq).toBe(10) + }) + + it('stays exact after deletes', () => { + for (let v = 1; v <= 10; v++) addLat(`e${v}`, v) + index.onEntityDeleted('e10', { type: NounType.Event, metadata: { svc: 'api', ms: 10 } }) + const [row] = index.queryAggregate({ name: 'lat' }) + expect(row.metrics.p50).toBeCloseTo(5, 10) // median of 1..9 + expect(row.metrics.uniq).toBe(9) + }) + + it('distinctCount counts distinct values, not occurrences', () => { + addLat('a', 1); addLat('b', 2); addLat('c', 2) + const [row] = index.queryAggregate({ name: 'lat' }) + expect(row.metrics.uniq).toBe(2) // {1, 2} + }) + + it('percentile + distinctCount survive a persist + reload', async () => { + for (let v = 1; v <= 10; v++) addLat(`e${v}`, v) + await index.flush() + + const reloaded = new AggregationIndex(storage) + await reloaded.init() + const [row] = reloaded.queryAggregate({ name: 'lat' }) + expect(row.metrics.p50).toBeCloseTo(5.5, 10) // valueCounts round-tripped through storage + expect(row.metrics.uniq).toBe(10) + await reloaded.close() + }) + }) })