nx_distexec_gate.nx source
↩ module page · 130 lines · 5661 B
1// nx_distexec_gate.nx -- the correctness a distributed scheduler stands on: the merged result is IDENTICAL
2// for any partition count (partition-invariance), for sum/count/min/max AND group-by-sum, including a
3// partition count that does NOT divide the row count (ragged partitions). If any of these failed, running
4// partitions on different workers would silently return a wrong answer. Hand-computed known answers + a
5// cross-P invariance sweep on a larger frame. D001 verdict via nx_gate_verdict. expect_exit: 0
6import "nx_gate_verdict.nx"
7import "nx_distexec.nx"
8
9func dg_eq(a: i64, b: i64) -> i64 { if a == b { return 1 } return 0 }
10func dg_frame1(a: *i64, nm: *u8, n: i64) -> *u8 {
11 let cols: *i64 = sys_mmap(8) as *i64
12 let names: *i64 = sys_mmap(8) as *i64
13 cols[0] = a as i64
14 names[0] = nm as i64
15 let tot: i64 = cf_encoded_bytes(names, 1, n)
16 let f: *u8 = sys_mmap(tot + 64)
17 cf_encode(cols, names, 1, n, f)
18 return f
19}
20func dg_frame2(a: *i64, b: *i64, na: *u8, nb: *u8, n: i64) -> *u8 {
21 let cols: *i64 = sys_mmap(8 * 2) as *i64
22 let names: *i64 = sys_mmap(8 * 2) as *i64
23 cols[0] = a as i64
24 cols[1] = b as i64
25 names[0] = na as i64
26 names[1] = nb as i64
27 let tot: i64 = cf_encoded_bytes(names, 2, n)
28 let f: *u8 = sys_mmap(tot + 64)
29 cf_encode(cols, names, 2, n, f)
30 return f
31}
32
33func main() -> i64 {
34 let ctr: *i64 = gv_ctr()
35 gv_head("nx_distexec_gate -- is the partition->merge algebra correct and PARTITION-INVARIANT?" as *u8)
36
37 // ---- tiny hand-computed frame: val = [10,20,30,40,50,60,70] (n=7, prime -> ragged partitions) ----
38 let val: *i64 = sys_mmap(8 * 7) as *i64
39 var i: i64 = 0
40 while i < 7 { val[i] = (i + 1) * 10; i = i + 1 }
41 let f: *u8 = dg_frame1(val, "val" as *u8, 7)
42 // sum=280 count=7 min=10 max=70
43 let a: *i64 = sys_mmap(8 * 4) as *i64
44 de_run_agg(f, 0, 1, a)
45 var t1: i64 = 1
46 if a[DE_SUM] != 280 { t1 = 0 }
47 if a[DE_CNT] != 7 { t1 = 0 }
48 if a[DE_MIN] != 10 { t1 = 0 }
49 if a[DE_MAX] != 70 { t1 = 0 }
50 gv_check("T1 single-partition agg is the hand-computed sum280/cnt7/min10/max70" as *u8, t1, ctr)
51
52 // PARTITION-INVARIANCE: P=1,2,3,4,7 must ALL give the identical merged result on n=7 (ragged)
53 let b: *i64 = sys_mmap(8 * 4) as *i64
54 var invok: i64 = 1
55 var P: i64 = 2
56 while P <= 7 {
57 de_run_agg(f, 0, P, b)
58 if b[DE_SUM] != a[DE_SUM] { invok = 0 }
59 if b[DE_CNT] != a[DE_CNT] { invok = 0 }
60 if b[DE_MIN] != a[DE_MIN] { invok = 0 }
61 if b[DE_MAX] != a[DE_MAX] { invok = 0 }
62 P = P + 1
63 }
64 gv_check("T2 agg is PARTITION-INVARIANT for P=2..7 over 7 rows (ragged partitions, no lost/double rows)" as *u8, invok, ctr)
65
66 // P GREATER than n: extra partitions are empty and must not corrupt the merge (count stays 7)
67 de_run_agg(f, 0, 16, b)
68 var t3: i64 = 1
69 if b[DE_CNT] != 7 { t3 = 0 }
70 if b[DE_SUM] != 280 { t3 = 0 }
71 gv_check("T3 more partitions than rows -> empty partitions merge cleanly (still sum280/cnt7)" as *u8, t3, ctr)
72
73 // ---- GROUP BY key SUM, partitioned == monolithic ---------------------------------------------
74 // key = [0,1,0,1,2,2,0], val=[1,2,3,4,5,6,7]; group sums: k0=1+3+7=11, k1=2+4=6, k2=5+6=11
75 let gk: *i64 = sys_mmap(8 * 7) as *i64
76 let gv: *i64 = sys_mmap(8 * 7) as *i64
77 gk[0] = 0
78 gk[1] = 1
79 gk[2] = 0
80 gk[3] = 1
81 gk[4] = 2
82 gk[5] = 2
83 gk[6] = 0
84 i = 0
85 while i < 7 { gv[i] = i + 1; i = i + 1 }
86 let gf: *u8 = dg_frame2(gv, gk, "val" as *u8, "key" as *u8, 7)
87 let bk1: *i64 = sys_mmap(8 * 8) as *i64
88 de_run_groupby(gf, 1, 0, 8, 1, bk1)
89 var t4: i64 = 1
90 if bk1[0] != 11 { t4 = 0 }
91 if bk1[1] != 6 { t4 = 0 }
92 if bk1[2] != 11 { t4 = 0 }
93 gv_check("T4 GROUP BY key SUM single-partition = hand-computed k0=11,k1=6,k2=11" as *u8, t4, ctr)
94 // partitioned group-by must match monolithic for P=2,3,4,7
95 let bkP: *i64 = sys_mmap(8 * 8) as *i64
96 var gbinv: i64 = 1
97 P = 2
98 while P <= 7 {
99 de_run_groupby(gf, 1, 0, 8, P, bkP)
100 var k: i64 = 0
101 while k < 3 { if bkP[k] != bk1[k] { gbinv = 0 } k = k + 1 }
102 P = P + 1
103 }
104 gv_check("T5 GROUP BY is partition-invariant for P=2..7 (the shuffle-merge is correct)" as *u8, gbinv, ctr)
105
106 // ---- invariance on a LARGER random frame across the vector boundary ---------------------------
107 let N: i64 = 20000
108 let rv: *i64 = sys_mmap(8 * N) as *i64
109 var s: i64 = 31
110 i = 0
111 while i < N { s = (s * 1103515245 + 12345) & 0x7fffffff; rv[i] = (s % 100000) - 50000; i = i + 1 }
112 let rf: *u8 = dg_frame1(rv, "v" as *u8, N)
113 let m1: *i64 = sys_mmap(8 * 4) as *i64
114 let m2: *i64 = sys_mmap(8 * 4) as *i64
115 de_run_agg(rf, 0, 1, m1)
116 de_run_agg(rf, 0, 13, m2) // 13 does not divide 20000 -> ragged
117 var t6: i64 = 1
118 if m1[DE_SUM] != m2[DE_SUM] { t6 = 0 }
119 if m1[DE_MIN] != m2[DE_MIN] { t6 = 0 }
120 if m1[DE_MAX] != m2[DE_MAX] { t6 = 0 }
121 if m1[DE_CNT] != m2[DE_CNT] { t6 = 0 }
122 gv_check("T7 20000 rows, P=13 (ragged) == P=1 for sum/min/max/count -- invariance holds at scale" as *u8, t6, ctr)
123 // count over all partitions equals n exactly (no row lost or double-counted -- the correctness a
124 // scheduler depends on to place partitions on separate workers)
125 gv_check("T8 every row is covered EXACTLY once (count == n) under partitioning" as *u8, dg_eq(m2[DE_CNT], N), ctr)
126
127 let rc: i64 = gv_verdict("DISTEXEC-GATE" as *u8, ctr, "partition->map->merge is correct and partition-invariant for sum/count/min/max and group-by -- the algebra a distributed scheduler runs" as *u8)
128 sys_exit(rc)
129 return rc
130}