src/scan/VisibleCursor.bend

mylsm-lsm-store@0.6.0.1 · 524 lines · 52 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.1/src/Keys.bend as Keys
4import mylsm-lsm-store@0.6.0.1/src/scan/ByteRange.bend as ByteRange
5import mylsm-lsm-store@0.6.0.1/src/scan/RangeCursor.bend as RangeCursor
6import mylsm-lsm-store@0.6.0.1/src/scan/ScanMerge.bend as ScanMerge
7import mylsm-lsm-store@0.6.0.1/src/scan/SstCursor.bend as SstCursor
8import mylsm-lsm-store@0.6.0.1/src/scan/VisibleSeekPolicy.bend as VisibleSeekPolicy
9import mylsm-lsm-store@0.6.0.1/src/scan/CursorPreviewPolicy.bend as PreviewPolicy
10import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
11
12# Retains one SST head and one ordered delta source between reads.
13type Cursor is Type:
14  Active{base: Maybe<&1, SstCursor.Cursor>, pending: Maybe<&1, ByteMemTable.Entry>,
15    delta: List<&1, ByteMemTable.Entry>, bounds: RangeCursor.Bounds,
16    staged: Maybe<&1, ByteMemTable.Entry>}
17  Ended{}
18  Closed{}
19  Failed{error: U32 & String}
20
21# Returns one visible item, filtered source step, terminal end, or IO error.
22type Step is Type:
23  Item{entry: ByteMemTable.Entry, cursor: Cursor}
24  Skip{cursor: Cursor}
25  End{}
26  Error{error: U32 & String}
27
28# Carries the state for Preview.
29type Preview is Type:
30  PreviewReady{cursor: Cursor, payload_bytes: Nat}
31  PreviewEnded{cursor: Cursor}
32  PreviewFailed{cursor: Cursor}
33
34# Carries the state for PreviewState.
35type PreviewState is Type:
36  PreviewNeed{cursor: Cursor}
37  PreviewRead{step: Step}
38
39# Carries one in-range delta entry or exhaustion at the upper bound.
40type DeltaHead is Type:
41  DeltaReady{entry: ByteMemTable.Entry, tail: List<&1, ByteMemTable.Entry>}
42  DeltaDone{}
43
44# Normalizes the block cursor outcome before merge resolution.
45type BaseResult is Type:
46  BaseItem{cursor: SstCursor.Cursor, entry: ByteMemTable.Entry}
47  BaseSkip{cursor: SstCursor.Cursor}
48  BaseDone{}
49  BaseError{error: U32 & String}
50
51# Holds both source heads while the byte comparator decides their order.
52type CompareState is Type:
53  Compare{base: Maybe<&1, SstCursor.Cursor>, base_entry: ByteMemTable.Entry,
54    delta_entry: ByteMemTable.Entry, delta_tail: List<&1, ByteMemTable.Entry>,
55    bounds: RangeCursor.Bounds, order: Cmp}
56
57# Preserves comparison metadata while cloning an affine delta entry.
58type ResolveState is Type:
59  Resolve{base: Maybe<&1, SstCursor.Cursor>, base_entry: ByteMemTable.Entry,
60    delta_tail: List<&1, ByteMemTable.Entry>, bounds: RangeCursor.Bounds,
61    order: Cmp}
62
63# Opens the merge state over an already bounded SST cursor and captured delta.
64def open(
65  base: Maybe<&1, SstCursor.Cursor>,
66  delta: List<&1, ByteMemTable.Entry>,
67  bounds: RangeCursor.Bounds
68) -> Cursor:
69  Active{base, None{}, delta, bounds, None{}}
70
71def base.seek(base: Maybe<&1, SstCursor.Cursor>, lower: Maybe<&1, Bytes.Bytes>) -> Maybe<&1, SstCursor.Cursor>:
72  match base:
73    case Some{cursor}: Some{SstCursor.seek(cursor, lower)}
74    case None{}: None{}
75
76def lower.triple.second(
77  first: Bytes.Bytes,
78  copies: Bytes.Bytes & Bytes.Bytes
79) -> Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>:
80  match copies:
81    case (second, third): (Some{first}, Some{second}, Some{third})
82
83def lower.triple.copy(
84  values: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
85) -> Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>:
86  match values:
87    case (None{}, None{}): (None{}, None{}, None{})
88    case (None{}, Some{_}): (None{}, None{}, None{})
89    case (Some{_}, None{}): (None{}, None{}, None{})
90    case (Some{first}, Some{second}):
91      lower.triple.second(first, ByteMemTable.bytes.clone(second))
92
93def lower.triple(
94  lower: Maybe<&1, Bytes.Bytes>
95) ->
96  Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>:
97  lower.triple.copy(ByteMemTable.value.clone(lower))
98
99def seek.ready(
100  base: Maybe<&1, SstCursor.Cursor>,
101  delta: List<&1, ByteMemTable.Entry>,
102  bounds: RangeCursor.Bounds,
103  lowers: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
104) -> Cursor:
105  match lowers:
106    case (base_lower, (delta_lower, bounds_lower)):
107      Active{base.seek(base, base_lower), None{}, RangeCursor.seek(delta, delta_lower),
108        VisibleSeekPolicy.seek.bounds(bounds, bounds_lower), None{}}
109
110# Advances every retained source to the same inclusive lower key.
111def seek(cursor: Cursor, lower: Maybe<&1, Bytes.Bytes>) -> Cursor:
112  match cursor:
113    case Active{base, _, delta, bounds, _}:
114      seek.ready(base, delta, bounds, lower.triple(lower))
115    case Ended{}: Ended{}
116    case Closed{}: Closed{}
117    case Failed{error}: Failed{error}
118
119def delta.head.result(
120  reached: Bool,
121  entry: ByteMemTable.Entry,
122  tail: List<&1, ByteMemTable.Entry>
123) -> DeltaHead:
124  match reached:
125    case True{}: DeltaDone{}
126    case False{}: DeltaReady{entry, tail}
127
128def delta.head.checked(
129  entry: ByteMemTable.Entry,
130  tail: List<&1, ByteMemTable.Entry>,
131  key: Bytes.Bytes,
132  upper: Maybe<&1, Bytes.Bytes>
133) -> DeltaHead:
134  delta.head.result(ByteRange.at_or_after_upper(key, upper), entry, tail)
135
136def delta.head.copied(
137  entries: ByteMemTable.Entry & Bytes.Bytes,
138  tail: List<&1, ByteMemTable.Entry>,
139  bounds: RangeCursor.Bounds
140) -> DeltaHead:
141  match entries:
142    case (entry, key):
143      match bounds:
144        case RangeCursor.Bounds{_, upper, _}:
145          delta.head.checked(entry, tail, key, upper)
146
147def delta.head(
148  entry: ByteMemTable.Entry,
149  tail: List<&1, ByteMemTable.Entry>,
150  bounds: RangeCursor.Bounds
151) -> DeltaHead:
152  delta.head.copied(ByteMemTable.entry.key.copy(entry), tail, bounds)
153
154def delta.next(entries: List<&1, ByteMemTable.Entry>, bounds: RangeCursor.Bounds) -> DeltaHead:
155  match entries:
156    case Nil{}: DeltaDone{}
157    case Con{entry, tail}: delta.head(entry, tail, bounds)
158
159def base.result(result: SstCursor.Step) -> BaseResult:
160  match result:
161    case SstCursor.Item{entry, cursor}: BaseItem{cursor, entry}
162    case SstCursor.Skip{cursor}: BaseSkip{cursor}
163    case SstCursor.End{}: BaseDone{}
164    case SstCursor.Error{error}: BaseError{error}
165
166def delta.output.result(
167  visible: Bool,
168  live: Bool,
169  entry: ByteMemTable.Entry,
170  base: Maybe<&1, SstCursor.Cursor>,
171  pending: Maybe<&1, ByteMemTable.Entry>,
172  tail: List<&1, ByteMemTable.Entry>,
173  bounds: RangeCursor.Bounds
174) -> Step:
175  match visible:
176    case True{}:
177      match live:
178        case True{}: Item{entry, Active{base, pending, tail, bounds, None{}}}
179        case False{}: Skip{Active{base, pending, tail, bounds, None{}}}
180    case False{}: Skip{Active{base, pending, tail, bounds, None{}}}
181
182def delta.output.checks(
183  entry: ByteMemTable.Entry,
184  live: Bool,
185  key: Bytes.Bytes,
186  base: Maybe<&1, SstCursor.Cursor>,
187  pending: Maybe<&1, ByteMemTable.Entry>,
188  tail: List<&1, ByteMemTable.Entry>,
189  copies: RangeCursor.Bounds & RangeCursor.Bounds
190) -> Step:
191  match copies:
192    case (checked, retained):
193      delta.output.result(RangeCursor.matches.key(ByteMemTable.bytes.clone(key), checked), live,
194        entry, base, pending, tail, retained)
195
196def delta.output.live.kept(
197  pair: ByteMemTable.Entry & Bool,
198  key: Bytes.Bytes,
199  base: Maybe<&1, SstCursor.Cursor>,
200  pending: Maybe<&1, ByteMemTable.Entry>,
201  tail: List<&1, ByteMemTable.Entry>,
202  copies: RangeCursor.Bounds & RangeCursor.Bounds
203) -> Step:
204  match pair:
205    case (retained, live):
206      delta.output.checks(retained, live, key, base, pending, tail, copies)
207
208def delta.output.live(
209  entry: ByteMemTable.Entry,
210  key: Bytes.Bytes,
211  base: Maybe<&1, SstCursor.Cursor>,
212  pending: Maybe<&1, ByteMemTable.Entry>,
213  tail: List<&1, ByteMemTable.Entry>,
214  copies: RangeCursor.Bounds & RangeCursor.Bounds
215) -> Step:
216  delta.output.live.kept(ByteMemTable.entry.live.keep(entry), key,
217    base, pending, tail, copies)
218
219def delta.output.copied(
220  pair: ByteMemTable.Entry & Bytes.Bytes,
221  base: Maybe<&1, SstCursor.Cursor>,
222  pending: Maybe<&1, ByteMemTable.Entry>,
223  tail: List<&1, ByteMemTable.Entry>,
224  bounds: RangeCursor.Bounds
225) -> Step:
226  match pair:
227    case (retained, key):
228      delta.output.live(retained, key, base, pending, tail,
229        RangeCursor.bounds.clone(bounds))
230
231def delta.output(
232  entry: ByteMemTable.Entry,
233  base: Maybe<&1, SstCursor.Cursor>,
234  pending: Maybe<&1, ByteMemTable.Entry>,
235  tail: List<&1, ByteMemTable.Entry>,
236  bounds: RangeCursor.Bounds
237) -> Step:
238  delta.output.copied(ByteMemTable.entry.key.copy(entry), base, pending,
239    tail, bounds)
240
241def decision.step(
242  decision: ScanMerge.Decision,
243  base: Maybe<&1, SstCursor.Cursor>,
244  base_entry: ByteMemTable.Entry,
245  delta_entry: ByteMemTable.Entry,
246  delta_tail: List<&1, ByteMemTable.Entry>,
247  bounds: RangeCursor.Bounds
248) -> Step:
249  match decision:
250    case ScanMerge.EmitDelta{}:
251      delta.output(delta_entry, base, Some{base_entry}, delta_tail, bounds)
252    case ScanMerge.EmitBase{}:
253      match base:
254        case Some{cursor}:
255          Item{base_entry, Active{Some{cursor}, None{},
256            Con{delta_entry, delta_tail}, bounds, None{}}}
257        case None{}: End{}
258    case ScanMerge.EmitBoth{}:
259      delta.output(delta_entry, base, None{}, delta_tail, bounds)
260    case ScanMerge.SkipDelta{}:
261      Skip{Active{base, Some{base_entry}, delta_tail, bounds, None{}}}
262    case ScanMerge.SkipBoth{}:
263      Skip{Active{base, None{}, delta_tail, bounds, None{}}}
264    case ScanMerge.Finished{}: End{}
265
266def compare.order.only(result: Bytes.Bytes & Bytes.Bytes & Cmp) -> Cmp:
267  match result:
268    case (_, _, order): order
269
270def compare.order(base_key: Bytes.Bytes, delta_key: Bytes.Bytes) -> Cmp:
271  compare.order.only(Keys.Bytes.cmp.with_inputs(delta_key, base_key))
272
273def compare.final(
274  base: Maybe<&1, SstCursor.Cursor>,
275  base_entry: ByteMemTable.Entry,
276  delta_entry: ByteMemTable.Entry,
277  delta_tail: List<&1, ByteMemTable.Entry>,
278  bounds: RangeCursor.Bounds,
279  base_key: Bytes.Bytes,
280  delta_key: Bytes.Bytes,
281) -> CompareState:
282  Compare{base, base_entry, delta_entry, delta_tail, bounds,
283    compare.order(base_key, delta_key)}
284
285def compare.delta.copied(
286  base: Maybe<&1, SstCursor.Cursor>,
287  base_entry: ByteMemTable.Entry,
288  delta_entries: ByteMemTable.Entry & Bytes.Bytes,
289  delta_tail: List<&1, ByteMemTable.Entry>,
290  bounds: RangeCursor.Bounds,
291  base_key: Bytes.Bytes
292) -> CompareState:
293  match delta_entries:
294    case (delta_entry, delta_key):
295      compare.final(base, base_entry, delta_entry, delta_tail, bounds,
296        base_key, delta_key)
297
298def compare.base.copied(
299  base: Maybe<&1, SstCursor.Cursor>,
300  base_entries: ByteMemTable.Entry & Bytes.Bytes,
301  delta_entry: ByteMemTable.Entry,
302  delta_tail: List<&1, ByteMemTable.Entry>,
303  bounds: RangeCursor.Bounds
304) -> CompareState:
305  match base_entries:
306    case (base_entry, base_key):
307      compare.delta.copied(base, base_entry,
308        ByteMemTable.entry.key.copy(delta_entry), delta_tail, bounds, base_key)
309
310# Clones source heads and computes their canonical byte ordering.
311def compare(
312  base: Maybe<&1, SstCursor.Cursor>,
313  base_entry: ByteMemTable.Entry,
314  delta_entry: ByteMemTable.Entry,
315  delta_tail: List<&1, ByteMemTable.Entry>,
316  bounds: RangeCursor.Bounds
317) -> CompareState:
318  compare.base.copied(base, ByteMemTable.entry.key.copy(base_entry),
319    delta_entry, delta_tail, bounds)
320
321def resolve.delta.pair(
322  state: ResolveState,
323  delta_entry: ByteMemTable.Entry & Bool
324) -> Step:
325  match state:
326    case Resolve{base, base_entry, delta_tail, bounds, order}:
327      match delta_entry:
328        case (entry, live):
329          decision.step(ScanMerge.choose(True{}, True{}, live, order),
330            base, base_entry, entry, delta_tail, bounds)
331
332def resolve.compared(state: CompareState) -> Step:
333  match state:
334    case Compare{base, base_entry, delta_entry, delta_tail, bounds, order}:
335      resolve.delta.pair(Resolve{base, base_entry, delta_tail, bounds, order},
336        ByteMemTable.entry.live.keep(delta_entry))
337
338# Applies the shared newest-wins decision to the current source heads.
339def resolve(
340  base: Maybe<&1, SstCursor.Cursor>,
341  base_entry: ByteMemTable.Entry,
342  delta_entry: ByteMemTable.Entry,
343  delta_tail: List<&1, ByteMemTable.Entry>,
344  bounds: RangeCursor.Bounds
345) -> Step:
346  resolve.compared(compare(base, base_entry, delta_entry, delta_tail, bounds))
347
348def ready.delta(
349  base: Maybe<&1, SstCursor.Cursor>,
350  pending: Maybe<&1, ByteMemTable.Entry>,
351  bounds: RangeCursor.Bounds,
352  head: DeltaHead
353) -> Step:
354  match base:
355    case Some{base_cursor}:
356      match pending:
357        case Some{base_entry}:
358          match head:
359            case DeltaDone{}:
360              Item{base_entry, Active{Some{base_cursor}, None{}, Nil{}, bounds, None{}}}
361            case DeltaReady{entry, tail}:
362              resolve(Some{base_cursor}, base_entry, entry, tail, bounds)
363        case None{}: End{}
364    case None{}:
365      match pending:
366        case None{}:
367          match head:
368            case DeltaDone{}: End{}
369            case DeltaReady{entry, tail}:
370              delta.output(entry, None{}, None{}, tail, bounds)
371        case Some{_}: End{}
372
373def ready.copied(
374  base: Maybe<&1, SstCursor.Cursor>,
375  pending: Maybe<&1, ByteMemTable.Entry>,
376  delta: List<&1, ByteMemTable.Entry>,
377  copies: RangeCursor.Bounds & RangeCursor.Bounds
378) -> Step:
379  match copies:
380    case (bounds, delta_bounds):
381      ready.delta(base, pending, bounds, delta.next(delta, delta_bounds))
382
383# Resolves one already-read SST head against the next bounded delta head.
384def ready(
385  base: Maybe<&1, SstCursor.Cursor>,
386  pending: Maybe<&1, ByteMemTable.Entry>,
387  delta: List<&1, ByteMemTable.Entry>,
388  bounds: RangeCursor.Bounds
389) -> Step:
390  ready.copied(base, pending, delta, RangeCursor.bounds.clone(bounds))
391
392def base.next.convert(
393  delta: List<&1, ByteMemTable.Entry>,
394  bounds: RangeCursor.Bounds,
395  result: BaseResult
396) -> IO(Step):
397  match result:
398    case BaseItem{cursor, entry}:
399      IO.pure(Step, ready(Some{cursor}, Some{entry}, delta, bounds))
400    case BaseSkip{cursor}:
401      IO.pure(Step, Skip{Active{Some{cursor}, None{}, delta, bounds, None{}}})
402    case BaseDone{}:
403      IO.pure(Step, ready(None{}, None{}, delta, bounds))
404    case BaseError{error}: IO.pure(Step, Error{error})
405
406def base.next.result(
407  delta: List<&1, ByteMemTable.Entry>,
408  bounds: RangeCursor.Bounds,
409  result: SstCursor.Step
410) -> IO(Step):
411  base.next.convert(delta, bounds, base.result(result))
412
413# Advances the SST source by at most one record and merges one visible result.
414def next(cursor: Cursor) -> IO(Step):
415  match cursor:
416    case Active{base, pending, delta, bounds, Some{entry}}:
417      IO.pure(Step, Item{entry, Active{base, pending, delta, bounds, None{}}})
418    case Active{Some{base}, None{}, delta, bounds, None{}}:
419      do IO<Step>:
420        result : SstCursor.Step <- SstCursor.next(base)
421        step : Step <- base.next.result(delta, bounds, result)
422        return step
423    case Active{base, pending, delta, bounds, None{}}:
424      IO.pure(Step, ready(base, pending, delta, bounds))
425    case Ended{}: IO.pure(Step, End{})
426    case Closed{}: IO.pure(Step, End{})
427    case Failed{error}: IO.pure(Step, Error{error})
428
429# Closes the active SST file descriptor and leaves no source owned by the caller.
430def close(cursor: Cursor) -> IO(Result<&1, &1, U32 & String, Unit>):
431  match cursor:
432    case Active{Some{base}, _, _, _, _}: SstCursor.close(base)
433    case _:
434      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
435
436def preview.stage(entry: ByteMemTable.Entry, cursor: Cursor,
437  payload_bytes: Nat) -> Preview:
438  match cursor:
439    case Active{base, pending, delta, bounds, _}:
440      PreviewReady{Active{base, pending, delta, bounds, Some{entry}},
441        payload_bytes}
442    case _:
443      PreviewReady{cursor, payload_bytes}
444
445def preview.item.value.length(
446  key: Bytes.Bytes,
447  +key_length: U32,
448  measured: Bytes.Bytes & U32
449) -> ByteMemTable.Entry & Nat:
450  match measured:
451    case (retained_bytes, value_length):
452      (ByteMemTable.Entry{key, Some{retained_bytes}},
453        PreviewPolicy.cursor.item.payload.bytes(
454          U32.to_nat(key_length), U32.to_nat(value_length)))
455
456def preview.item.key.length(
457  bytes: Bytes.Bytes,
458  measured: Bytes.Bytes & U32
459) -> ByteMemTable.Entry & Nat:
460  match measured:
461    case (key, key_length):
462      preview.item.value.length(key, key_length,
463        Bytes.length(bytes))
464
465def preview.item.length(entry: ByteMemTable.Entry)
466  -> ByteMemTable.Entry & Nat:
467  match entry:
468    case ByteMemTable.Entry{key, value}:
469      match value:
470        case Some{bytes}:
471          preview.item.key.length(bytes, Bytes.length(key))
472        case None{}: (entry, 0n)
473
474def preview.item.result(
475  measured: ByteMemTable.Entry & Nat,
476  cursor: Cursor
477) -> Preview:
478  match measured:
479    case (retained, payload_bytes):
480      match retained:
481        case ByteMemTable.Entry{_, value}:
482          match value:
483            case Some{_}:
484              preview.stage(retained, cursor, payload_bytes)
485            case None{}: PreviewFailed{cursor}
486
487def preview.item(entry: ByteMemTable.Entry, cursor: Cursor) -> Preview:
488  preview.item.result(preview.item.length(entry), cursor)
489
490def preview.loop(fuel: Nat, state: PreviewState) -> IO(Preview):
491  match fuel:
492    case 0n:
493      match state:
494        case PreviewNeed{cursor}: IO.pure(Preview, PreviewFailed{cursor})
495        case PreviewRead{step}:
496          match step:
497            case Item{_, cursor}:
498              IO.pure(Preview, PreviewFailed{cursor})
499            case Skip{cursor}:
500              IO.pure(Preview, PreviewFailed{cursor})
501            case End{}:
502              IO.pure(Preview, PreviewEnded{Ended{}})
503            case Error{error}:
504              IO.pure(Preview, PreviewFailed{Failed{error}})
505    case 1n+rest:
506      match state:
507        case PreviewNeed{cursor}:
508          do IO<Preview>:
509            step : Step <- next(cursor)
510            preview.loop(rest, PreviewRead{step})
511        case PreviewRead{step}:
512          match step:
513            case Item{entry, advanced}:
514              IO.pure(Preview, preview.item(entry, advanced))
515            case Skip{advanced}:
516              preview.loop(rest, PreviewNeed{advanced})
517            case End{}:
518              IO.pure(Preview, PreviewEnded{Ended{}})
519            case Error{error}:
520              IO.pure(Preview, PreviewFailed{Failed{error}})
521
522# Handles preview.
523def preview(cursor: Cursor) -> IO(Preview):
524  preview.loop(4096n, PreviewNeed{cursor})