src/scan/SstCursor.bend
mylsm-lsm-store@0.6.0.0 · 443 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/ByteWal.bend as ByteWal
4import mylsm-lsm-store@0.6.0.0/src/scan/ByteBlockReader.bend as ByteBlockReader
5import mylsm-lsm-store@0.6.0.0/src/scan/ByteRange.bend as ByteRange
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Carries half-open byte bounds and an optional exact byte prefix.
9type Bounds is Type:
10 Bounds{lower: Maybe<&1, Bytes.Bytes>, upper: Maybe<&1, Bytes.Bytes>,
11 prefix: Maybe<&1, Bytes.Bytes>}
12
13# Owns one validated SST reader and the unread part of its current block.
14type Cursor is Type:
15 Active{inner: ByteBlockReader.EntryCursor, bounds: Bounds}
16 Ended{}
17 Closed{}
18 Failed{error: U32 & String}
19
20# Represents a streamed record, filtered entry, end, or terminal error.
21type Step is Type:
22 Item{entry: ByteMemTable.Entry, cursor: Cursor}
23 Skip{cursor: Cursor}
24 End{}
25 Error{error: U32 & String}
26
27# Carries each bounded block validation transition before cursor publication.
28type VerifyState is Type:
29 VerifyNeed{reader: ByteBlockReader.Reader, bounds: Bounds}
30 VerifyRead{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>,
31 bounds: Bounds}
32
33# Carries one block read or one current block during incremental scanning.
34type ScanState is Type:
35 ScanNeed{reader: ByteBlockReader.Reader, bounds: Bounds}
36 ScanRead{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>,
37 bounds: Bounds}
38 ScanBlock{reader: ByteBlockReader.Reader,
39 entries: List<&1, ByteMemTable.Entry>, bounds: Bounds}
40
41def bounds.copy.values(
42 lower: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>,
43 upper: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>,
44 prefix: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
45) -> Bounds & Bounds:
46 match lower:
47 case (lower_original, lower_copy):
48 match upper:
49 case (upper_original, upper_copy):
50 match prefix:
51 case (prefix_original, prefix_copy):
52 (Bounds{lower_original, upper_original, prefix_original},
53 Bounds{lower_copy, upper_copy, prefix_copy})
54
55def bounds.copy(bounds: Bounds) -> Bounds & Bounds:
56 match bounds:
57 case Bounds{lower, upper, prefix}:
58 bounds.copy.values(ByteMemTable.value.clone(lower),
59 ByteMemTable.value.clone(upper), ByteMemTable.value.clone(prefix))
60
61def scan.prefix(key: Bytes.Bytes, prefix: Maybe<&1, Bytes.Bytes>) -> Bool:
62 match prefix:
63 case None{}: True{}
64 case Some{bytes}: ByteRange.has_prefix(key, bytes)
65
66def scan.entry.choose(
67 valid: Bool,
68 key: Bytes.Bytes,
69 value: Bytes.Bytes,
70 cursor: ByteBlockReader.EntryCursor,
71 bounds: Bounds & Bounds
72) -> IO(Step):
73 match valid:
74 case True{}:
75 match bounds:
76 case (item_bounds, _):
77 IO.pure(Step, Item{ByteMemTable.Entry{key, Some{value}},
78 Active{cursor, item_bounds}})
79 case False{}:
80 match bounds:
81 case (_, skip_bounds): IO.pure(Step, Skip{Active{cursor, skip_bounds}})
82
83def scan.entry.checked(
84 key: Bytes.Bytes,
85 value: Maybe<&1, Bytes.Bytes>,
86 cursor: ByteBlockReader.EntryCursor,
87 bounds: Bounds & Bounds,
88 key_copies: Bytes.Bytes & Bytes.Bytes
89) -> IO(Step):
90 match value:
91 case Some{bytes}:
92 match bounds:
93 case (checked, retained):
94 match checked:
95 case Bounds{lower, upper, prefix}:
96 match key_copies:
97 case (range_key, prefix_key):
98 scan.entry.choose(
99 ByteRange.in_range(range_key, lower, upper) &&
100 scan.prefix(prefix_key, prefix),
101 key, bytes, cursor, bounds.copy(retained))
102 case None{}:
103 match bounds:
104 case (_, retained): IO.pure(Step, Skip{Active{cursor, retained}})
105
106def scan.entry.upper.closed(
107 closed: Result<&1, &1, U32 & String, Unit>
108) -> Step:
109 match closed:
110 case Done{Unit{}}: End{}
111 case Fail{error}: Error{error}
112
113def scan.entry.upper.result(
114 reached: Bool,
115 key: Bytes.Bytes,
116 value: Maybe<&1, Bytes.Bytes>,
117 cursor: ByteBlockReader.EntryCursor,
118 lower: Maybe<&1, Bytes.Bytes>,
119 prefix: Maybe<&1, Bytes.Bytes>,
120 retained: Bounds,
121 upper_checked: Maybe<&1, Bytes.Bytes>,
122 key_copies: Bytes.Bytes & Bytes.Bytes
123) -> IO(Step):
124 match reached:
125 case True{}:
126 do IO<Step>:
127 closed : Result<&1, &1, U32 & String, Unit> <-
128 ByteBlockReader.entry.close(cursor)
129 return scan.entry.upper.closed(closed)
130 case False{}:
131 scan.entry.checked(key, value, cursor,
132 (Bounds{lower, upper_checked, prefix}, retained), key_copies)
133
134def scan.entry.upper.copies(
135 key: Bytes.Bytes,
136 value: Maybe<&1, Bytes.Bytes>,
137 cursor: ByteBlockReader.EntryCursor,
138 lower: Maybe<&1, Bytes.Bytes>,
139 prefix: Maybe<&1, Bytes.Bytes>,
140 retained: Bounds,
141 upper: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>,
142 key_copies: Bytes.Bytes & Bytes.Bytes
143) -> IO(Step):
144 match upper:
145 case (upper_test, upper_checked):
146 match key_copies:
147 case (upper_key, filter_key):
148 scan.entry.upper.result(
149 ByteRange.at_or_after_upper(upper_key, upper_test),
150 key, value, cursor, lower, prefix, retained,
151 upper_checked,
152 ByteMemTable.bytes.clone(filter_key))
153
154def scan.entry.upper.fields(
155 key: Bytes.Bytes,
156 value: Maybe<&1, Bytes.Bytes>,
157 cursor: ByteBlockReader.EntryCursor,
158 lower: Maybe<&1, Bytes.Bytes>,
159 upper: Maybe<&1, Bytes.Bytes>,
160 prefix: Maybe<&1, Bytes.Bytes>,
161 retained: Bounds,
162 key_copies: Bytes.Bytes & Bytes.Bytes
163) -> IO(Step):
164 scan.entry.upper.copies(key, value, cursor, lower, prefix, retained,
165 ByteMemTable.value.clone(upper), key_copies)
166
167def scan.entry.upper.checked(
168 key: Bytes.Bytes,
169 value: Maybe<&1, Bytes.Bytes>,
170 cursor: ByteBlockReader.EntryCursor,
171 bounds: Bounds & Bounds,
172 key_copies: Bytes.Bytes & Bytes.Bytes
173) -> IO(Step):
174 match bounds:
175 case (checked, retained):
176 match checked:
177 case Bounds{lower, upper, prefix}:
178 scan.entry.upper.fields(key, value, cursor, lower, upper, prefix,
179 retained, key_copies)
180
181def scan.entry.key.copied(
182 value: Maybe<&1, Bytes.Bytes>,
183 cursor: ByteBlockReader.EntryCursor,
184 bounds: Bounds & Bounds,
185 pair: Bytes.Bytes & Bytes.Bytes
186) -> IO(Step):
187 match pair:
188 case (key, check_key):
189 scan.entry.upper.checked(key, value, cursor, bounds,
190 ByteMemTable.bytes.clone(check_key))
191
192def scan.entry(
193 entry: ByteMemTable.Entry,
194 cursor: ByteBlockReader.EntryCursor,
195 bounds: Bounds
196) -> IO(Step):
197 match entry:
198 case ByteMemTable.Entry{key, value}:
199 scan.entry.key.copied(value, cursor, bounds.copy(bounds),
200 ByteMemTable.bytes.clone(key))
201
202def scan.exhausted(state: ScanState) -> IO(Step):
203 match state:
204 case ScanNeed{reader, _}:
205 do IO<Step>:
206 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
207 return Error{(7, "SST cursor exceeded its transition budget")}
208 case ScanRead{result, _}:
209 match result:
210 case Done{ByteBlockReader.Block{reader, _}}:
211 do IO<Step>:
212 _closed : Result<&1, &1, U32 & String, Unit> <-
213 ByteBlockReader.close(reader)
214 return Error{(7, "SST cursor exceeded its transition budget")}
215 case _:
216 IO.pure(Step, Error{(7, "SST cursor exceeded its transition budget")})
217 case ScanBlock{reader, _, _}:
218 do IO<Step>:
219 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
220 return Error{(7, "SST cursor exceeded its transition budget")}
221
222def scan.loop(fuel: Nat, state: ScanState) -> IO(Step):
223 match fuel:
224 case 0n: scan.exhausted(state)
225 case 1n+rest:
226 match state:
227 case ScanNeed{reader, bounds}:
228 do IO<Step>:
229 result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <-
230 ByteBlockReader.next(reader)
231 scan.loop(rest, ScanRead{result, bounds})
232 case ScanRead{result, bounds}:
233 match result:
234 case Fail{error}:
235 IO.pure(Step, Error{error})
236 case Done{ByteBlockReader.End{}}:
237 IO.pure(Step, End{})
238 case Done{ByteBlockReader.Block{reader, entries}}:
239 scan.loop(rest, ScanBlock{reader, entries, bounds})
240 case ScanBlock{reader, entries, bounds}:
241 match entries:
242 case Nil{}:
243 scan.loop(rest, ScanNeed{reader, bounds})
244 case Con{entry, tail}:
245 scan.entry(entry, ByteBlockReader.EntryReady{reader, tail}, bounds)
246
247def verify.exhausted(state: VerifyState) -> IO(Result<&1, &1, U32 & String, Cursor>):
248 match state:
249 case VerifyNeed{reader, _}:
250 do IO<Result<&1, &1, U32 & String, Cursor>>:
251 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader)
252 return Fail{(7, "SST validation exceeded its transition budget")}
253 case VerifyRead{result, _}:
254 match result:
255 case Done{ByteBlockReader.Block{reader, _}}:
256 do IO<Result<&1, &1, U32 & String, Cursor>>:
257 _closed : Result<&1, &1, U32 & String, Unit> <-
258 ByteBlockReader.close(reader)
259 return Fail{(7, "SST validation exceeded its transition budget")}
260 case _:
261 IO.pure(Result<&1, &1, U32 & String, Cursor>,
262 Fail{(7, "SST validation exceeded its transition budget")})
263
264def verify.complete.routed.result(
265 bounds: Bounds,
266 result: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
267) -> IO(Result<&1, &1, U32 & String, Cursor>):
268 match result:
269 case Fail{error}:
270 IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error})
271 case Done{reader}:
272 IO.pure(Result<&1, &1, U32 & String, Cursor>,
273 Done{Active{ByteBlockReader.entry.open(reader), bounds}})
274
275def verify.complete.routed(
276 bounds: Bounds,
277 result: IO(Result<&1, &1, U32 & String, ByteBlockReader.Reader>)
278) -> IO(Result<&1, &1, U32 & String, Cursor>):
279 do IO<Result<&1, &1, U32 & String, Cursor>>:
280 routed : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <- result
281 verify.complete.routed.result(bounds, routed)
282
283def verify.complete.copied(
284 pair: Bounds & Bounds,
285 reader: ByteBlockReader.Reader
286) -> IO(Result<&1, &1, U32 & String, Cursor>):
287 match pair:
288 case (retained, seek_bounds):
289 match seek_bounds:
290 case Bounds{lower, _, _}:
291 match lower:
292 case None{}:
293 IO.pure(Result<&1, &1, U32 & String, Cursor>,
294 Done{Active{ByteBlockReader.entry.open(reader), retained}})
295 case Some{key}:
296 verify.complete.routed(retained,
297 ByteBlockReader.seek_key_after_validation(reader, key))
298
299def verify.complete(
300 bounds: Bounds,
301 result: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
302) -> IO(Result<&1, &1, U32 & String, Cursor>):
303 match result:
304 case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error})
305 case Done{reader}:
306 verify.complete.copied(bounds.copy(bounds), reader)
307
308def verify.loop(
309 fuel: Nat,
310 path: String,
311 state: VerifyState
312) -> IO(Result<&1, &1, U32 & String, Cursor>):
313 match fuel:
314 case 0n: verify.exhausted(state)
315 case 1n+rest:
316 match state:
317 case VerifyNeed{reader, bounds}:
318 do IO<Result<&1, &1, U32 & String, Cursor>>:
319 result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <-
320 ByteBlockReader.next(reader)
321 verify.loop(rest, path, VerifyRead{result, bounds})
322 case VerifyRead{result, bounds}:
323 match result:
324 case Fail{error}:
325 IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error})
326 case Done{ByteBlockReader.End{}}:
327 do IO<Result<&1, &1, U32 & String, Cursor>>:
328 reopened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
329 ByteBlockReader.open(path)
330 verify.complete(bounds, reopened)
331 case Done{ByteBlockReader.Block{reader, _}}:
332 verify.loop(rest, path, VerifyNeed{reader, bounds})
333
334def open.budget(
335 path: String,
336 bounds: Bounds,
337 pair: Nat & ByteBlockReader.Reader
338) -> IO(Result<&1, &1, U32 & String, Cursor>):
339 match pair:
340 case (fuel, reader):
341 verify.loop(fuel, path, VerifyNeed{reader, bounds})
342
343def open.result(
344 path: String,
345 bounds: Bounds,
346 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
347) -> IO(Result<&1, &1, U32 & String, Cursor>):
348 match opened:
349 case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error})
350 case Done{reader}:
351 open.budget(path, bounds, ByteBlockReader.transition_budget(reader))
352
353def open.trusted.result(
354 bounds: Bounds,
355 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
356) -> IO(Result<&1, &1, U32 & String, Cursor>):
357 match opened:
358 case Fail{error}:
359 IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error})
360 case Done{reader}:
361 verify.complete(bounds, Done{reader})
362
363# Reopens a previously validated generation without rescanning every block.
364def open.trusted(
365 +path: String,
366 bounds: Bounds
367) -> IO(Result<&1, &1, U32 & String, Cursor>):
368 do IO<Result<&1, &1, U32 & String, Cursor>>:
369 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
370 ByteBlockReader.open(path)
371 open.trusted.result(bounds, opened)
372
373def measure.key.opened(
374 key: Bytes.Bytes,
375 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader>
376) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
377 match opened:
378 case Fail{error}:
379 IO.pure(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>,
380 Fail{error})
381 case Done{reader}: ByteBlockReader.measure_key(reader, key)
382
383# Handles measure key.
384def measure_key(path: String, key: Bytes.Bytes)
385 -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>):
386 do IO<Result<&1, &1, U32 & String, ByteWal.ValueMetadata>>:
387 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
388 ByteBlockReader.open(path)
389 measure.key.opened(key, opened)
390
391# Validates the complete SST before opening a second bounded scan pass.
392def open(
393 +path: String,
394 bounds: Bounds
395) ->
396 IO(Result<&1, &1, U32 & String, Cursor>):
397 do IO<Result<&1, &1, U32 & String, Cursor>>:
398 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <-
399 ByteBlockReader.open(path)
400 open.result(path, bounds, opened)
401
402def scan.start(pair: Nat & ByteBlockReader.Reader, bounds: Bounds) -> IO(Step):
403 match pair:
404 case (fuel, reader): scan.loop(fuel, ScanNeed{reader, bounds})
405
406# Returns the next bounded record, range skip, end, or terminal read error.
407def next(cursor: Cursor) -> IO(Step):
408 match cursor:
409 case Active{inner, bounds}:
410 match inner:
411 case ByteBlockReader.EntryReady{reader, entries}:
412 match entries:
413 case Con{entry, tail}:
414 scan.entry(entry, ByteBlockReader.EntryReady{reader, tail}, bounds)
415 case Nil{}:
416 scan.start(ByteBlockReader.transition_budget(reader), bounds)
417 case ByteBlockReader.EntryEnded{}:
418 IO.pure(Step, End{})
419 case ByteBlockReader.EntryClosed{}:
420 IO.pure(Step, End{})
421 case ByteBlockReader.EntryFailed{error}:
422 IO.pure(Step, Error{error})
423 case Ended{}: IO.pure(Step, End{})
424 case Closed{}: IO.pure(Step, End{})
425 case Failed{error}: IO.pure(Step, Error{error})
426
427# Advances the remaining scan to an inclusive lower byte key.
428def seek(cursor: Cursor, lower: Maybe<&1, Bytes.Bytes>) -> Cursor:
429 match cursor:
430 case Active{inner, bounds}:
431 match bounds:
432 case Bounds{_, upper, prefix}:
433 Active{inner, Bounds{lower, upper, prefix}}
434 case Ended{}: Ended{}
435 case Closed{}: Closed{}
436 case Failed{error}: Failed{error}
437
438# Releases the active file reader, preserving terminal cursors.
439def close(cursor: Cursor) -> IO(Result<&1, &1, U32 & String, Unit>):
440 match cursor:
441 case Active{inner, _}: ByteBlockReader.entry.close(inner)
442 case _:
443 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})