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.
This commit is contained in:
parent
bca3736f4c
commit
fe4f5df8c9
3 changed files with 135 additions and 2 deletions
|
|
@ -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<string, number>, 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') {
|
||||
|
|
|
|||
|
|
@ -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<string, number>
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue