src/scan/VisibleCursor.bend
mylsm-lsm-store@0.6.0.0 · 524 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/Keys.bend as Keys
4import mylsm-lsm-store@0.6.0.0/src/scan/ByteRange.bend as ByteRange
5import mylsm-lsm-store@0.6.0.0/src/scan/RangeCursor.bend as RangeCursor
6import mylsm-lsm-store@0.6.0.0/src/scan/ScanMerge.bend as ScanMerge
7import mylsm-lsm-store@0.6.0.0/src/scan/SstCursor.bend as SstCursor
8import mylsm-lsm-store@0.6.0.0/src/scan/VisibleSeekPolicy.bend as VisibleSeekPolicy
9import mylsm-lsm-store@0.6.0.0/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})