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

@@ -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') {

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<string, number>
 }
 
 /**

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()
+    })
+  })
 })