sketch_kll.nx source

1// sketch_kll.nx -- compactor-hierarchy quantile sketch. 2// 3// MRL99-style (Manku-Rajagopalan-Lindsay 1999) "compactor cascade": 4// maintain levels 0, 1, 2, ... each holding up to k items. Items 5// at level h carry weight 2^h. When level h fills, sort it, flip 6// a coin, promote either even-indexed or odd-indexed items to 7// level h+1 (the other half discarded). 8// 9// Karnin-Lang-Liberty 2016 (FOCS) tightens this via geometric 10// per-level capacity decay; we use uniform capacity here for 11// simpler code at slightly looser bound. Rank-error declared 12// conservatively as ~1/sqrt(k) at conf 0.95. 13// 14// Versus Reservoir (sketch_reservoir.nx) HONEST COMPARISON 15// (2026-05-20 audit, sketch_kll_vs_reservoir_bench.nx): 16// - At MATCHED slot count, Reservoir wins on RANK-ERROR magnitude. 17// KLL k=64 declared eps=0.12 vs Reservoir cap=1024 declared 18// eps=0.0425 (both ~8KB). Empirically Reservoir wins on 19// uniform 1..10000 stream too. 20// - KLL's structural advantages live ELSEWHERE: 21// (1) DETERMINISTIC mergeable across instances (Reservoir 22// merge requires algorithmic choices that change semantics) 23// (2) O(1) amortized update at any N (Reservoir's update is 24// O(1/N) probability; quickly stops sampling at large N) 25// (3) Cascade preserves rank info from FULL stream history 26// (Reservoir's sample is a fixed random subset) 27// - Earlier doc claim of "tighter rank-error at fixed memory" was 28// a misframe (compared at same k, not same memory). Corrected. 29// 30// LOSSLESS-LANGUAGE DISCIPLINE (doc 20): 31// Same NX_ENV_RANK_ERROR envelope as Reservoir; substrate doesn't 32// double-count. Multiple primitives sharing an envelope kind is 33// fine -- the discriminator is about the SHAPE of the error, not 34// uniqueness. 35 36import "syscalls.nx" 37import "sketch_types.nx" 38 39const NX_KLL_K_MIN: i64 = 8 40const NX_KLL_K_MAX: i64 = 10000 41const NX_KLL_MAX_LEVELS: i64 = 16 42 43// LCG constants (same as Reservoir). 44const NX_KLL_LCG_A: i64 = 1103515245 45const NX_KLL_LCG_C: i64 = 12345 46const NX_KLL_LCG_MOD: i64 = 0x7FFFFFFF 47 48struct Kll { 49 k: i64, 50 n_levels: i64, // levels currently allocated 51 level_items: *i64, // contiguous: max_levels * k entries 52 level_count: *i64, // current item count per level 53 total_items: i64, 54 min_val: i64, 55 max_val: i64, 56 rng_state: i64, 57 view_vals: *i64, // bits-up: hoisted from nx_kll_quantile; 58 // sized max_levels * k once at alloc, avoids 59 // a per-query sys_mmap. 60 view_wts: *i64, 61} 62 63// === construction ================================================= 64 65func nx_kll_alloc(k: i64, seed: i64) -> *Kll { 66 if k < NX_KLL_K_MIN { return 0 as *Kll } 67 if k > NX_KLL_K_MAX { return 0 as *Kll } 68 let raw: *u8 = sys_mmap(80) 69 let s: *Kll = raw as *Kll 70 s.k = k 71 s.n_levels = 1 // level 0 always present 72 let items_bytes: i64 = NX_KLL_MAX_LEVELS * k * 8 73 let items_raw: *u8 = sys_mmap(items_bytes) 74 s.level_items = items_raw as *i64 75 let count_raw: *u8 = sys_mmap(NX_KLL_MAX_LEVELS * 8) 76 s.level_count = count_raw as *i64 77 let view_cap_bytes: i64 = NX_KLL_MAX_LEVELS * k * 8 78 s.view_vals = sys_mmap(view_cap_bytes) as *i64 79 s.view_wts = sys_mmap(view_cap_bytes) as *i64 80 var i: i64 = 0 81 while i < NX_KLL_MAX_LEVELS { 82 s.level_count[i] = 0 83 i = i + 1 84 } 85 s.total_items = 0 86 // Sentinels for min/max -- use first add to bootstrap. 87 s.min_val = 0x7FFFFFFFFFFFFFFF 88 s.max_val = -1 - 0x7FFFFFFFFFFFFFFF // i64 min 89 s.rng_state = seed | 1 90 return s 91} 92 93// LCG state mutation. 94func nx_kll_rng_next(s: *Kll) -> i64 { 95 let next: i64 = ((s.rng_state * NX_KLL_LCG_A) + NX_KLL_LCG_C) & NX_KLL_LCG_MOD 96 s.rng_state = next 97 return next 98} 99 100// === level-buffer access helpers ================================= 101// Buffer for level h starts at level_items[h * k]. 102 103func nx_kll_level_addr(s: *Kll, level: i64) -> *i64 { 104 return (s.level_items as i64 + level * s.k * 8) as *i64 105} 106 107func nx_kll_sort_level(s: *Kll, level: i64) -> i64 { 108 let count: i64 = s.level_count[level] 109 let buf: *i64 = nx_kll_level_addr(s, level) 110 var i: i64 = 1 111 while i < count { 112 let cur: i64 = buf[i] 113 var j: i64 = i - 1 114 var done: i64 = 0 115 while done == 0 { 116 if j < 0 { done = 1 } 117 if done == 0 { 118 let prev: i64 = buf[j] 119 if prev <= cur { done = 1 } 120 if done == 0 { 121 buf[j + 1] = prev 122 j = j - 1 123 } 124 } 125 } 126 buf[j + 1] = cur 127 i = i + 1 128 } 129 return 0 130} 131 132// === compact: sort + coin-flip + promote half ==================== 133 134func nx_kll_compact(s: *Kll, level: i64) -> i64 { 135 if level >= NX_KLL_MAX_LEVELS - 1 { 136 // No room to promote -- drop the level (rare; happens only 137 // at extreme stream sizes). 138 s.level_count[level] = 0 139 return 0 140 } 141 nx_kll_sort_level(s, level) 142 let count: i64 = s.level_count[level] 143 let buf: *i64 = nx_kll_level_addr(s, level) 144 // Coin flip: 0 = take even indices (0, 2, ...), 1 = take odd. 145 let pick_odd: i64 = nx_kll_rng_next(s) & 1 146 // Promote chosen half to level+1. 147 let next_level: i64 = level + 1 148 let next_buf: *i64 = nx_kll_level_addr(s, next_level) 149 let next_count_before: i64 = s.level_count[next_level] 150 var i: i64 = pick_odd // start at 0 or 1 151 var promoted: i64 = 0 152 while i < count { 153 let dst_idx: i64 = next_count_before + promoted 154 if dst_idx < s.k { 155 next_buf[dst_idx] = buf[i] 156 promoted = promoted + 1 157 } 158 i = i + 2 159 } 160 s.level_count[next_level] = next_count_before + promoted 161 s.level_count[level] = 0 162 // Grow n_levels if we just pushed into a new level. 163 if next_level >= s.n_levels { 164 s.n_levels = next_level + 1 165 } 166 // Cascade if next level is now full. 167 if s.level_count[next_level] >= s.k { 168 nx_kll_compact(s, next_level) 169 } 170 return 0 171} 172 173// === add ========================================================== 174 175func nx_kll_add(s: *Kll, value: i64) -> i64 { 176 if value < s.min_val { s.min_val = value } 177 if value > s.max_val { s.max_val = value } 178 s.total_items = s.total_items + 1 179 let buf: *i64 = nx_kll_level_addr(s, 0) 180 let count: i64 = s.level_count[0] 181 buf[count] = value 182 s.level_count[0] = count + 1 183 if s.level_count[0] >= s.k { 184 nx_kll_compact(s, 0) 185 } 186 return 0 187} 188 189// === total weight (for normalisation) ============================ 190 191func nx_kll_total_weight(s: *Kll) -> i64 { 192 var total: i64 = 0 193 var h: i64 = 0 194 while h < s.n_levels { 195 total = total + s.level_count[h] * (1 << h) 196 h = h + 1 197 } 198 return total 199} 200 201// === build sorted weighted view + query ========================== 202// 203// Concatenate every level's items with weight=2^h, sort by value, 204// scan accumulating cumulative weight, return value at target rank. 205// Sort is insertion-style over the combined view (O(N log N) at 206// query but N is small -- bounded by k * log(total_items)). 207 208func nx_kll_quantile(s: *Kll, p_milli: i64) -> i64 { 209 if s.total_items == 0 { return 0 } 210 let total_weight: i64 = nx_kll_total_weight(s) 211 if total_weight == 0 { return 0 } 212 // Materialise view: collect (value, weight) pairs. Cap entry 213 // count at max_levels * k. 214 // Bits-up: scratch hoisted to struct (was per-query sys_mmap). 215 let view_vals: *i64 = s.view_vals 216 let view_wts: *i64 = s.view_wts 217 var n: i64 = 0 218 var h: i64 = 0 219 while h < s.n_levels { 220 let buf: *i64 = nx_kll_level_addr(s, h) 221 let cnt: i64 = s.level_count[h] 222 let w: i64 = 1 << h 223 var i: i64 = 0 224 while i < cnt { 225 view_vals[n] = buf[i] 226 view_wts[n] = w 227 n = n + 1 228 i = i + 1 229 } 230 h = h + 1 231 } 232 // Sort the view by value (insertion sort; n is small). 233 var i: i64 = 1 234 while i < n { 235 let cur_v: i64 = view_vals[i] 236 let cur_w: i64 = view_wts[i] 237 var j: i64 = i - 1 238 var done: i64 = 0 239 while done == 0 { 240 if j < 0 { done = 1 } 241 if done == 0 { 242 if view_vals[j] <= cur_v { done = 1 } 243 if done == 0 { 244 view_vals[j + 1] = view_vals[j] 245 view_wts[j + 1] = view_wts[j] 246 j = j - 1 247 } 248 } 249 } 250 view_vals[j + 1] = cur_v 251 view_wts[j + 1] = cur_w 252 i = i + 1 253 } 254 // Walk sorted view, find target rank. 255 let target: i64 = (p_milli * total_weight) / 1000 256 var cum: i64 = 0 257 var k: i64 = 0 258 while k < n { 259 cum = cum + view_wts[k] 260 if cum >= target { return view_vals[k] } 261 k = k + 1 262 } 263 return view_vals[n - 1] 264} 265 266// === rank-of-value =============================================== 267 268func nx_kll_rank(s: *Kll, value: i64) -> i64 { 269 if s.total_items == 0 { return 0 } 270 if value < s.min_val { return 0 } 271 if value >= s.max_val { return 1000 } 272 let total_weight: i64 = nx_kll_total_weight(s) 273 if total_weight == 0 { return 0 } 274 // Sum weights of items with value <= query value. 275 var cum: i64 = 0 276 var h: i64 = 0 277 while h < s.n_levels { 278 let buf: *i64 = nx_kll_level_addr(s, h) 279 let cnt: i64 = s.level_count[h] 280 let w: i64 = 1 << h 281 var i: i64 = 0 282 while i < cnt { 283 if buf[i] <= value { 284 cum = cum + w 285 } 286 i = i + 1 287 } 288 h = h + 1 289 } 290 return (cum * 1000) / total_weight 291} 292 293// === error bound + query ========================================= 294// 295// MRL/KLL rank-error: ~1.36/sqrt(k) at 95% confidence (Hoeffding- 296// style conservative bound, same shape as Reservoir but k is the 297// per-level capacity, not the total reservoir size). 298 299func nx_kll_rank_error_ppb(k: i64) -> i64 { 300 if k <= 8 { return 480000000 } // 0.48 301 if k <= 32 { return 240000000 } 302 if k <= 128 { return 120000000 } // 0.12 303 if k <= 512 { return 60000000 } // 0.06 304 if k <= 2048 { return 30000000 } // 0.03 305 return 15000000 // 0.015 306} 307 308func nx_kll_query_quantile(s: *Kll, p_milli: i64) -> *ApproxI64 { 309 let v: i64 = nx_kll_quantile(s, p_milli) 310 return nx_approx_new(v, NX_ENV_RANK_ERROR, 311 nx_kll_rank_error_ppb(s.k), 312 950000000, 313 NX_MATURITY_REFERENCE_IMPL, 314 NX_ADV_HONEST) 315} 316 317// === memory introspection ======================================== 318 319func nx_kll_memory_bytes(s: *Kll) -> i64 { 320 return 64 + NX_KLL_MAX_LEVELS * s.k * 8 + NX_KLL_MAX_LEVELS * 8 321} 322 323func nx_kll_levels_used(s: *Kll) -> i64 { 324 return s.n_levels 325}