1	import Base
2	import ./Keys.bend as Keys
3	import ./MemTable.bend as MemTable
4	
5	# Pure reference semantics and stable sorted-run merging. This module stays
6	# independent of Sstable so table construction can depend on it later.
7	
8	def bool_not(rhs: Bool) -> Bool:
9	  match rhs:
10	    case False{}:
11	      True{}
12	    case True{}:
13	      False{}
14	
15	def bool_and(lhs: Bool, rhs: Bool) -> Bool:
16	  match lhs:
17	    case False{}:
18	      False{}
19	    case True{}:
20	      rhs
21	
22	def bool_or(lhs: Bool, rhs: Bool) -> Bool:
23	  match lhs:
24	    case False{}:
25	      rhs
26	    case True{}:
27	      True{}
28	
29	def entry_cmp(e1: MemTable.Entry, e2: MemTable.Entry) -> Cmp:
30	  match e1:
31	    case MemTable.Entry{k1, v1}:
32	      match e2:
33	        case MemTable.Entry{k2, v2}:
34	          Keys.cmp(k1, k2)
35	
36	def entry_key_eq(entry: MemTable.Entry, +key: String) -> Bool:
37	  match entry:
38	    case MemTable.Entry{k, v}:
39	      Keys.eq(k, key)
40	
41	# --- Reference predicates and observable model ---
42	
43	def contains_key(+entry: MemTable.Entry, entries: List<&2, MemTable.Entry>) -> Bool:
44	  match entry:
45	    case MemTable.Entry{+key, val}:
46	      match entries:
47	        case Nil{}:
48	          False{}
49	        case Con{h, t}:
50	          bool_or(entry_key_eq(h, key), contains_key(MemTable.Entry{key, val}, t))
51	
52	def is_unique(entries: List<&2, MemTable.Entry>) -> Bool:
53	  match entries:
54	    case Nil{}:
55	      True{}
56	    case Con{+h, +t}:
57	      bool_and(bool_not(contains_key(h, t)), is_unique(t))
58	
59	def is_strict(entries: List<&2, MemTable.Entry>) -> Bool:
60	  match entries:
61	    case Nil{}:
62	      True{}
63	    case Con{h, Nil{}}:
64	      True{}
65	    case Con{+h, Con{+next, +rest}}:
66	      bool_and(Keys.cmp_is_lt(entry_cmp(h, next)), is_strict(Con{next, rest}))
67	
68	def resolve_step(done: Bool, hit: Bool, val: Maybe<&2, String>, ans: Maybe<&2, Maybe<&2, String>>) -> Bool & Maybe<&2, Maybe<&2, String>>:
69	  match done:
70	    case True{}:
71	      (True{}, ans)
72	    case False{}:
73	      match hit:
74	        case True{}:
75	          (True{}, Some{val})
76	        case False{}:
77	          (False{}, ans)
78	
79	def resolve_state(entries: List<&2, MemTable.Entry>, +key: String, state: Bool & Maybe<&2, Maybe<&2, String>>) -> Bool & Maybe<&2, Maybe<&2, String>>:
80	  match entries:
81	    case Nil{}:
82	      state
83	    case Con{MemTable.Entry{k, val}, t}:
84	      (done, ans) = state
85	      resolve_state(t, key, resolve_step(done, Keys.eq(k, key), val, ans))
86	
87	def resolve_answer(state: Bool & Maybe<&2, Maybe<&2, String>>) -> Maybe<&2, Maybe<&2, String>>:
88	  match state:
89	    case (done, ans):
90	      ans
91	
92	def resolve(entries: List<&2, MemTable.Entry>, key: String) -> Maybe<&2, Maybe<&2, String>>:
93	  resolve_answer(resolve_state(entries, key, (False{}, None{})))
94	
95	def resolve_newest_go(runs: List<&2, List<&2, MemTable.Entry>>, +key: String, state: Bool & Maybe<&2, Maybe<&2, String>>) -> Bool & Maybe<&2, Maybe<&2, String>>:
96	  match runs:
97	    case Nil{}:
98	      state
99	    case Con{run, rest}:
100	      resolve_newest_go(rest, key, resolve_state(run, key, state))
101	
102	def resolve_newest(runs: List<&2, List<&2, MemTable.Entry>>, key: String) -> Maybe<&2, Maybe<&2, String>>:
103	  resolve_answer(resolve_newest_go(runs, key, (False{}, None{})))
104	
105	
106	def runs_well_formed(runs: List<&2, List<&2, MemTable.Entry>>) -> Bool:
107	  match runs:
108	    case Nil{}:
109	      True{}
110	    case Con{+run, rest}:
111	      bool_and(bool_and(is_strict(run), is_unique(run)), runs_well_formed(rest))
112	
113	# L1 helpers operate on entry runs to avoid a SortedRun <-> Sstable cycle.
114	def first_key(entries: List<&2, MemTable.Entry>) -> Maybe<&2, String>:
115	  match entries:
116	    case Nil{}:
117	      None{}
118	    case Con{MemTable.Entry{key, val}, t}:
119	      Some{key}
120	
121	def last_key_go(entries: List<&2, MemTable.Entry>, last: String) -> Maybe<&2, String>:
122	  match entries:
123	    case Nil{}:
124	      Some{last}
125	    case Con{MemTable.Entry{key, val}, t}:
126	      last_key_go(t, key)
127	
128	def last_key(entries: List<&2, MemTable.Entry>) -> Maybe<&2, String>:
129	  match entries:
130	    case Nil{}:
131	      None{}
132	    case Con{MemTable.Entry{key, val}, t}:
133	      last_key_go(t, key)
134	
135	def range_disjoint_bounds(
136	  af: Maybe<&2, String>,
137	  al: Maybe<&2, String>,
138	  bf: Maybe<&2, String>,
139	  bl: Maybe<&2, String>
140	) -> Bool:
141	  match af al bf bl:
142	    case None{} _ _ _:
143	      True{}
144	    case _ None{} _ _:
145	      True{}
146	    case _ _ None{} _:
147	      True{}
148	    case _ _ _ None{}:
149	      True{}
150	    case Some{a_first} Some{a_last} Some{b_first} Some{b_last}:
151	      bool_or(Keys.lt(a_last, b_first), Keys.lt(b_last, a_first))
152	
153	def range_disjoint(+xs: List<&2, MemTable.Entry>, +ys: List<&2, MemTable.Entry>) -> Bool:
154	  range_disjoint_bounds(first_key(xs), last_key(xs), first_key(ys), last_key(ys))
155	
156	def disjoint_with(+run: List<&2, MemTable.Entry>, runs: List<&2, List<&2, MemTable.Entry>>) -> Bool:
157	  match runs:
158	    case Nil{}:
159	      True{}
160	    case Con{+h, t}:
161	      bool_and(range_disjoint(run, h), disjoint_with(run, t))
162	
163	def pairwise_disjoint(runs: List<&2, List<&2, MemTable.Entry>>) -> Bool:
164	  match runs:
165	    case Nil{}:
166	      True{}
167	    case Con{+h, +t}:
168	      bool_and(disjoint_with(h, t), pairwise_disjoint(t))
169	
170	def level_disjoint(+runs: List<&2, List<&2, MemTable.Entry>>) -> Bool:
171	  bool_and(runs_well_formed(runs), pairwise_disjoint(runs))
172	
173	def levels_well_formed(levels: List<&2, List<&2, List<&2, MemTable.Entry>>>) -> Bool:
174	  match levels:
175	    case Nil{}:
176	      True{}
177	    case Con{+level, rest}:
178	      bool_and(level_disjoint(level), levels_well_formed(rest))
179	
180	# Expressible prerequisite for lower-shadow reasoning: every supplied level
181	# strictly below target is itself well formed and pairwise range-disjoint.
182	def lower_levels_well_formed(levels: List<&2, List<&2, List<&2, MemTable.Entry>>>, target: Nat) -> Bool:
183	  match levels target:
184	    case Nil{} _:
185	      True{}
186	    case Con{level, below} 0n:
187	      levels_well_formed(below)
188	    case Con{level, rest} 1n+n:
189	      lower_levels_well_formed(rest, n)
190	
191	# --- Fuel-bounded stable two-run linear merge ---
192	
193	def merge_step(
194	  order: Cmp,
195	  +newer_head: MemTable.Entry,
196	  newer_tail: List<&2, MemTable.Entry>,
197	  +older_head: MemTable.Entry,
198	  older_tail: List<&2, MemTable.Entry>,
199	  acc: List<&2, MemTable.Entry>
200	) -> List<&2, MemTable.Entry> & List<&2, MemTable.Entry> & List<&2, MemTable.Entry>:
201	  match order:
202	    case LT{}:
203	      (Con{newer_head, acc}, newer_tail, Con{older_head, older_tail})
204	    case EQ{}:
205	      (Con{newer_head, acc}, newer_tail, older_tail)
206	    case GT{}:
207	      (Con{older_head, acc}, Con{newer_head, newer_tail}, older_tail)
208	
209	def merge_go(
210	  fuel: Nat,
211	  state: List<&2, MemTable.Entry> & List<&2, MemTable.Entry> & List<&2, MemTable.Entry>
212	) -> List<&2, MemTable.Entry>:
213	  match fuel:
214	    case 0n:
215	      (acc, newer, older) = state
216	      List.reverse.go(&2, MemTable.Entry, acc, List.append(&2, MemTable.Entry, newer, older))
217	    case 1n+f:
218	      (acc, newer, older) = state
219	      match newer older:
220	        case Nil{} Nil{}:
221	          List.reverse(&2, MemTable.Entry, acc)
222	        case Nil{} rest:
223	          List.reverse.go(&2, MemTable.Entry, acc, rest)
224	        case rest Nil{}:
225	          List.reverse.go(&2, MemTable.Entry, acc, rest)
226	        case Con{+nh, nt} Con{+oh, ot}:
227	          merge_go(f, merge_step(entry_cmp(nh, oh), nh, nt, oh, ot, acc))
228	
229	def merge_newer(+newer: List<&2, MemTable.Entry>, +older: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:
230	  fuel = Nat.add(Nat.add(List.length(&2, MemTable.Entry, newer), List.length(&2, MemTable.Entry, older)), 1n)
231	  merge_go(fuel, (Nil{}, newer, older))
232	
233	# --- Precedence-preserving balanced multi-run merge ---
234	
235	type RunGroup is Data:
236	  RunGroup{
237	    newest_rank: Nat,
238	    oldest_rank: Nat,
239	    entries: List<&2, MemTable.Entry>
240	  }
241	
242	def rank_runs(runs: List<&2, List<&2, MemTable.Entry>>, +rank: Nat) -> List<&2, RunGroup>:
243	  match runs:
244	    case Nil{}:
245	      Nil{}
246	    case Con{entries, rest}:
247	      Con{RunGroup{rank, rank, entries}, rank_runs(rest, 1n+rank)}
248	
249	def merge_groups(newer: RunGroup, older: RunGroup) -> RunGroup:
250	  match newer:
251	    case RunGroup{newest_rank, oldest_rank, newer_entries}:
252	      match older:
253	        case RunGroup{older_newest, older_oldest, older_entries}:
254	          RunGroup{newest_rank, older_oldest, merge_newer(newer_entries, older_entries)}
255	
256	def merge_round(groups: List<&2, RunGroup>) -> List<&2, RunGroup>:
257	  match groups:
258	    case Nil{}:
259	      Nil{}
260	    case Con{group, Nil{}}:
261	      Con{group, Nil{}}
262	    case Con{newer, Con{older, rest}}:
263	      merged tail = merge_groups(newer, older) merge_round(rest)
264	      Con{merged, tail}
265	
266	def group_entries(groups: List<&2, RunGroup>) -> List<&2, MemTable.Entry>:
267	  match groups:
268	    case Nil{}:
269	      Nil{}
270	    case Con{RunGroup{newest_rank, oldest_rank, entries}, rest}:
271	      entries
272	
273	def merge_many_go(fuel: Nat, groups: List<&2, RunGroup>) -> List<&2, MemTable.Entry>:
274	  match fuel:
275	    case 0n:
276	      group_entries(groups)
277	    case 1n+f:
278	      merge_many_go(f, merge_round(groups))
279	
280	def merge_many_newest(+runs: List<&2, List<&2, MemTable.Entry>>) -> List<&2, MemTable.Entry>:
281	  fuel = Nat.add(List.length(&2, List<&2, MemTable.Entry>, runs), 1n)
282	  merge_many_go(fuel, rank_runs(runs, 0n))
283	
284	# Canonicalize raw newest-first entries in O(N log N): each entry starts as one
285	# already-sorted run, and stable EQ selection preserves the earlier/newer entry.
286	def singleton_runs(entries: List<&2, MemTable.Entry>) -> List<&2, List<&2, MemTable.Entry>>:
287	  match entries:
288	    case Nil{}:
289	      Nil{}
290	    case Con{entry, rest}:
291	      Con{Con{entry, Nil{}}, singleton_runs(rest)}
292	
293	def sort_newest(entries: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:
294	  merge_many_newest(singleton_runs(entries))
295	
296	def merge_disjoint_dec(ok: Bool, runs: List<&2, List<&2, MemTable.Entry>>) -> Maybe<&2, List<&2, MemTable.Entry>>:
297	  match ok:
298	    case False{}:
299	      None{}
300	    case True{}:
301	      Some{merge_many_newest(runs)}
302	
303	# L1 has no age tie-breaker. Reject malformed overlapping/duplicate ranges;
304	# for accepted disjoint runs, balanced merge order is observationally irrelevant.
305	def merge_many_disjoint(+runs: List<&2, List<&2, MemTable.Entry>>) -> Maybe<&2, List<&2, MemTable.Entry>>:
306	  merge_disjoint_dec(level_disjoint(runs), runs)
307	
308	# --- Pure bounded output partitioning ---
309	
310	# A zero target cannot bound a non-empty partition, so the public partitioning
311	# API and its bound predicate consistently interpret it as the smallest safe
312	# target: one entry per chunk.
313	def partition_target(target: Nat) -> Nat:
314	  match target:
315	    case 0n:
316	      1n
317	    case 1n+n:
318	      1n+n
319	
320	def partition_finish(
321	  current_rev: List<&2, MemTable.Entry>,
322	  chunks_rev: List<&2, List<&2, MemTable.Entry>>
323	) -> List<&2, List<&2, MemTable.Entry>>:
324	  match current_rev:
325	    case Nil{}:
326	      List.reverse(&2, List<&2, MemTable.Entry>, chunks_rev)
327	    case Con{entry, rest}:
328	      chunk = List.reverse(&2, MemTable.Entry, Con{entry, rest})
329	      List.reverse(&2, List<&2, MemTable.Entry>, Con{chunk, chunks_rev})
330	
331	type PartitionState is Data:
332	  PartitionState{
333	    room: Nat,
334	    entries: List<&2, MemTable.Entry>,
335	    current_rev: List<&2, MemTable.Entry>,
336	    chunks_rev: List<&2, List<&2, MemTable.Entry>>
337	  }
338	
339	def partition_step(
340	  full: Bool,
341	  target: Nat,
342	  room_after: Nat,
343	  entry: MemTable.Entry,
344	  rest: List<&2, MemTable.Entry>,
345	  current_rev: List<&2, MemTable.Entry>,
346	  chunks_rev: List<&2, List<&2, MemTable.Entry>>
347	) -> PartitionState:
348	  match full:
349	    case False{}:
350	      PartitionState{room_after, rest, Con{entry, current_rev}, chunks_rev}
351	    case True{}:
352	      chunk = List.reverse(&2, MemTable.Entry, Con{entry, current_rev})
353	      PartitionState{target, rest, Nil{}, Con{chunk, chunks_rev}}
354	
355	def partition_go(
356	  fuel: Nat,
357	  +target: Nat,
358	  state: PartitionState
359	) -> List<&2, List<&2, MemTable.Entry>>:
360	  match fuel state:
361	    case 0n PartitionState{room, entries, current_rev, chunks_rev}:
362	      partition_finish(current_rev, chunks_rev)
363	    case 1n+f PartitionState{0n, entries, current_rev, chunks_rev}:
364	      partition_finish(current_rev, chunks_rev)
365	    case 1n+f PartitionState{1n+r, Nil{}, current_rev, chunks_rev}:
366	      partition_finish(current_rev, chunks_rev)
367	    case 1n+f PartitionState{1n++r, Con{entry, rest}, current_rev, chunks_rev}:
368	      partition_go(f, target, partition_step(Nat.is_eq(r, 0n), target, r, entry, rest, current_rev, chunks_rev))
369	
370	def partition(+target: Nat, +entries: List<&2, MemTable.Entry>) -> List<&2, List<&2, MemTable.Entry>>:
371	  +size = partition_target(target)
372	  partition_go(List.length(&2, MemTable.Entry, entries), size, PartitionState{size, entries, Nil{}, Nil{}})
373	
374	def concat_chunks(chunks: List<&2, List<&2, MemTable.Entry>>) -> List<&2, MemTable.Entry>:
375	  match chunks:
376	    case Nil{}:
377	      Nil{}
378	    case Con{chunk, rest}:
379	      List.append(&2, MemTable.Entry, chunk, concat_chunks(rest))
380	
381	def chunks_bounded_go(+target: Nat, chunks: List<&2, List<&2, MemTable.Entry>>) -> Bool:
382	  match chunks:
383	    case Nil{}:
384	      True{}
385	    case Con{chunk, rest}:
386	      bool_and(
387	        Nat.is_le(List.length(&2, MemTable.Entry, chunk), target),
388	        chunks_bounded_go(target, rest))
389	
390	def chunks_bounded(target: Nat, chunks: List<&2, List<&2, MemTable.Entry>>) -> Bool:
391	  chunks_bounded_go(partition_target(target), chunks)
392	
393	def chunks_strict_unique(chunks: List<&2, List<&2, MemTable.Entry>>) -> Bool:
394	  runs_well_formed(chunks)
395	
396	def chunks_pairwise_disjoint(chunks: List<&2, List<&2, MemTable.Entry>>) -> Bool:
397	  pairwise_disjoint(chunks)