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
|
const oldMean = state.count > 1 ? (state.sum - val) / (state.count - 1) : 0
|
||||||
state.m2 = (state.m2 ?? 0) + (val - oldMean) * (val - mean)
|
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.
|
* Note: removing from Welford's is the inverse update.
|
||||||
*/
|
*/
|
||||||
function updateMetricRemove(state: MetricState, val: number, op: AggregationOp): void {
|
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) {
|
if (state.count <= 1) {
|
||||||
state.sum = 0
|
state.sum = 0
|
||||||
state.count = 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 {
|
export class AggregationIndex {
|
||||||
private storage: StorageAdapter
|
private storage: StorageAdapter
|
||||||
private nativeProvider?: AggregationProvider
|
private nativeProvider?: AggregationProvider
|
||||||
|
|
@ -595,6 +645,12 @@ export class AggregationIndex {
|
||||||
case 'stddev':
|
case 'stddev':
|
||||||
metrics[metricName] = state.count > 1 ? Math.sqrt((state.m2 ?? 0) / (state.count - 1)) : 0
|
metrics[metricName] = state.count > 1 ? Math.sqrt((state.m2 ?? 0) / (state.count - 1)) : 0
|
||||||
break
|
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') {
|
if (metricDef.op === 'count') {
|
||||||
|
|
|
||||||
|
|
@ -696,7 +696,18 @@ export interface TraverseParams {
|
||||||
/**
|
/**
|
||||||
* Supported aggregation operations
|
* 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
|
* Time window granularity for GROUP BY time dimensions
|
||||||
|
|
@ -752,8 +763,10 @@ export interface AggregateDefinition {
|
||||||
export interface AggregateMetricDef {
|
export interface AggregateMetricDef {
|
||||||
/** Aggregation operation */
|
/** Aggregation operation */
|
||||||
op: AggregationOp
|
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
|
field?: string
|
||||||
|
/** Percentile fraction in `[0, 1]` — required when `op === 'percentile'`, ignored otherwise. */
|
||||||
|
p?: number
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -767,6 +780,12 @@ export interface MetricState {
|
||||||
max: number
|
max: number
|
||||||
/** Running M2 for Welford's online variance (sum of squared differences from mean) */
|
/** Running M2 for Welford's online variance (sum of squared differences from mean) */
|
||||||
m2?: number
|
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