sketch_hllmap.nx source
↩ module page · 245 lines · 8404 B
1// sketch_hllmap.nx -- Map<key, HLL> for "distinct count per key" queries.
2//
3// E.g. distinct-IPs-per-URL, distinct-events-per-user, distinct-terms-
4// per-document. Compose a hash map with the HLL primitive: each map
5// entry holds a separate HLL sketch.
6//
7// CAPABILITY STOMP vs DataSketches:
8// DS HllMap supports add + estimate but does NOT support merge of
9// two HllMaps (their docs flag this as future work). We ship merge
10// on day one because:
11// - HLL merge is register-wise max (cheap)
12// - hash-map merge is set-union of keys (linear)
13// together they compose without state surgery.
14//
15// Hash table: linear-probing, power-of-2 capacity, open addressing.
16// Entry = {key_hash: i64, hll: *Hll}. Empty slots: key_hash == 0
17// (we map true-hash 0 -> 1 to avoid collision with the empty
18// sentinel).
19//
20// Per the lossless-language doctrine: per-key estimates ship the
21// same Approximate<i64> envelope as base HLL (rel_stddev = 1.04/sqrt(m)).
22
23import "syscalls.nx"
24import "murmur3.nx"
25import "sketch_hll.nx"
26import "sketch_types.nx"
27
28const NX_HMAP_EMPTY: i64 = 0
29const NX_HMAP_LF_MAX_PCT: i64 = 70 // resize at >70% load
30
31struct HllMapEntry {
32 key_hash: i64,
33 hll: *Hll,
34}
35
36struct HllMap {
37 entries: *HllMapEntry,
38 capacity: i64, // power of 2
39 n_entries: i64,
40 lg_k: i64,
41 seed: i64,
42}
43
44// === construction =================================================
45
46func nx_hmap_is_pow2(n: i64) -> i64 {
47 if n < 4 { return 0 }
48 if (n & (n - 1)) != 0 { return 0 }
49 return 1
50}
51
52func nx_hllmap_alloc(capacity: i64, lg_k: i64, seed: i64) -> *HllMap {
53 if nx_hmap_is_pow2(capacity) != 1 { return 0 as *HllMap }
54 if lg_k < 4 { return 0 as *HllMap }
55 if lg_k > 10 { return 0 as *HllMap }
56 let raw: *u8 = sys_mmap(48)
57 let m: *HllMap = raw as *HllMap
58 let ent_bytes: i64 = capacity * 16
59 let ent_raw: *u8 = sys_mmap(ent_bytes)
60 m.entries = ent_raw as *HllMapEntry
61 var i: i64 = 0
62 while i < ent_bytes {
63 ent_raw[i] = 0 // zero-init: key_hash=0 (empty), hll=NULL
64 i = i + 1
65 }
66 m.capacity = capacity
67 m.n_entries = 0
68 m.lg_k = lg_k
69 m.seed = seed
70 return m
71}
72
73// === entry access =================================================
74
75func nx_hmap_entry_at(m: *HllMap, i: i64) -> *HllMapEntry {
76 return (m.entries as i64 + i * 16) as *HllMapEntry
77}
78
79// === probe for a key's slot =======================================
80//
81// Returns the entry index (0..capacity-1) holding `key_hash`, or the
82// first empty slot we'd insert into. Always finds a target because
83// load factor capped at 70%.
84
85func nx_hllmap_probe(m: *HllMap, key_hash: i64) -> i64 {
86 let mask: i64 = m.capacity - 1
87 var i: i64 = key_hash & mask
88 var found_idx: i64 = -1
89 var done: i64 = 0
90 while done == 0 {
91 let e: *HllMapEntry = nx_hmap_entry_at(m, i)
92 if e.key_hash == NX_HMAP_EMPTY {
93 found_idx = i
94 done = 1
95 }
96 if done == 0 {
97 if e.key_hash == key_hash {
98 found_idx = i
99 done = 1
100 }
101 }
102 if done == 0 {
103 i = (i + 1) & mask
104 }
105 }
106 return found_idx
107}
108
109// === insert / lookup ==============================================
110//
111// nx_hllmap_add finds-or-creates the per-key HLL and adds the value
112// to it. Returns 0 on success, -1 if map is too full (no expansion
113// in v1 -- caller sized capacity appropriately).
114
115func nx_hllmap_key_hash(m: *HllMap, key: *u8, len: i64) -> i64 {
116 let h: i64 = murmur3_32(m.seed, key, len) & 0xFFFFFFFF
117 if h == NX_HMAP_EMPTY { return 1 }
118 return h
119}
120
121func nx_hllmap_add(m: *HllMap, key: *u8, key_len: i64,
122 value: *u8, value_len: i64) -> i64 {
123 // Refuse insert if at load limit AND no existing entry.
124 let key_hash: i64 = nx_hllmap_key_hash(m, key, key_len)
125 let idx: i64 = nx_hllmap_probe(m, key_hash)
126 let e: *HllMapEntry = nx_hmap_entry_at(m, idx)
127 if e.key_hash == NX_HMAP_EMPTY {
128 // New entry.
129 let load_after: i64 = ((m.n_entries + 1) * 100) / m.capacity
130 if load_after > NX_HMAP_LF_MAX_PCT { return -1 }
131 e.key_hash = key_hash
132 e.hll = nx_hll_alloc(m.lg_k, m.seed)
133 if e.hll == (0 as *Hll) { return -1 }
134 m.n_entries = m.n_entries + 1
135 }
136 nx_hll_add(e.hll, value, value_len)
137 return 0
138}
139
140func nx_hllmap_estimate(m: *HllMap, key: *u8, key_len: i64) -> i64 {
141 let key_hash: i64 = nx_hllmap_key_hash(m, key, key_len)
142 let idx: i64 = nx_hllmap_probe(m, key_hash)
143 let e: *HllMapEntry = nx_hmap_entry_at(m, idx)
144 if e.key_hash == NX_HMAP_EMPTY { return 0 }
145 return nx_hll_estimate(e.hll)
146}
147
148func nx_hllmap_query(m: *HllMap, key: *u8, key_len: i64) -> *ApproxI64 {
149 let key_hash: i64 = nx_hllmap_key_hash(m, key, key_len)
150 let idx: i64 = nx_hllmap_probe(m, key_hash)
151 let e: *HllMapEntry = nx_hmap_entry_at(m, idx)
152 if e.key_hash == NX_HMAP_EMPTY {
153 return nx_approx_new(0, NX_ENV_ABS, 0, 1000000000,
154 NX_MATURITY_REFERENCE_IMPL, NX_ADV_HONEST)
155 }
156 return nx_hll_query(e.hll)
157}
158
159// === merge ========================================================
160//
161// new_map = HllMapWithCapacity max(a.cap, b.cap)
162// for each entry in a: copy hll over (set entry, install pointer to copy)
163// for each entry in b: probe; if exists, merge into existing; else copy
164//
165// Requires a.lg_k == b.lg_k and a.seed == b.seed.
166
167func nx_hllmap_max2(x: i64, y: i64) -> i64 {
168 if x > y { return x }
169 return y
170}
171
172// Copy an HLL into a freshly-allocated one (for the case where we
173// own the result lifetime).
174func nx_hllmap_clone_hll(src: *Hll) -> *Hll {
175 let out: *Hll = nx_hll_alloc(src.lg_k, src.seed)
176 var i: i64 = 0
177 while i < src.m {
178 out.regs[i] = src.regs[i]
179 i = i + 1
180 }
181 return out
182}
183
184func nx_hllmap_merge(a: *HllMap, b: *HllMap) -> *HllMap {
185 if a.lg_k != b.lg_k { return 0 as *HllMap }
186 if a.seed != b.seed { return 0 as *HllMap }
187 let cap: i64 = nx_hllmap_max2(a.capacity, b.capacity)
188 let out: *HllMap = nx_hllmap_alloc(cap, a.lg_k, a.seed)
189 // Copy a's entries.
190 var i: i64 = 0
191 while i < a.capacity {
192 let ea: *HllMapEntry = nx_hmap_entry_at(a, i)
193 if ea.key_hash != NX_HMAP_EMPTY {
194 let idx_o: i64 = nx_hllmap_probe(out, ea.key_hash)
195 let eo: *HllMapEntry = nx_hmap_entry_at(out, idx_o)
196 eo.key_hash = ea.key_hash
197 eo.hll = nx_hllmap_clone_hll(ea.hll)
198 out.n_entries = out.n_entries + 1
199 }
200 i = i + 1
201 }
202 // Merge b's entries.
203 i = 0
204 while i < b.capacity {
205 let eb: *HllMapEntry = nx_hmap_entry_at(b, i)
206 if eb.key_hash != NX_HMAP_EMPTY {
207 let idx_o: i64 = nx_hllmap_probe(out, eb.key_hash)
208 let eo: *HllMapEntry = nx_hmap_entry_at(out, idx_o)
209 var just_installed: i64 = 0
210 if eo.key_hash == NX_HMAP_EMPTY {
211 // Key wasn't in a; install fresh clone of eb's HLL.
212 eo.key_hash = eb.key_hash
213 eo.hll = nx_hllmap_clone_hll(eb.hll)
214 out.n_entries = out.n_entries + 1
215 just_installed = 1
216 }
217 if just_installed == 0 {
218 // Pre-existing from a's pass -- merge eb's HLL into
219 // eo's HLL in-place via register-wise max.
220 var r: i64 = 0
221 while r < eo.hll.m {
222 let va: i64 = eo.hll.regs[r]
223 let vb: i64 = eb.hll.regs[r]
224 if vb > va { eo.hll.regs[r] = vb }
225 r = r + 1
226 }
227 }
228 }
229 i = i + 1
230 }
231 return out
232}
233
234// === introspection ================================================
235
236func nx_hllmap_memory_bytes(m: *HllMap) -> i64 {
237 // 48 (HllMap) + cap*16 (entries) + n_entries * (24 + m_bytes) (HLLs).
238 // m_bytes = 1 << lg_k for HLL_8.
239 let m_per_hll: i64 = 1 << m.lg_k
240 return 48 + m.capacity * 16 + m.n_entries * (24 + m_per_hll)
241}
242
243func nx_hllmap_n_entries(m: *HllMap) -> i64 {
244 return m.n_entries
245}