sketch_space_saving.nx source
↩ module page · 220 lines · 7186 B
1// sketch_space_saving.nx -- SpaceSaving top-K heavy hitters.
2//
3// Metwally-Agrawal-El-Abbadi 2005. Bounded set of (key, count,
4// error) triples of capacity K. On add: if key tracked, increment;
5// else if room, insert with count=1, error=0; else evict the
6// minimum-count entry and install (new_key, min_count + count,
7// min_count).
8//
9// Guarantee: any item with true frequency > N/K is tracked (no
10// false negatives for heavy hitters). The reported count
11// over-estimates by at most `error`; true_count is in
12// [count - error, count] for tracked items.
13//
14// COMPLEMENTS CMS:
15// - CMS: per-query frequency for ANY key (overestimate-only).
16// - SpaceSaving: enumerate the top-K most frequent keys directly.
17// Together they cover the full frequency-counting axis.
18//
19// API takes i64 keys (caller hashes strings to i64 via murmur3 if
20// they have string-keyed streams).
21//
22// LOSSLESS-LANGUAGE DISCIPLINE (doc 20):
23// nx_ss_query returns ApproxI64 with envelope_kind = NX_ENV_ABS;
24// param_a holds the per-key error bound (count - error <= true <=
25// count); conf_ppb = 1e9 (deterministic; not probabilistic).
26
27import "syscalls.nx"
28import "sketch_types.nx"
29
30const NX_SS_K_MIN: i64 = 2
31const NX_SS_K_MAX: i64 = 100000
32
33// Counter triple: 24 bytes per entry.
34struct SsCounter {
35 key: i64,
36 count: i64,
37 error: i64,
38}
39
40struct SpaceSaving {
41 k: i64,
42 n_tracked: i64, // <= k
43 counters: *SsCounter,
44 total_count: i64,
45}
46
47// === construction =================================================
48
49func nx_ss_alloc(k: i64) -> *SpaceSaving {
50 if k < NX_SS_K_MIN { return 0 as *SpaceSaving }
51 if k > NX_SS_K_MAX { return 0 as *SpaceSaving }
52 let raw: *u8 = sys_mmap(40)
53 let s: *SpaceSaving = raw as *SpaceSaving
54 s.k = k
55 s.n_tracked = 0
56 let table_bytes: i64 = k * 24
57 let table_raw: *u8 = sys_mmap(table_bytes)
58 s.counters = table_raw as *SsCounter
59 s.total_count = 0
60 return s
61}
62
63// === find / insert helpers ========================================
64
65func nx_ss_find_key(s: *SpaceSaving, key: i64) -> i64 {
66 // Returns index of counter holding `key`, or -1.
67 var i: i64 = 0
68 while i < s.n_tracked {
69 let c: *SsCounter = (s.counters as i64 + i * 24) as *SsCounter
70 if c.key == key { return i }
71 i = i + 1
72 }
73 return -1
74}
75
76func nx_ss_find_min(s: *SpaceSaving) -> i64 {
77 // Returns index of counter with smallest count (any tie broken
78 // by lowest index).
79 if s.n_tracked == 0 { return -1 }
80 var min_idx: i64 = 0
81 let c0: *SsCounter = s.counters
82 var min_count: i64 = c0.count
83 var i: i64 = 1
84 while i < s.n_tracked {
85 let c: *SsCounter = (s.counters as i64 + i * 24) as *SsCounter
86 if c.count < min_count {
87 min_count = c.count
88 min_idx = i
89 }
90 i = i + 1
91 }
92 return min_idx
93}
94
95// === add ==========================================================
96
97func nx_ss_add(s: *SpaceSaving, key: i64, count: i64) -> i64 {
98 if count <= 0 { return 0 }
99 s.total_count = s.total_count + count
100 let pos: i64 = nx_ss_find_key(s, key)
101 if pos >= 0 {
102 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
103 c.count = c.count + count
104 return 0
105 }
106 if s.n_tracked < s.k {
107 let slot: *SsCounter = (s.counters as i64 + s.n_tracked * 24) as *SsCounter
108 slot.key = key
109 slot.count = count
110 slot.error = 0
111 s.n_tracked = s.n_tracked + 1
112 return 0
113 }
114 // Evict minimum.
115 let min_idx: i64 = nx_ss_find_min(s)
116 let m: *SsCounter = (s.counters as i64 + min_idx * 24) as *SsCounter
117 let evicted_count: i64 = m.count
118 m.key = key
119 m.count = evicted_count + count
120 m.error = evicted_count
121 return 0
122}
123
124// === query ========================================================
125
126func nx_ss_estimate(s: *SpaceSaving, key: i64) -> i64 {
127 let pos: i64 = nx_ss_find_key(s, key)
128 if pos < 0 { return 0 }
129 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
130 return c.count
131}
132
133func nx_ss_lower_bound(s: *SpaceSaving, key: i64) -> i64 {
134 let pos: i64 = nx_ss_find_key(s, key)
135 if pos < 0 { return 0 }
136 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
137 return c.count - c.error
138}
139
140func nx_ss_error(s: *SpaceSaving, key: i64) -> i64 {
141 let pos: i64 = nx_ss_find_key(s, key)
142 if pos < 0 { return 0 }
143 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
144 return c.error
145}
146
147// Maximum possible over-count: per Metwally et al. the error of
148// any tracked entry is bounded by ceil(N / k).
149func nx_ss_max_overcount(s: *SpaceSaving) -> i64 {
150 return (s.total_count + s.k - 1) / s.k
151}
152
153func nx_ss_query(s: *SpaceSaving, key: i64) -> *ApproxI64 {
154 let est: i64 = nx_ss_estimate(s, key)
155 return nx_approx_new(est, NX_ENV_ABS, nx_ss_max_overcount(s),
156 1000000000,
157 NX_MATURITY_REFERENCE_IMPL,
158 NX_ADV_HONEST)
159}
160
161// === top-K enumeration ===========================================
162//
163// Return the top-N keys by count (descending). Caller pre-allocates
164// the (keys, counts) arrays of size n. Uses simple sort-then-copy
165// (k is small).
166
167func if_min(a: i64, b: i64) -> i64 {
168 if a < b { return a }
169 return b
170}
171
172func nx_ss_top_k(s: *SpaceSaving, n: i64, out_keys: *i64, out_counts: *i64) -> i64 {
173 // Build an index array, sort by count descending, materialize.
174 let limit: i64 = if_min(n, s.n_tracked)
175 // Insertion sort over counter indices by count descending.
176 let idx_raw: *u8 = sys_mmap(s.n_tracked * 8)
177 let idx: *i64 = idx_raw as *i64
178 var i: i64 = 0
179 while i < s.n_tracked { idx[i] = i; i = i + 1 }
180 // Sort idx[0..n_tracked) by counters[idx[*]].count descending.
181 i = 1
182 while i < s.n_tracked {
183 let cur: i64 = idx[i]
184 let cur_ptr: *SsCounter = (s.counters as i64 + cur * 24) as *SsCounter
185 let cur_count: i64 = cur_ptr.count
186 var j: i64 = i - 1
187 var done: i64 = 0
188 while done == 0 {
189 if j < 0 { done = 1 }
190 if done == 0 {
191 let prev_ptr: *SsCounter = (s.counters as i64 + idx[j] * 24) as *SsCounter
192 let prev_count: i64 = prev_ptr.count
193 if prev_count >= cur_count {
194 done = 1
195 }
196 if done == 0 {
197 idx[j + 1] = idx[j]
198 j = j - 1
199 }
200 }
201 }
202 idx[j + 1] = cur
203 i = i + 1
204 }
205 i = 0
206 while i < limit {
207 let ci: i64 = idx[i]
208 let c: *SsCounter = (s.counters as i64 + ci * 24) as *SsCounter
209 out_keys[i] = c.key
210 out_counts[i] = c.count
211 i = i + 1
212 }
213 return limit
214}
215
216// === memory introspection =========================================
217
218func nx_ss_memory_bytes(s: *SpaceSaving) -> i64 {
219 return 40 + s.k * 24
220}