src/ByteMemTable.bend

mylsm-lsm-store@0.6.0.1 · 695 lines · 83 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/Keys.bend as Keys
3import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4
5# Affine byte key and value pair stored newest-first.
6type Entry is Type:
7  Entry{key: Bytes.Bytes, val: Maybe<&1, Bytes.Bytes>}
8
9# Owns the complete newest-first mutation log.
10type Table is Type:
11  Table{entries: List<&1, Entry>}
12
13# Carries rebuilt table ownership alongside an optional cloned value.
14type Lookup is Type:
15  Lookup{entries: List<&1, Entry>, value: Maybe<&1, Bytes.Bytes>, found: Bool}
16
17# Distinguishes no delta mutation, a tombstone, and a live delta value.
18type Probe is Type:
19  Absent{}
20  Deleted{}
21  Present{value: Bytes.Bytes}
22
23# Carries the state for LengthProbe.
24type LengthProbe is Data:
25  LengthAbsent{}
26  LengthDeleted{}
27  LengthPresent{value_bytes: U32}
28
29# Carries the state for LengthLookup.
30type LengthLookup is Type:
31  LengthLookup{entries: List<&1, Entry>, result: LengthProbe, found: Bool}
32
33# Carries the state for LengthSearch.
34type LengthSearch is Type:
35  LengthSearch{query: Bytes.Bytes, entries: List<&1, Entry>,
36    prefix: List<&1, Entry>, order: Cmp}
37
38# Carries affine search state across lexicographic comparisons.
39type Search is Type:
40  Search{query: Bytes.Bytes, entries: List<&1, Entry>, prefix: List<&1, Entry>, order: Cmp}
41
42# Carries the pending entry through a unique-key insertion into sorted entries.
43type OrderedMergeState is Type:
44  OrderedMergeState{newer: List<&1, Entry>, older: List<&1, Entry>, reversed: List<&1, Entry>}
45
46# Creates a byte table with no mutations.
47def empty() -> Table:
48  Table{Nil{}}
49
50# Prepends a byte value mutation without decoding either buffer.
51def put(table: Table, key: Bytes.Bytes, value: Bytes.Bytes) -> Table:
52  match table:
53    case Table{entries}:
54      Table{Con{Entry{key, Some{value}}, entries}}
55
56# Prepends a tombstone for an arbitrary byte key.
57def del(table: Table, key: Bytes.Bytes) -> Table:
58  match table:
59    case Table{entries}:
60      Table{Con{Entry{key, None{}}, entries}}
61
62# Counts each owned mutation in a byte table.
63def count.go(entries: List<&1, Entry>, total: Nat) -> Nat:
64  match entries:
65    case Nil{}:
66      total
67    case Con{_, tail}:
68      count.go(tail, Nat.add(total, 1n))
69
70# Returns the number of mutations, including tombstones.
71def count(table: Table) -> Nat:
72  match table:
73    case Table{entries}:
74      count.go(entries, 0n)
75
76# Counts entries while preserving the list for a subsequent lookup.
77def count.keep.go(
78  entries: List<&1, Entry>,
79  reversed: List<&1, Entry>,
80  total: Nat
81) -> List<&1, Entry> & Nat:
82  match entries:
83    case Nil{}:
84      (List.reverse(&1, Entry, reversed), total)
85    case Con{entry, tail}:
86      count.keep.go(tail, Con{entry, reversed}, Nat.add(total, 1n))
87
88# Returns both the original entry order and its size.
89def count.keep(entries: List<&1, Entry>) -> List<&1, Entry> & Nat:
90  count.keep.go(entries, Nil{}, 0n)
91
92# Reifies the two independently owned buffers returned by the copy effect.
93def bytes.clone.pair(+len: U32, copied: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
94  match copied:
95    case (left, right):
96      (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
97
98# Duplicates a byte buffer only when ownership must be retained by the table.
99def bytes.clone(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
100  match bytes:
101    case Bytes.Bytes{+len, buf}:
102      bytes.clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
103
104# Wraps the original and copied buffers as independent optional values.
105def value.clone.pair(pair: Bytes.Bytes & Bytes.Bytes) -> Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>:
106  match pair:
107    case (original, copy):
108      (Some{original}, Some{copy})
109
110# Clones only a present value while preserving absence.
111def value.clone(value: Maybe<&1, Bytes.Bytes>) -> Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>:
112  match value:
113    case None{}:
114      (None{}, None{})
115    case Some{bytes}:
116      value.clone.pair(bytes.clone(bytes))
117
118# Rebuilds two independently owned entries from cloned fields.
119def entry.clone.fields(
120  keys: Bytes.Bytes & Bytes.Bytes,
121  values: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
122) -> Entry & Entry:
123  match keys:
124    case (left_key, right_key):
125      match values:
126        case (left_value, right_value):
127          (Entry{left_key, left_value}, Entry{right_key, right_value})
128
129# Clones both buffers and the optional value in one entry.
130def entry.clone(entry: Entry) -> Entry & Entry:
131  match entry:
132    case Entry{key, val}:
133      entry.clone.fields(bytes.clone(key), value.clone(val))
134
135def entry.key.copy.pair(
136  copies: Bytes.Bytes & Bytes.Bytes,
137  value: Maybe<&1, Bytes.Bytes>
138) -> Entry & Bytes.Bytes:
139  match copies:
140    case (key, probe): (Entry{key, value}, probe)
141
142def entry.key.copy(entry: Entry) -> Entry & Bytes.Bytes:
143  match entry:
144    case Entry{key, value}:
145      entry.key.copy.pair(bytes.clone(key), value)
146
147def entry.live.keep(entry: Entry) -> Entry & Bool:
148  match entry:
149    case Entry{key, value}:
150      match value:
151        case None{}: (Entry{key, None{}}, False{})
152        case Some{bytes}: (Entry{key, Some{bytes}}, True{})
153
154def entry.live.matches(pair: Entry & Bool) -> Bool:
155  match pair:
156    case (Entry{_, None{}}, False{}): True{}
157    case (Entry{_, Some{_}}, True{}): True{}
158    case _: False{}
159
160# Prepends cloned heads to the corresponding entry lists.
161def entries.clone.push(
162  head: Entry & Entry,
163  tail: List<&1, Entry> & List<&1, Entry>
164) -> List<&1, Entry> & List<&1, Entry>:
165  match head:
166    case (left, right):
167      match tail:
168        case (left_tail, right_tail):
169          (Con{left, left_tail}, Con{right, right_tail})
170
171# Clones an affine entry list while preserving its order.
172def entries.clone(entries: List<&1, Entry>) -> List<&1, Entry> & List<&1, Entry>:
173  match entries:
174    case Nil{}: (Nil{}, Nil{})
175    case Con{entry, tail}:
176      entries.clone.push(entry.clone(entry), entries.clone(tail))
177
178# Appends the remaining search suffix after the reversed visited prefix.
179def append(left: List<&1, Entry>, right: List<&1, Entry>) -> List<&1, Entry>:
180  match left:
181    case Nil{}:
182      right
183    case Con{head, tail}:
184      Con{head, append(tail, right)}
185
186def lookup.result(
187  prefix: List<&1, Entry>,
188  remaining: List<&1, Entry>,
189  value: Maybe<&1, Bytes.Bytes>,
190  found: Bool
191) -> Lookup:
192  Lookup{append(List.reverse(&1, Entry, prefix), remaining), value, found}
193
194# Restores a found entry and preserves its value for the caller.
195def lookup.hit(
196  prefix: List<&1, Entry>,
197  key: Bytes.Bytes,
198  tail: List<&1, Entry>,
199  copies: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
200) -> Lookup:
201  match copies:
202    case (original, visible):
203      lookup.result(prefix, Con{Entry{key, original}, tail}, visible, True{})
204
205# Clones a found value so the table remains readable after the lookup.
206def lookup.hit.value(
207  value: Maybe<&1, Bytes.Bytes>,
208  prefix: List<&1, Entry>,
209  key: Bytes.Bytes,
210  tail: List<&1, Entry>
211) -> Lookup:
212  match value:
213    case None{}:
214      lookup.result(prefix, Con{Entry{key, None{}}, tail}, None{}, True{})
215    case Some{bytes}:
216      lookup.hit(prefix, key, tail, value.clone(Some{bytes}))
217
218# Associates comparison output with the entry that produced it.
219def lookup.state(
220  prefix: List<&1, Entry>,
221  value: Maybe<&1, Bytes.Bytes>,
222  tail: List<&1, Entry>,
223  comparison: Bytes.Bytes & Bytes.Bytes & Cmp
224) -> Search:
225  match comparison:
226    case (key, query, order):
227      Search{query, Con{Entry{key, value}, tail}, prefix, order}
228
229# Advances the affine scan while retaining the prior entry in its prefix.
230def lookup.advance.state(
231  query: Bytes.Bytes,
232  key: Bytes.Bytes,
233  value: Maybe<&1, Bytes.Bytes>,
234  tail: List<&1, Entry>,
235  prefix: List<&1, Entry>
236) -> Search:
237  match tail:
238    case Nil{}:
239      Search{query, Nil{}, Con{Entry{key, value}, prefix}, EQ{}}
240    case Con{Entry{next_key, next_value}, rest}:
241      lookup.state(Con{Entry{key, value}, prefix}, next_value, rest,
242        Keys.Bytes.cmp.with_inputs(next_key, query))
243
244# Searches newest-first entries under a structural recursion bound.
245def lookup.go(fuel: Nat, state: Search) -> Lookup:
246  match fuel:
247    case 0n:
248      match state:
249        case Search{_, entries, prefix, _}:
250          lookup.result(prefix, entries, None{}, False{})
251    case 1n+rest:
252      match state:
253        case Search{query, entries, prefix, order}:
254          match entries:
255            case Nil{}:
256              lookup.result(prefix, Nil{}, None{}, False{})
257            case Con{Entry{key, val}, tail}:
258              match order:
259                case EQ{}:
260                  lookup.hit.value(val, prefix, key, tail)
261                case LT{}:
262                  lookup.go(rest, lookup.advance.state(query, key, val, tail, prefix))
263                case GT{}:
264                  lookup.go(rest, lookup.advance.state(query, key, val, tail, prefix))
265
266# Restores the table and surfaces the optional lookup value.
267def lookup.start.table(result: Lookup) -> Table & Maybe<&1, Bytes.Bytes>:
268  match result:
269    case Lookup{entries, value, _}:
270      (Table{entries}, value)
271
272def lookup.start.probe.value(table: Table, value: Maybe<&1, Bytes.Bytes>, found: Bool) -> Table & Probe:
273  match value:
274    case None{}:
275      match found:
276        case False{}: (table, Absent{})
277        case True{}: (table, Deleted{})
278    case Some{bytes}:
279      match found:
280        case False{}: (table, Absent{})
281        case True{}: (table, Present{bytes})
282
283def lookup.start.probe(result: Lookup) -> Table & Probe:
284  match result:
285    case Lookup{entries, value, found}:
286      lookup.start.probe.value(Table{entries}, value, found)
287
288def lookup.start.counted.lookup(
289  query: Bytes.Bytes,
290  entries: List<&1, Entry>,
291  count: Nat
292) -> Lookup:
293  match entries:
294    case Nil{}: Lookup{Nil{}, None{}, False{}}
295    case Con{Entry{key, val}, tail}:
296      lookup.go(Nat.add(count, 1n),
297        lookup.state(Nil{}, val, tail, Keys.Bytes.cmp.with_inputs(key, query)))
298
299# Starts a lookup with a bound computed from the list length.
300def lookup.start.counted(
301  query: Bytes.Bytes,
302  entries: List<&1, Entry>,
303  count: Nat
304) -> Table & Maybe<&1, Bytes.Bytes>:
305  lookup.start.table(lookup.start.counted.lookup(query, entries, count))
306
307def lookup.start.counted.probe(
308  query: Bytes.Bytes,
309  entries: List<&1, Entry>,
310  count: Nat
311) -> Table & Probe:
312  lookup.start.probe(lookup.start.counted.lookup(query, entries, count))
313
314# Unpacks the preserved list and count before scanning.
315def lookup.start.counted.pair(query: Bytes.Bytes, counted: List<&1, Entry> & Nat) -> Table & Maybe<&1, Bytes.Bytes>:
316  match counted:
317    case (entries, count):
318      lookup.start.counted(query, entries, count)
319
320# Counts and scans a table while preserving its ownership.
321def lookup.start(query: Bytes.Bytes, entries: List<&1, Entry>) -> Table & Maybe<&1, Bytes.Bytes>:
322  lookup.start.counted.pair(query, count.keep(entries))
323
324def lookup.start.probe.pair(
325  query: Bytes.Bytes,
326  counted: List<&1, Entry> & Nat
327) -> Table & Probe:
328  match counted:
329    case (entries, count): lookup.start.counted.probe(query, entries, count)
330
331# Looks up a key while preserving whether the newest delta mutation exists.
332def probe(table: Table, query: Bytes.Bytes) -> Table & Probe:
333  match table:
334    case Table{entries}:
335      lookup.start.probe.pair(query, count.keep(entries))
336
337def lookup.length.result(
338  prefix: List<&1, Entry>,
339  remaining: List<&1, Entry>,
340  result: LengthProbe,
341  found: Bool
342) -> LengthLookup:
343  LengthLookup{append(List.reverse(&1, Entry, prefix), remaining), result,
344    found}
345
346def lookup.length.hit(
347  prefix: List<&1, Entry>,
348  key: Bytes.Bytes,
349  tail: List<&1, Entry>,
350  val: Maybe<&1, Bytes.Bytes>
351) -> LengthLookup:
352  match val:
353    case None{}:
354      lookup.length.result(prefix, Con{Entry{key, None{}}, tail},
355        LengthDeleted{}, True{})
356    case Some{Bytes.Bytes{+len, buf}}:
357      lookup.length.result(prefix,
358        Con{Entry{key, Some{Bytes.Bytes{len, buf}}}, tail},
359        LengthPresent{len}, True{})
360
361def lookup.length.state(
362  prefix: List<&1, Entry>,
363  val: Maybe<&1, Bytes.Bytes>,
364  tail: List<&1, Entry>,
365  comparison: Bytes.Bytes & Bytes.Bytes & Cmp
366) -> LengthSearch:
367  match comparison:
368    case (key, query, order):
369      LengthSearch{query, Con{Entry{key, val}, tail}, prefix, order}
370
371def lookup.length.advance(
372  query: Bytes.Bytes,
373  key: Bytes.Bytes,
374  val: Maybe<&1, Bytes.Bytes>,
375  tail: List<&1, Entry>,
376  prefix: List<&1, Entry>
377) -> LengthSearch:
378  match tail:
379    case Nil{}:
380      LengthSearch{query, Nil{}, Con{Entry{key, val}, prefix}, EQ{}}
381    case Con{Entry{next_key, next_val}, rest}:
382      lookup.length.state(Con{Entry{key, val}, prefix}, next_val, rest,
383        Keys.Bytes.cmp.with_inputs(next_key, query))
384
385def lookup.length.go(fuel: Nat, state: LengthSearch) -> LengthLookup:
386  match fuel:
387    case 0n:
388      match state:
389        case LengthSearch{_, entries, prefix, _}:
390          lookup.length.result(prefix, entries, LengthAbsent{}, False{})
391    case 1n+rest:
392      match state:
393        case LengthSearch{query, entries, prefix, order}:
394          match entries:
395            case Nil{}:
396              lookup.length.result(prefix, Nil{}, LengthAbsent{}, False{})
397            case Con{Entry{key, val}, tail}:
398              match order:
399                case EQ{}: lookup.length.hit(prefix, key, tail, val)
400                case LT{}:
401                  lookup.length.go(rest,
402                    lookup.length.advance(query, key, val, tail, prefix))
403                case GT{}:
404                  lookup.length.go(rest,
405                    lookup.length.advance(query, key, val, tail, prefix))
406
407def lookup.length.output(result: LengthLookup) -> Table & LengthProbe:
408  match result:
409    case LengthLookup{entries, result, _}: (Table{entries}, result)
410
411def lookup.length.start(
412  query: Bytes.Bytes,
413  entries: List<&1, Entry>,
414  count: Nat
415) -> Table & LengthProbe:
416  match entries:
417    case Nil{}: (Table{Nil{}}, LengthAbsent{})
418    case Con{Entry{key, val}, tail}:
419      lookup.length.output(lookup.length.go(Nat.add(count, 1n),
420        lookup.length.state(Nil{}, val, tail,
421          Keys.Bytes.cmp.with_inputs(key, query))))
422
423def lookup.length.start.counted(
424  query: Bytes.Bytes,
425  counted: List<&1, Entry> & Nat
426) -> Table & LengthProbe:
427  match counted:
428    case (entries, count): lookup.length.start(query, entries, count)
429
430def probe.length(table: Table, query: Bytes.Bytes) -> Table & LengthProbe:
431  match table:
432    case Table{entries}:
433      lookup.length.start.counted(query, count.keep(entries))
434
435# Reads a byte key and returns the still-owned table with the result.
436def get(table: Table, query: Bytes.Bytes) -> Table & Maybe<&1, Bytes.Bytes>:
437  match table:
438    case Table{entries}:
439      lookup.start(query, entries)
440
441# Measures a value by its exact encoded byte length.
442def payload.value(value: Maybe<&1, Bytes.Bytes>) -> Nat:
443  match value:
444    case None{}:
445      0n
446    case Some{Bytes.Bytes{+len, _}}:
447      U32.to_nat(len)
448
449# Sums key and present-value byte lengths over the mutation log.
450def payload.go(entries: List<&1, Entry>, total: Nat) -> Nat:
451  match entries:
452    case Nil{}:
453      total
454    case Con{Entry{Bytes.Bytes{+len, _}, val}, tail}:
455      payload.go(tail, Nat.add(Nat.add(total, U32.to_nat(len)), payload.value(val)))
456
457# Returns exact key/value payload bytes, excluding structural overhead.
458def payload_bytes(table: Table) -> Nat:
459  match table:
460    case Table{entries}:
461      payload.go(entries, 0n)
462
463# Adds a fixed per-entry estimate to encoded payload bytes for retention limits.
464def retained.go(entries: List<&1, Entry>, total: Nat) -> Nat:
465  match entries:
466    case Nil{}: total
467    case Con{Entry{Bytes.Bytes{+len, _}, val}, tail}:
468      retained.go(tail,
469        Nat.add(Nat.add(Nat.add(total, 32n), U32.to_nat(len)), payload.value(val)))
470
471# Estimates packed payload and entry metadata retained by a snapshot copy.
472def retained_bytes(table: Table) -> Nat:
473  match table:
474    case Table{entries}:
475      retained.go(entries, 0n)
476
477def retained_bytes.keep.go(
478  entries: List<&1, Entry>,
479  reversed: List<&1, Entry>,
480  total: Nat
481) -> List<&1, Entry> & Nat:
482  match entries:
483    case Nil{}:
484      (List.reverse(&1, Entry, reversed), total)
485    case Con{Entry{key, value}, tail}:
486      match key:
487        case Bytes.Bytes{+key_len, key_buffer}:
488          match value:
489            case None{}:
490              retained_bytes.keep.go(tail,
491                Con{Entry{Bytes.Bytes{key_len, key_buffer}, None{}}, reversed},
492                Nat.add(total, Nat.add(32n, U32.to_nat(key_len))))
493            case Some{Bytes.Bytes{+value_len, value_buffer}}:
494              retained_bytes.keep.go(tail,
495                Con{Entry{Bytes.Bytes{key_len, key_buffer},
496                  Some{Bytes.Bytes{value_len, value_buffer}}}, reversed},
497                Nat.add(total, Nat.add(Nat.add(32n, U32.to_nat(key_len)),
498                  U32.to_nat(value_len))))
499
500def retained_bytes.keep.finish(pair: List<&1, Entry> & Nat) -> Table & Nat:
501  match pair:
502    case (entries, bytes): (Table{entries}, bytes)
503
504def retained_bytes.keep(table: Table) -> Table & Nat:
505  match table:
506    case Table{entries}:
507      retained_bytes.keep.finish(retained_bytes.keep.go(entries, Nil{}, 0n))
508
509# Drops tombstones after all older byte runs have been folded into one table.
510def compact.live(entries: List<&1, Entry>) -> List<&1, Entry>:
511  match entries:
512    case Nil{}:
513      Nil{}
514    case Con{Entry{_, None{}}, tail}:
515      compact.live(tail)
516    case Con{Entry{key, Some{value}}, tail}:
517      Con{Entry{key, Some{value}}, compact.live(tail)}
518
519# Accepts only a strict ascending comparison for neighboring ordered keys.
520def ordered.strict.order(comparison: Bytes.Bytes & Bytes.Bytes & Cmp) -> Bool:
521  match comparison:
522    case (_, _, LT{}):
523      True{}
524    case _:
525      False{}
526
527def ordered.merge.take_newer(order: Cmp) -> Bool:
528  match order:
529    case LT{}: True{}
530    case EQ{}: True{}
531    case GT{}: False{}
532
533# Checks strict key order with fuel derived from the number of entries.
534def ordered.strict.go(fuel: Nat, entries: List<&1, Entry>) -> Bool:
535  match fuel:
536    case 0n:
537      True{}
538    case 1n+rest:
539      match entries:
540        case Nil{}:
541          True{}
542        case Con{_, Nil{}}:
543          True{}
544        case Con{Entry{first, _}, Con{Entry{second, _}, tail}}:
545          +tail_valid = ordered.strict.go(rest, tail)
546          Bool.and(ordered.strict.order(Keys.Bytes.cmp.with_inputs(first, second)), tail_valid)
547
548# Unpacks an ordered list and its preserved recursion bound.
549def ordered.strict.counted(measured: List<&1, Entry> & Nat) -> Bool:
550  match measured:
551    case (entries, count):
552      ordered.strict.go(Nat.add(count, 1n), entries)
553
554# Checks that a sorted entry list is strictly increasing by byte key.
555def ordered.strict(entries: List<&1, Entry>) -> Bool:
556  ordered.strict.counted(count.keep(entries))
557
558def ordered.merge.decisions(+order: Cmp) -> Bool & Bool:
559  (ordered.merge.take_newer(order), Cmp.is_eq(order))
560
561def ordered.merge.resolve(
562  decisions: Bool & Bool,
563  newer: List<&1, Entry>,
564  older: List<&1, Entry>,
565  reversed: List<&1, Entry>,
566  newer_key: Bytes.Bytes,
567  newer_value: Maybe<&1, Bytes.Bytes>,
568  older_key: Bytes.Bytes,
569  older_value: Maybe<&1, Bytes.Bytes>
570) -> OrderedMergeState:
571  match decisions:
572    case (take_newer, equal):
573      match take_newer:
574        case True{}:
575          match equal:
576            case True{}:
577              OrderedMergeState{newer, older,
578                Con{Entry{newer_key, newer_value}, reversed}}
579            case False{}:
580              OrderedMergeState{newer,
581                Con{Entry{older_key, older_value}, older},
582                Con{Entry{newer_key, newer_value}, reversed}}
583        case False{}:
584          OrderedMergeState{
585            Con{Entry{newer_key, newer_value}, newer}, older,
586            Con{Entry{older_key, older_value}, reversed}}
587
588def ordered.merge.compared(
589  state: OrderedMergeState,
590  newer_value: Maybe<&1, Bytes.Bytes>,
591  older_value: Maybe<&1, Bytes.Bytes>,
592  retained_keys: Bytes.Bytes & Bytes.Bytes,
593  comparison: Bytes.Bytes & Bytes.Bytes & Cmp
594) -> OrderedMergeState:
595  match state:
596    case OrderedMergeState{newer, older, reversed}:
597      match retained_keys:
598        case (newer_key, older_key):
599          match comparison:
600            case (_, _, order):
601              ordered.merge.resolve(ordered.merge.decisions(order),
602                newer, older, reversed, newer_key, newer_value, older_key, older_value)
603
604def ordered.merge.cloned(
605  state: OrderedMergeState,
606  newer_value: Maybe<&1, Bytes.Bytes>,
607  older_value: Maybe<&1, Bytes.Bytes>,
608  copies: (Bytes.Bytes & Bytes.Bytes) & (Bytes.Bytes & Bytes.Bytes)
609) -> OrderedMergeState:
610  match copies:
611    case ((newer_key, newer_probe), (older_key, older_probe)):
612      ordered.merge.compared(state, newer_value, older_value,
613        (newer_key, older_key), Keys.Bytes.cmp.with_inputs(newer_probe, older_probe))
614
615def ordered.merge.compare(
616  state: OrderedMergeState,
617  newer: Entry,
618  older: Entry
619) -> OrderedMergeState:
620  match newer older:
621    case Entry{newer_key, newer_value} Entry{older_key, older_value}:
622      ordered.merge.cloned(state, newer_value, older_value,
623        (bytes.clone(newer_key), bytes.clone(older_key)))
624
625def ordered.merge.finish(state: OrderedMergeState) -> List<&1, Entry>:
626  match state:
627    case OrderedMergeState{newer, older, reversed}:
628      List.reverse.go(&1, Entry, reversed, append(newer, older))
629
630def ordered.merge.go(fuel: Nat, state: OrderedMergeState) -> List<&1, Entry>:
631  match fuel:
632    case 0n: ordered.merge.finish(state)
633    case 1n+rest:
634      match state:
635        case OrderedMergeState{newer, older, reversed}:
636          match newer older:
637            case Nil{} _: List.reverse.go(&1, Entry, reversed, older)
638            case _ Nil{}: List.reverse.go(&1, Entry, reversed, newer)
639            case Con{newer_head, newer_tail} Con{older_head, older_tail}:
640              ordered.merge.go(rest, ordered.merge.compare(
641                OrderedMergeState{newer_tail, older_tail, reversed},
642                newer_head, older_head))
643
644def ordered.merge.counted(
645  newer: List<&1, Entry> & Nat,
646  older: List<&1, Entry> & Nat
647) -> List<&1, Entry>:
648  match newer:
649    case (newer_entries, newer_count):
650      match older:
651        case (older_entries, older_count):
652          fuel = Nat.add(Nat.add(newer_count, older_count), 1n)
653          ordered.merge.go(fuel, OrderedMergeState{newer_entries, older_entries, Nil{}})
654
655def ordered.merge_newer(newer: List<&1, Entry>, older: List<&1, Entry>) -> List<&1, Entry>:
656  ordered.merge.counted(count.keep(newer), count.keep(older))
657
658def ordered.singleton_runs(entries: List<&1, Entry>) -> List<&1, List<&1, Entry>>:
659  match entries:
660    case Nil{}: Nil{}
661    case Con{entry, rest}:
662      Con{Con{entry, Nil{}}, ordered.singleton_runs(rest)}
663
664def ordered.merge_round(runs: List<&1, List<&1, Entry>>) -> List<&1, List<&1, Entry>>:
665  match runs:
666    case Nil{}: Nil{}
667    case Con{newer, Nil{}}: Con{newer, Nil{}}
668    case Con{newer, Con{older, rest}}:
669      Con{ordered.merge_newer(newer, older), ordered.merge_round(rest)}
670
671def ordered.merge_many.go(fuel: Nat, runs: List<&1, List<&1, Entry>>) -> List<&1, Entry>:
672  match fuel:
673    case 0n:
674      match runs:
675        case Nil{}: Nil{}
676        case Con{run, _}: run
677    case 1n+rest_fuel:
678      match runs:
679        case Nil{}: Nil{}
680        case Con{run, Nil{}}: run
681        case Con{newer, Con{older, rest}}:
682          ordered.merge_many.go(rest_fuel,
683            Con{ordered.merge_newer(newer, older), ordered.merge_round(rest)})
684
685def ordered.materialize.measured(measured: List<&1, Entry> & Nat) -> List<&1, Entry>:
686  match measured:
687    case (entries, +count):
688      ordered.merge_many.go(Nat.add(count, 1n),
689        ordered.singleton_runs(entries))
690
691# Produces strict byte-ordered unique entries with newest mutations winning.
692def ordered_entries(table: Table) -> List<&1, Entry>:
693  match table:
694    case Table{entries}:
695      ordered.materialize.measured(count.keep(entries))