nx_shard_query.nx source
↩ module page · 46 lines · 2744 B
1// nx_shard_query.nx -- LIB: sharded SCATTER-GATHER query = the distributed READ PATH that makes the index scale
2// horizontally to petabytes. The index is partitioned into shards (by host/hash -- nx_shard); a query fans out to ALL
3// shards, each returns its LOCAL top-k, and the coordinator MERGES the per-shard top-k into the GLOBAL top-k. EXACT by
4// invariant: a doc in the global top-k must be in its own shard's top-k (a shard can't hide a global winner in its
5// local top-k), so merging (nshards*k) candidates and taking top-k is identical to a single whole-corpus index. Built +
6// gated NOW on a fixture; the network fan-out + parallel shard servers run on the NAS/cluster later. No float.
7// license_tier: ORIGINAL
8import "nx_syscalls.nx"
9
10// insert (d,s) into a DESCENDING-by-score top-k (arrays size k); filled[0] tracks count. returns 1 if inserted.
11func sq_insert(tkd: *i64, tks: *i64, k: i64, filled: *i64, d: i64, s: i64) -> i64 {
12 var pos: i64 = 0 - 1
13 if filled[0] < k { pos = filled[0]; filled[0] = filled[0] + 1 } else { if s > tks[k-1] { pos = k - 1 } }
14 if pos < 0 { return 0 }
15 tkd[pos] = d; tks[pos] = s
16 var go: i64 = 1
17 while go == 1 {
18 if pos <= 0 { go = 0 } else {
19 if tks[pos] > tks[pos-1] { let a: i64=tks[pos]; tks[pos]=tks[pos-1]; tks[pos-1]=a; let b: i64=tkd[pos]; tkd[pos]=tkd[pos-1]; tkd[pos-1]=b; pos = pos - 1 } else { go = 0 }
20 }
21 }
22 return 1
23}
24// top-k over one (docs,scores) list; fills tkd/tks[k]; returns filled count.
25func sq_topk(docs: *i64, scores: *i64, n: i64, k: i64, tkd: *i64, tks: *i64) -> i64 {
26 let filled: *i64 = sys_mmap(8) as *i64; filled[0] = 0
27 var i: i64 = 0
28 while i < n { sq_insert(tkd, tks, k, filled, docs[i], scores[i]); i = i + 1 }
29 return filled[0]
30}
31// SCATTER-GATHER: sd[shard]/ss[shard]=doc/score arrays, sn[shard]=count. Each shard -> local top-k (scatter);
32// merge all local top-k -> global top-k (gather). fills tkd/tks[k]; returns filled.
33func sq_scatter_gather(sd: *i64, ss: *i64, sn: *i64, nshards: i64, k: i64, tkd: *i64, tks: *i64) -> i64 {
34 let pd: *i64 = sys_mmap(8 * nshards * k) as *i64
35 let ps: *i64 = sys_mmap(8 * nshards * k) as *i64
36 var np: i64 = 0
37 var sh: i64 = 0
38 while sh < nshards {
39 let ltd: *i64 = sys_mmap(8*k) as *i64; let lts: *i64 = sys_mmap(8*k) as *i64
40 let lf: i64 = sq_topk(sd[sh] as *i64, ss[sh] as *i64, sn[sh], k, ltd, lts) // scatter: shard-local top-k
41 var j: i64 = 0
42 while j < lf { pd[np] = ltd[j]; ps[np] = lts[j]; np = np + 1; j = j + 1 } // gather: pool the candidates
43 sh = sh + 1
44 }
45 return sq_topk(pd, ps, np, k, tkd, tks) // merge: global top-k over the pool
46}