1	import Base
2	import ./Keys.bend as Keys
3	import ./MemTable.bend as MemTable
4	import ./BitTree.bend as BitTree
5	import ./SortedRun.bend as SortedRun
6	
7	# SSTables (Task 4): immutable strict sorted runs with BitTree Bloom filters.
8	# `build` canonicalizes newest-first raw entries; `from_sorted_unique` trusts an
9	# already strict, unique run and deliberately performs no sorting.
10	
11	# Represent Bloom data used by the SSTable indexes, filters, and lookup.
12	type Bloom is Data:
13	  Blm{bits: BitTree.BitTree, size: Nat}
14	
15	# Represent Table data used by the SSTable indexes, filters, and lookup.
16	type Table is Data:
17	  Tbl{
18	    entries: List<&2, MemTable.Entry>,
19	    filter: Bloom,
20	    nbits: Nat,
21	    smallest: Maybe<&2, String>,
22	    largest: Maybe<&2, String>,
23	    count: Nat,
24	    chunks: List<&2, List<&2, MemTable.Entry>>,
25	    blocks: List<&2, String>
26	  }
27	
28	# Handle the block entries for the SSTable indexes, filters, and lookup.
29	def block_entries() -> Nat:
30	  64n
31	
32	# Handle the chunk go for the SSTable indexes, filters, and lookup.
33	def chunk_go(
34	  fuel: Nat,
35	  +entries: List<&2, MemTable.Entry>,
36	  +cur: List<&2, MemTable.Entry>,
37	  room: Nat,
38	  acc: List<&2, List<&2, MemTable.Entry>>
39	) -> List<&2, List<&2, MemTable.Entry>>:
40	  match fuel:
41	    case 0n:
42	      List.reverse(&2, List<&2, MemTable.Entry>, Con{List.reverse(&2, MemTable.Entry, cur), acc})
43	    case 1n+f:
44	      match entries:
45	        case Nil{}:
46	          List.reverse(&2, List<&2, MemTable.Entry>, Con{List.reverse(&2, MemTable.Entry, cur), acc})
47	        case Con{e, t}:
48	          match room:
49	            case 0n:
50	              chunk_go(f, t, Con{e, Nil{}}, Nat.sub(block_entries(), 1n), Con{List.reverse(&2, MemTable.Entry, cur), acc})
51	            case 1n+r:
52	              chunk_go(f, t, Con{e, cur}, r, acc)
53	
54	# Handle chunk in the SSTable indexes, filters, and lookup.
55	def chunk(+entries: List<&2, MemTable.Entry>) -> List<&2, List<&2, MemTable.Entry>>:
56	  match entries:
57	    case Nil{}:
58	      Nil{}
59	    case Con{e, t}:
60	      chunk_go(List.length(&2, MemTable.Entry, entries), t, Con{e, Nil{}}, Nat.sub(block_entries(), 1n), Nil{})
61	
62	# Handle the chunk head key for the SSTable indexes, filters, and lookup.
63	def chunk_head_key(+xs: List<&2, MemTable.Entry>) -> Maybe<&2, String>:
64	  match xs:
65	    case Nil{}:
66	      None{}
67	    case Con{MemTable.Entry{key, val}, t}:
68	      Some{key}
69	
70	# Handle the block cons acc for the SSTable indexes, filters, and lookup.
71	def block_cons_acc(head: Maybe<&2, String>, acc: List<&2, String>) -> List<&2, String>:
72	  match head:
73	    case None{}:
74	      acc
75	    case Some{k}:
76	      Con{k, acc}
77	
78	# Handle the block keys for the SSTable indexes, filters, and lookup.
79	def block_keys(+chunks: List<&2, List<&2, MemTable.Entry>>, acc: List<&2, String>) -> List<&2, String>:
80	  match chunks:
81	    case Nil{}:
82	      List.reverse(&2, String, acc)
83	    case Con{c, t}:
84	      block_keys(t, block_cons_acc(chunk_head_key(c), acc))
85	
86	# Represent Metadata data used by the SSTable indexes, filters, and lookup.
87	type Metadata is Data:
88	  Meta{smallest: Maybe<&2, String>, largest: Maybe<&2, String>, count: Nat}
89	
90	# --- Bloom allocator (Monkey-style: smaller/upper levels get more bits) ---
91	# Closed form pinned here; Dayan-exact coefficients are tuning follow-up.
92	def bloom_bits(level: Nat, est_keys: Nat) -> Nat:
93	  match level:
94	    case 0n:
95	      Nat.mul(est_keys, 20n)
96	    case 1n+m:
97	      match m:
98	        case 0n:
99	          Nat.mul(est_keys, 10n)
100	        case 1n+p:
101	          Nat.mul(est_keys, 5n)
102	
103	# --- Local byte hash (kept stable for on-disk/rebuild compatibility) ---
104	def bhash(str: String, acc: U32) -> U32:
105	  match str:
106	    case SNil{}:
107	      acc
108	    case SCon{c, t}:
109	      bhash(t, U32.add(U32.mul(acc, 31), Char.to_u32(c)))
110	
111	# --- U32->Nat bridges that stay checker-tractable ---
112	# The Bend checker does not share across sequentially-chained U32.mul terms:
113	# `U32.to_nat` applied to a hash accumulated over 4+ bytes diverges in
114	# `{==}` elaboration (runtime is unaffected). The bridges below therefore
115	# never materialize a chained U32 as a Nat: each `to_nat` covers at most two
116	# U32 ops (shift+mask on the ORIGINAL word — parallel, shared), and all
117	# accumulation happens in small-Nat land. Both are EXACT (no semantic
118	# change): bytes are the base-256 digits of the word (Horner), so
119	# u32_to_nat_exact(w) == U32.to_nat(w) and bhash_n computes the same mod.
120	def u32_byte(+word: U32, shift: Nat) -> Nat:
121	  U32.to_nat(U32.and(U32.shrn(word, shift), 255))
122	
123	# Encode or decode a 32-bit word for to nat exact for the SSTable indexes, filters, and lookup.
124	def u32_to_nat_exact(+word: U32) -> Nat:
125	  +b0 = u32_byte(word, 24n)
126	  +b1 = u32_byte(word, 16n)
127	  +b2 = u32_byte(word, 8n)
128	  +b3 = u32_byte(word, 0n)
129	  Nat.add(Nat.mul(b0, Nat.mul(Nat.mul(256n, 256n), 256n)), Nat.add(Nat.mul(b1, Nat.mul(256n, 256n)), Nat.add(Nat.mul(b2, 256n), b3)))
130	
131	# Handle bhash n pos in the SSTable indexes, filters, and lookup.
132	def bhash_n_pos(+str: String, +seed: U32, +mod: Nat) -> Nat:
133	  +w = bhash(str, seed)
134	  +a0 = Nat.mod(u32_byte(w, 24n), mod)
135	  +a1 = Nat.mod(Nat.add(Nat.mul(a0, 256n), u32_byte(w, 16n)), mod)
136	  +a2 = Nat.mod(Nat.add(Nat.mul(a1, 256n), u32_byte(w, 8n)), mod)
137	  Nat.mod(Nat.add(Nat.mul(a2, 256n), u32_byte(w, 0n)), mod)
138	
139	# Handle bhash n in the SSTable indexes, filters, and lookup.
140	def bhash_n(str: String, seed: U32, mod: Nat) -> Nat:
141	  match mod:
142	    case 0n:
143	      0n
144	    case 1n+p:
145	      bhash_n_pos(str, seed, mod)
146	
147	# BitTree stores whole U32 words while Bloom.size preserves the exact logical
148	# bit count selected by bloom_bits.
149	def bit_words(num: Nat) -> Nat:
150	  match num:
151	    case 0n:
152	      0n
153	    case 1n+m:
154	      Nat.add(1n, Nat.div(m, 32n))
155	
156	# Handle bit data for new for the SSTable indexes, filters, and lookup.
157	def bits_new(+num: Nat) -> BitTree.BitTree:
158	  BitTree.new(bit_words(num))
159	
160	# Handle bit set in the SSTable indexes, filters, and lookup.
161	def bit_set(idx: Nat, bits: BitTree.BitTree) -> BitTree.BitTree:
162	  BitTree.set(bits, idx)
163	
164	# Handle bit get in the SSTable indexes, filters, and lookup.
165	def bit_get(idx: Nat, bits: BitTree.BitTree) -> Bool:
166	  BitTree.test(bits, idx)
167	
168	# Handle Bloom filter data for new for the SSTable indexes, filters, and lookup.
169	def bloom_new(+num: Nat) -> Bloom:
170	  Blm{bits_new(num), num}
171	
172	# Handle Bloom filter data for well formed for the SSTable indexes, filters, and lookup.
173	def bloom_well_formed(bloom: Bloom) -> Bool:
174	  match bloom:
175	    case Blm{+bits, +size}:
176	      Bool.and(
177	        BitTree.well_formed(bits),
178	        Bool.and(
179	          Nat.is_eq(BitTree.word_count(bits), bit_words(size)),
180	          Nat.is_le(size, BitTree.bit_count(bits))))
181	
182	# Handle Bloom filter data for add for the SSTable indexes, filters, and lookup.
183	def bloom_add(bloom: Bloom, +key: String) -> Bloom:
184	  match bloom:
185	    case Blm{bits, +size}:
186	      Blm{bit_set(bhash_n(key, 17, size), bit_set(bhash_n(key, 257, size), bits)), size}
187	
188	# Handle both in the SSTable indexes, filters, and lookup.
189	def both(lhs: Bool, rhs: Bool) -> Bool:
190	  match lhs:
191	    case True{}:
192	      rhs
193	    case False{}:
194	      False{}
195	
196	# Handle Bloom filter data for test for the SSTable indexes, filters, and lookup.
197	def bloom_test(bloom: Bloom, +key: String) -> Bool:
198	  match bloom:
199	    case Blm{+bits, +size}:
200	      both(bit_get(bhash_n(key, 17, size), bits), bit_get(bhash_n(key, 257, size), bits))
201	
202	# Return the table bits for the SSTable indexes, filters, and lookup.
203	def table_bits(tab: Table) -> Nat:
204	  match tab:
205	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
206	      nbits
207	
208	# Handle the optional value for present for the SSTable indexes, filters, and lookup.
209	def maybe_present(tab: Table, +key: String) -> Bool:
210	  match tab:
211	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
212	      bloom_test(filter, key)
213	
214	# --- Parallel probe counters (Task 1): sequential leaf, two-way fork, `!` gate ---
215	# Single probe as 0/1 for counting.
216	def probe_hit_if(present: Bool) -> Nat:
217	  match present:
218	    case True{}:
219	      1n
220	    case False{}:
221	      0n
222	
223	# Count matching probes sequentially.
224	def probe_hit(+tab: Table, +key: String) -> Nat:
225	  probe_hit_if(maybe_present(tab, key))
226	
227	# Sequential probe count over a probe list.
228	def probe_count_seq(+probes: List<&2, String>, +tab: Table) -> Nat:
229	  match probes:
230	    case Nil{}:
231	      0n
232	    case Con{+q, t}:
233	      Nat.add(probe_count_seq(t, tab), probe_hit(tab, q))
234	
235	# Two-way parallel probe count; halves share no state.
236	def probe_count_par(+probes: List<&2, String>, +tab: Table) -> Nat:
237	  +half = Nat.div(List.length(&2, String, probes), 2n)
238	  left right = probe_count_seq(List.take(&2, String, probes, half), tab) probe_count_seq(List.drop(&2, String, probes, half), tab)
239	  Nat.add(left, right)
240	
241	# Bang gate: GPU pays only with enough probes to fill lanes.
242	def probe_count_bang_gate(+num: Nat) -> Bool:
243	  Nat.is_le(16384n, num)
244	
245	# Pick CPU-pool or GPU execution for the same fork tree.
246	def probe_count_pick(gate: Bool, +probes: List<&2, String>, +tab: Table) -> Nat:
247	  match gate:
248	    case False{}:
249	      probe_count_par(probes, tab)
250	    case True{}:
251	      probe_count_par!(probes, tab)
252	
253	# Size-gated entry point for parallel probe counting.
254	def probe_count_auto(+probes: List<&2, String>, +tab: Table) -> Nat:
255	  probe_count_pick(probe_count_bang_gate(List.length(&2, String, probes)), probes, tab)
256	
257	# --- Canonicalization and construction ---
258	def bloom_of(entries: List<&2, MemTable.Entry>, +blm: Bloom) -> Bloom:
259	  match entries:
260	    case Nil{}:
261	      blm
262	    case Con{e, t}:
263	      match e:
264	        case MemTable.Entry{key, val}:
265	          bloom_of(t, bloom_add(blm, key))
266	
267	# Handle Bloom filter data for union trees for the SSTable indexes, filters, and lookup.
268	def bloom_union_trees(+abits: BitTree.BitTree, +bbits: BitTree.BitTree, size: Nat) -> Bloom:
269	  match abits:
270	    case BitTree.BT{aroot, awords, acap, aheight}:
271	      match bbits:
272	        case BitTree.BT{broot, bwords, bcap, bheight}:
273	          Blm{BitTree.BT{BitTree.union_bits(aroot, broot), awords, acap, aheight}, size}
274	
275	# Handle Bloom filter data for union for the SSTable indexes, filters, and lookup.
276	def bloom_union(ba: Bloom, bb: Bloom) -> Bloom:
277	  match ba:
278	    case Blm{abits, asize}:
279	      match bb:
280	        case Blm{bbits, bsize}:
281	          bloom_union_trees(abits, bbits, Nat.max(asize, bsize))
282	
283	# Handle Bloom filter data for of chunk for the SSTable indexes, filters, and lookup.
284	def bloom_of_chunk(+chunk: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
285	  bloom_of(chunk, bloom_new(nb))
286	
287	# Two-half parallel build (spike 1): one spawn, one union. Sequential below.
288	def bloom_of_par(+entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
289	  +half = Nat.div(List.length(&2, MemTable.Entry, entries), 2n)
290	  left right = bloom_of_chunk(List.take(&2, MemTable.Entry, entries, half), nb) bloom_of_chunk(List.drop(&2, MemTable.Entry, entries, half), nb)
291	  bloom_union(left, right)
292	
293	# --- Four-chunk parallel build (Task 2): two-level fork, three unions ---
294	# Third chunk slice as take-after-drop.
295	def chunk_third(+entries: List<&2, MemTable.Entry>, +skip: Nat, +take: Nat) -> List<&2, MemTable.Entry>:
296	  List.take(&2, MemTable.Entry, List.drop(&2, MemTable.Entry, entries, skip), take)
297	
298	# Four-way chunk build with a two-level fork tree.
299	def bloom_of_chunks(+entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
300	  +num = List.length(&2, MemTable.Entry, entries)
301	  +q = Nat.div(num, 4n)
302	  a b = bloom_of_chunk(List.take(&2, MemTable.Entry, entries, q), nb) bloom_of_chunk(chunk_third(entries, q, q), nb)
303	  c d = bloom_of_chunk(chunk_third(entries, Nat.mul(q, 2n), q), nb) bloom_of_chunk(List.drop(&2, MemTable.Entry, entries, Nat.mul(q, 3n)), nb)
304	  bloom_union(bloom_union(a, b), bloom_union(c, d))
305	
306	# Bang gate: GPU pays only with enough entries to feed lanes.
307	def bloom_bang_gate(+num: Nat) -> Bool:
308	  Nat.is_le(8192n, num)
309	
310	# Pick CPU-pool or GPU execution for the same chunk tree.
311	def bloom_of_pick(gate: Bool, +entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
312	  match gate:
313	    case False{}:
314	      bloom_of_chunks(entries, nb)
315	    case True{}:
316	      bloom_of_chunks!(entries, nb)
317	
318	# Size-gated `!` entry point: the bang fires only at >= 8192 entries, where
319	# every flush in the durable write path lands (frozen merge of 4096 + 4096).
320	def bloom_of_bang(+entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
321	  bloom_of_pick(bloom_bang_gate(List.length(&2, MemTable.Entry, entries)), entries, nb)
322	
323	# --- Hybrid bloom (spike): parallel index collection, sequential assembly ---
324	# Phase 1 emits a flat index list (two per key); joins concatenate, so no
325	# chunk tree or union is ever built. Phase 2 folds the indexes into ONE tree
326	# on the CPU. This removes the tree-allocation amplification that costs ~10%
327	# at 1M scale in the chunk+union design, while keeping a fork tree that a
328	# future `!` can ship once the GPU recovery crash is fixed upstream.
329	# Both bit positions of one key as a flat two-element list.
330	def key_idxs(+key: String, +size: Nat) -> List<&2, Nat>:
331	  Con{bhash_n(key, 17, size), Con{bhash_n(key, 257, size), Nil{}}}
332	
333	# Sequential index collection over entries.
334	def hash_idxs_seq(+entries: List<&2, MemTable.Entry>, +size: Nat) -> List<&2, Nat>:
335	  match entries:
336	    case Nil{}:
337	      Nil{}
338	    case Con{MemTable.Entry{key, val}, t}:
339	      List.append(&2, Nat, key_idxs(key, size), hash_idxs_seq(t, size))
340	
341	# Balanced parallel index collection; reusable fuel bounds the fork depth.
342	def hash_idxs_par_go(+fuel: Nat, +entries: List<&2, MemTable.Entry>, +size: Nat) -> List<&2, Nat>:
343	  match fuel:
344	    case 0n:
345	      hash_idxs_seq(entries, size)
346	    case 1n+f:
347	      match entries:
348	        case Nil{}:
349	          Nil{}
350	        case Con{MemTable.Entry{key, val}, Nil{}}:
351	          key_idxs(key, size)
352	        case Con{_, Con{_, _}}:
353	          +half = Nat.div(List.length(&2, MemTable.Entry, entries), 2n)
354	          left right = hash_idxs_par_go(f, List.take(&2, MemTable.Entry, entries, half), size) hash_idxs_par_go(f, List.drop(&2, MemTable.Entry, entries, half), size)
355	          List.append(&2, Nat, left, right)
356	
357	# Parallel entry point; fuel covers the fork depth.
358	def hash_idxs_par(+entries: List<&2, MemTable.Entry>, +size: Nat) -> List<&2, Nat>:
359	  +fuel = List.length(&2, MemTable.Entry, entries)
360	  hash_idxs_par_go(fuel, entries, size)
361	
362	# Sequential assembly of one filter from a flat index list.
363	def bloom_set_all(+idxs: List<&2, Nat>, +blm: Bloom) -> Bloom:
364	  match idxs:
365	    case Nil{}:
366	      blm
367	    case Con{+i, t}:
368	      match blm:
369	        case Blm{bits, +size}:
370	          bloom_set_all(t, Blm{bit_set(i, bits), size})
371	
372	# Hybrid build: parallel hashing, single-tree assembly.
373	def bloom_of_hybrid(+entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
374	  bloom_set_all(hash_idxs_par(entries, nb), bloom_new(nb))
375	
376	# Size-gated entry point: sequential on both sides of the gate.
377	# MEASURED VERDICT (2026-09-26, 1M interleaved A/B, Darwin arm64, Bend 2.0.28):
378	# sequential median 225105 ms (4442 ops/s) vs 4-chunk 249150 ms (4014 ops/s)
379	# vs hybrid 261746 ms — the parallel builds cost ~10% (fork/join plus
380	# allocation overhead dominates; invisible at 100k, compounds over 244
381	# flushes). Confirmed under two machine states (all arms shifted +15k ms on
382	# the slow day; relative order unchanged). The parallel builds stay as
383	# law-covered spikes; the `!` variants stay opt-in behind the GPU recovery
384	# crash documented below.
385	def bloom_pick(small: Bool, +entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
386	  match small:
387	    case True{}:
388	      bloom_of(entries, bloom_new(nb))
389	    case False{}:
390	      bloom_of(entries, bloom_new(nb))
391	
392	# Handle Bloom filter data for of auto for the SSTable indexes, filters, and lookup.
393	def bloom_of_auto(+entries: List<&2, MemTable.Entry>, +nb: Nat) -> Bloom:
394	  bloom_pick(Nat.is_lt(List.length(&2, MemTable.Entry, entries), 2048n), entries, nb)
395	
396	# Build table metadata for tail for the SSTable indexes, filters, and lookup.
397	def metadata_tail(entries: List<&2, MemTable.Entry>, +first: String, last: String, +count: Nat) -> Metadata:
398	  match entries:
399	    case Nil{}:
400	      Meta{Some{first}, Some{last}, count}
401	    case Con{MemTable.Entry{key, val}, rest}:
402	      metadata_tail(rest, first, key, Nat.add(count, 1n))
403	
404	# Handle metadata in the SSTable indexes, filters, and lookup.
405	def metadata(entries: List<&2, MemTable.Entry>) -> Metadata:
406	  match entries:
407	    case Nil{}:
408	      Meta{None{}, None{}, 0n}
409	    case Con{MemTable.Entry{+key, val}, rest}:
410	      metadata_tail(rest, key, key, 1n)
411	
412	# Handle from sorted unique est meta in the SSTable indexes, filters, and lookup.
413	def from_sorted_unique_est_meta(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat, meta: Metadata) -> Table:
414	  match meta:
415	    case Meta{smallest, largest, count}:
416	      +nb = bloom_bits(level, est_keys)
417	      Tbl{entries, bloom_of_auto(entries, nb), nb, smallest, largest, count, chunk(entries), block_keys(chunk(entries), Nil{})}
418	
419	# Handle from sorted unique meta in the SSTable indexes, filters, and lookup.
420	def from_sorted_unique_meta(+entries: List<&2, MemTable.Entry>, level: Nat, meta: Metadata) -> Table:
421	  match meta:
422	    case Meta{smallest, largest, +count}:
423	      +nb = bloom_bits(level, count)
424	      Tbl{entries, bloom_of_auto(entries, nb), nb, smallest, largest, count, chunk(entries), block_keys(chunk(entries), Nil{})}
425	
426	# Handle from sorted unique in the SSTable indexes, filters, and lookup.
427	def from_sorted_unique(+entries: List<&2, MemTable.Entry>, level: Nat) -> Table:
428	  from_sorted_unique_meta(entries, level, metadata(entries))
429	
430	# Build sorted for the SSTable indexes, filters, and lookup.
431	def build_sorted(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat) -> Table:
432	  from_sorted_unique_est_meta(entries, level, est_keys, metadata(entries))
433	
434	# Handle build in the SSTable indexes, filters, and lookup.
435	def build(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat) -> Table:
436	  +effective = Nat.max(est_keys, List.length(&2, MemTable.Entry, entries))
437	  build_sorted(SortedRun.sort_newest(entries), level, effective)
438	
439	# Metadata accessors keep clients independent of the Tbl field layout.
440	def table_smallest(tab: Table) -> Maybe<&2, String>:
441	  match tab:
442	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
443	      smallest
444	
445	# Return the table largest for the SSTable indexes, filters, and lookup.
446	def table_largest(tab: Table) -> Maybe<&2, String>:
447	  match tab:
448	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
449	      largest
450	
451	# Return the table count for the SSTable indexes, filters, and lookup.
452	def table_count(tab: Table) -> Nat:
453	  match tab:
454	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
455	      count
456	
457	# Total table identity for the read cache: smallest|largest|count.
458	def opt_name(+opt: Maybe<&2, String>) -> String:
459	  match opt:
460	    case None{}:
461	      "<none>"
462	    case Some{s}:
463	      s
464	
465	# Return the table id for the SSTable indexes, filters, and lookup.
466	def table_id(tab: Table) -> String:
467	  match tab:
468	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
469	      opt_name(smallest) ++ "|" ++ opt_name(largest) ++ "|" ++ Nat.show(count)
470	
471	# Open bridge: metadata is matched explicitly so the Bloom schedule projects
472	# through `build` without relying on reduction of an open entry list.
473	def table_bits_from_meta(
474	  +entries: List<&2, MemTable.Entry>,
475	  level: Nat,
476	  est_keys: Nat,
477	  meta: Metadata,
478	) -> {table_bits(from_sorted_unique_est_meta(entries, level, est_keys, meta)) == bloom_bits(level, est_keys) : Nat}:
479	  match meta:
480	    case Meta{smallest, largest, count}:
481	      {==}
482	
483	# Return the table bits build for the SSTable indexes, filters, and lookup.
484	def table_bits_build(
485	  +entries: List<&2, MemTable.Entry>,
486	  level: Nat,
487	  est_keys: Nat,
488	) -> {table_bits(build(entries, level, est_keys)) == bloom_bits(level, Nat.max(est_keys, List.length(&2, MemTable.Entry, entries))) : Nat}:
489	  +effective = Nat.max(est_keys, List.length(&2, MemTable.Entry, entries))
490	  table_bits_from_meta(SortedRun.sort_newest(entries), level, effective, metadata(SortedRun.sort_newest(entries)))
491	
492	
493	# Check key ranges for disjoint for the SSTable indexes, filters, and lookup.
494	def ranges_disjoint(+ta: Table, +tb: Table) -> Bool:
495	  SortedRun.range_disjoint_bounds(table_smallest(ta), table_largest(ta), table_smallest(tb), table_largest(tb))
496	
497	# Return the size of of in the SSTable indexes, filters, and lookup.
498	def size_of(bloom: Bloom) -> Nat:
499	  match bloom:
500	    case Blm{bits, size}:
501	      size
502	
503	# Bloom insertions only set tree bits and preserve the exact logical size.
504	def bloom_size_stable(
505	  entries: List<&2, MemTable.Entry>,
506	  bits: BitTree.BitTree,
507	  +size: Nat
508	) -> {size_of(bloom_of(entries, Blm{bits, size})) == size : Nat}:
509	  match entries:
510	    case Nil{}:
511	      {==}
512	    case Con{e, t}:
513	      match e:
514	        case MemTable.Entry{+key, val}:
515	          bloom_size_stable(t, bit_set(bhash_n(key, 17, size), bit_set(bhash_n(key, 257, size), bits)), size)
516	
517	# Block-scoped lookup. Checker constraints (proven by scratch experiments):
518	# multi-scrutinee matches follow declaration order, and only Nat answers flow
519	# through match arms — so selection returns an index, picked up by chunk_at_go.
520	def pick_ix(bt: List<&2, String>, ct: List<&2, List<&2, MemTable.Entry>>, ix: Nat, +key: String, order: Cmp) -> Nat:
521	  match bt ct order:
522	    case _ _ GT{}:
523	      Nat.sub(ix, 1n)
524	    case Nil{} _ _:
525	      ix
526	    case _ Nil{} _:
527	      ix
528	    case Con{b2, bt2} Con{d2, ct2} LT{}:
529	      pick_ix(bt2, ct2, Nat.add(ix, 1n), key, Keys.cmp(b2, key))
530	    case Con{b2, bt2} Con{d2, ct2} EQ{}:
531	      ix
532	
533	# Handle the chunk at go for the SSTable indexes, filters, and lookup.
534	def chunk_at_go(fuel: Nat, +chunks: List<&2, List<&2, MemTable.Entry>>, ix: Nat) -> List<&2, MemTable.Entry>:
535	  match fuel chunks:
536	    case 0n _:
537	      Nil{}
538	    case 1n+f Nil{}:
539	      Nil{}
540	    case 1n+f Con{c, t}:
541	      match ix:
542	        case 0n:
543	          c
544	        case 1n+p:
545	          chunk_at_go(f, t, p)
546	
547	# Handle the chunk at for the SSTable indexes, filters, and lookup.
548	def chunk_at(+chunks: List<&2, List<&2, MemTable.Entry>>, ix: Nat) -> List<&2, MemTable.Entry>:
549	  chunk_at_go(List.length(&2, List<&2, MemTable.Entry>, chunks), chunks, ix)
550	
551	# Handle the block scan for the SSTable indexes, filters, and lookup.
552	def block_scan(+chunk: List<&2, MemTable.Entry>, +key: String) -> Maybe<&2, String>:
553	  MemTable.get(MemTable.MT{chunk}, key)
554	
555	# Handle the block pick for the SSTable indexes, filters, and lookup.
556	def block_pick(
557	  +chunks: List<&2, List<&2, MemTable.Entry>>,
558	  +blocks: List<&2, String>,
559	  +key: String
560	) -> List<&2, MemTable.Entry>:
561	  match chunks blocks:
562	    case Nil{} _:
563	      Nil{}
564	    case _ Nil{}:
565	      Nil{}
566	    case Con{c, ct} Con{b, bt}:
567	      chunk_at(chunks, pick_ix(bt, ct, 0n, key, Keys.cmp(b, key)))
568	
569	# Handle the block finish for the SSTable indexes, filters, and lookup.
570	def block_finish(
571	  picked: List<&2, MemTable.Entry>,
572	  +entries: List<&2, MemTable.Entry>,
573	  +key: String
574	) -> Maybe<&2, String>:
575	  match picked:
576	    case Nil{}:
577	      MemTable.get(MemTable.MT{entries}, key)
578	    case Con{e, t}:
579	      block_scan(Con{e, t}, key)
580	
581	# Resolve the hit for finish for the SSTable indexes, filters, and lookup.
582	def hit_finish(
583	  picked: List<&2, MemTable.Entry>,
584	  +entries: List<&2, MemTable.Entry>,
585	  +key: String
586	) -> Maybe<&2, Maybe<&2, String>>:
587	  match picked:
588	    case Nil{}:
589	      MemTable.get_hit(MemTable.MT{entries}, key)
590	    case Con{e, t}:
591	      MemTable.get_hit(MemTable.MT{Con{e, t}}, key)
592	
593	# Handle the block get hit for the SSTable indexes, filters, and lookup.
594	def block_get_hit(tab: Table, +key: String) -> Maybe<&2, Maybe<&2, String>>:
595	  match tab:
596	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
597	      hit_finish(block_pick(chunks, blocks, key), entries, key)
598	
599	# Handle the block get for the SSTable indexes, filters, and lookup.
600	def block_get(tab: Table, +key: String) -> Maybe<&2, String>:
601	  match tab:
602	    case Tbl{entries, filter, nbits, smallest, largest, count, chunks, blocks}:
603	      block_finish(block_pick(chunks, blocks, key), entries, key)
604	
605	# Handle lookup in the SSTable indexes, filters, and lookup.
606	def lookup(tab: Table, +key: String) -> Maybe<&2, String>:
607	  block_get(tab, key)