src/SstFileV2.bend
mylsm-lsm-store@0.2.0.0 · 438 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.2.0.0/src/Decimal.bend as Decimal
3import mylsm-lsm-store@0.2.0.0/src/Keys.bend as Keys
4import mylsm-lsm-store@0.2.0.0/src/MemTable.bend as MemTable
5import mylsm-lsm-store@0.2.0.0/src/Sstable.bend as Sstable
6import mylsm-lsm-store@0.2.0.0/src/SstChecksum.bend as SstChecksum
7
8# Compact SSTable v2:
9# S2;<level>;<count>;<entries>#<checksum>
10# P;<key-length>;<value-length>;<key><value>
11# D;<key-length>;<key>
12# The checksum is SstChecksum.digest(body) (hub SHA-256, hex).
13
14def max_file_chars() -> Nat:
15 Nat.add(4294967295n, 1n)
16
17def max_entries() -> Nat:
18 16777216n
19
20def max_key_chars() -> Nat:
21 16777216n
22
23def max_value_chars() -> Nat:
24 1073741824n
25
26def max_level() -> Nat:
27 255n
28
29def read_chunk_chars() -> Nat:
30 1048576n
31
32type Error is Data:
33 UnknownVersion{}
34 MalformedDecimal{}
35 DecimalOverflow{}
36 InvalidLevel{}
37 EntryCountExceeded{}
38 InvalidEntryTag{}
39 KeyLengthExceeded{}
40 ValueLengthExceeded{}
41 TruncatedHeader{}
42 TruncatedEntry{}
43 TruncatedChecksum{}
44 UnsortedOrDuplicate{}
45 ChecksumMismatch{}
46 TrailingData{}
47 FileTooLarge{}
48
49type Tag is Data:
50 PutTag{}
51 DelTag{}
52
53type Phase is Data:
54 NeedS{}
55 Need2{}
56 NeedHeaderSemi{}
57 ReadLevel{scan: Decimal.Scan}
58 ReadCount{scan: Decimal.Scan}
59 ReadTag{}
60 ReadTagSemi{tag: Tag}
61 ReadKeyLength{tag: Tag, scan: Decimal.Scan}
62 ReadValueLength{key_length: Nat, scan: Decimal.Scan}
63 ReadKey{tag: Tag, remaining: Nat, value_length: Nat, reversed: String}
64 ReadValue{key: String, remaining: Nat, reversed: String}
65 NeedHash{}
66 ReadChecksum{reversed: String}
67
68type Transition is Data:
69 Next{phase: Phase}
70 LevelRead{level: Nat}
71 CountRead{count: Nat}
72 EntryRead{entry: MemTable.Entry}
73 TransitionRejected{error: Error}
74
75type Decoder is Data:
76 Dec{
77 phase: Phase,
78 remaining_entries: Nat,
79 previous_key: Maybe<&2, String>,
80 reversed_entries: List<&2, MemTable.Entry>,
81 level: Nat,
82 declared_count: Nat,
83 body_rev: String,
84 consumed: Nat
85 }
86 DecoderRejected{error: Error}
87
88type ParseResult is Data:
89 ParseRejected{error: Error}
90 Parsed{table: Sstable.Table}
91
92# --- Serialization: reversed fragments followed by one String.join ---
93
94def entry_fragments(entries: List<&2, MemTable.Entry>, acc: List<&2, String>) -> List<&2, String>:
95 match entries:
96 case Nil{}:
97 acc
98 case Con{MemTable.Entry{+key, +val}, rest}:
99 match val:
100 case None{}:
101 entry_fragments(rest, Con{key, Con{";", Con{Decimal.render(String.length(key)), Con{"D;", acc}}}})
102 case Some{+value}:
103 entry_fragments(rest, Con{value, Con{key, Con{";", Con{Decimal.render(String.length(value)), Con{";", Con{Decimal.render(String.length(key)), Con{"P;", acc}}}}}}})
104
105def header_fragments(count: Nat, level: Nat) -> List<&2, String>:
106 Con{";", Con{Decimal.render(count), Con{";", Con{Decimal.render(level), Con{"S2;", Nil{}}}}}}
107
108def body(+entries: List<&2, MemTable.Entry>, level: Nat) -> String:
109 +header_reversed = header_fragments(List.length(&2, MemTable.Entry, entries), level)
110 String.join(List.reverse(&2, String, entry_fragments(entries, header_reversed)), "")
111
112def serialize(+entries: List<&2, MemTable.Entry>, level: Nat) -> String:
113 +encoded_body = body(entries, level)
114 encoded_body ++ "#" ++ SstChecksum.digest(encoded_body)
115
116# --- Pure one-character decoder transitions ---
117
118def expected_char(ok: Bool, next: Phase, error: Error) -> Transition:
119 match ok:
120 case True{}:
121 Next{next}
122 case False{}:
123 TransitionRejected{error}
124
125def decimal_error(error: Decimal.Error, overflow: Error) -> Transition:
126 match error:
127 case Decimal.Overflow{}:
128 TransitionRejected{overflow}
129 case _:
130 TransitionRejected{MalformedDecimal{}}
131
132def level_scan(scan: Decimal.Scan) -> Transition:
133 match scan:
134 case Decimal.Reading{value, started, leading_zero}:
135 Next{ReadLevel{Decimal.Reading{value, started, leading_zero}}}
136 case Decimal.Finished{value}:
137 LevelRead{value}
138 case Decimal.Failed{error}:
139 decimal_error(error, InvalidLevel{})
140
141def count_scan(scan: Decimal.Scan) -> Transition:
142 match scan:
143 case Decimal.Reading{value, started, leading_zero}:
144 Next{ReadCount{Decimal.Reading{value, started, leading_zero}}}
145 case Decimal.Finished{value}:
146 CountRead{value}
147 case Decimal.Failed{error}:
148 decimal_error(error, EntryCountExceeded{})
149
150def delete_key_length(value: Nat) -> Transition:
151 match value:
152 case 0n:
153 EntryRead{MemTable.Entry{"", None{}}}
154 case 1n+rest:
155 Next{ReadKey{DelTag{}, 1n+rest, 0n, ""}}
156
157def key_length_done(tag: Tag, value: Nat) -> Transition:
158 match tag:
159 case PutTag{}:
160 Next{ReadValueLength{value, Decimal.Reading{0n, False{}, False{}}}}
161 case DelTag{}:
162 delete_key_length(value)
163
164def key_length_scan(tag: Tag, scan: Decimal.Scan) -> Transition:
165 match scan:
166 case Decimal.Reading{value, started, leading_zero}:
167 Next{ReadKeyLength{tag, Decimal.Reading{value, started, leading_zero}}}
168 case Decimal.Failed{error}:
169 decimal_error(error, KeyLengthExceeded{})
170 case Decimal.Finished{value}:
171 key_length_done(tag, value)
172
173def put_lengths(key_length: Nat, value_length: Nat) -> Transition:
174 match key_length:
175 case 0n:
176 match value_length:
177 case 0n:
178 EntryRead{MemTable.Entry{"", Some{""}}}
179 case 1n+v:
180 Next{ReadValue{"", 1n+v, ""}}
181 case 1n+k:
182 Next{ReadKey{PutTag{}, 1n+k, value_length, ""}}
183
184def value_length_scan(key_length: Nat, scan: Decimal.Scan) -> Transition:
185 match scan:
186 case Decimal.Reading{value, started, leading_zero}:
187 Next{ReadValueLength{key_length, Decimal.Reading{value, started, leading_zero}}}
188 case Decimal.Failed{error}:
189 decimal_error(error, ValueLengthExceeded{})
190 case Decimal.Finished{value}:
191 put_lengths(key_length, value)
192
193def tag_del(is_del: Bool) -> Transition:
194 match is_del:
195 case True{}:
196 Next{ReadTagSemi{DelTag{}}}
197 case False{}:
198 TransitionRejected{InvalidEntryTag{}}
199
200def tag_put(is_put: Bool, ch: Char) -> Transition:
201 match is_put:
202 case True{}:
203 Next{ReadTagSemi{PutTag{}}}
204 case False{}:
205 tag_del(Char.is_eq(ch, 'D'))
206
207def tag_step(+ch: Char) -> Transition:
208 tag_put(Char.is_eq(ch, 'P'), ch)
209
210def key_last(tag: Tag, value_length: Nat, reversed: String) -> Transition:
211 match tag:
212 case DelTag{}:
213 EntryRead{MemTable.Entry{String.reverse(reversed), None{}}}
214 case PutTag{}:
215 match value_length:
216 case 0n:
217 EntryRead{MemTable.Entry{String.reverse(reversed), Some{""}}}
218 case 1n+v:
219 Next{ReadValue{String.reverse(reversed), 1n+v, ""}}
220
221def key_step(tag: Tag, remaining: Nat, value_length: Nat, reversed: String, ch: Char) -> Transition:
222 match remaining:
223 case 0n:
224 TransitionRejected{TruncatedEntry{}}
225 case 1n+rest:
226 match rest:
227 case 0n:
228 key_last(tag, value_length, SCon{ch, reversed})
229 case 1n+more:
230 Next{ReadKey{tag, 1n+more, value_length, SCon{ch, reversed}}}
231
232def value_step(key: String, remaining: Nat, reversed: String, ch: Char) -> Transition:
233 match remaining:
234 case 0n:
235 TransitionRejected{TruncatedEntry{}}
236 case 1n+rest:
237 match rest:
238 case 0n:
239 EntryRead{MemTable.Entry{key, Some{String.reverse(SCon{ch, reversed})}}}
240 case 1n+more:
241 Next{ReadValue{key, 1n+more, SCon{ch, reversed}}}
242
243def checksum_step(+reversed: String, +ch: Char) -> Transition:
244 Next{ReadChecksum{SCon{ch, reversed}}}
245
246def phase_step(phase: Phase, +ch: Char) -> Transition:
247 match phase:
248 case NeedS{}:
249 expected_char(Char.is_eq(ch, 'S'), Need2{}, UnknownVersion{})
250 case Need2{}:
251 expected_char(Char.is_eq(ch, '2'), NeedHeaderSemi{}, UnknownVersion{})
252 case NeedHeaderSemi{}:
253 expected_char(Char.is_eq(ch, ';'), ReadLevel{Decimal.Reading{0n, False{}, False{}}}, UnknownVersion{})
254 case ReadLevel{scan}:
255 level_scan(Decimal.scan_step(ch, scan, max_level(), Decimal.Delimited{';'}))
256 case ReadCount{scan}:
257 count_scan(Decimal.scan_step(ch, scan, max_entries(), Decimal.Delimited{';'}))
258 case ReadTag{}:
259 tag_step(ch)
260 case ReadTagSemi{tag}:
261 expected_char(Char.is_eq(ch, ';'), ReadKeyLength{tag, Decimal.Reading{0n, False{}, False{}}}, InvalidEntryTag{})
262 case ReadKeyLength{tag, scan}:
263 key_length_scan(tag, Decimal.scan_step(ch, scan, max_key_chars(), Decimal.Delimited{';'}))
264 case ReadValueLength{key_length, scan}:
265 value_length_scan(key_length, Decimal.scan_step(ch, scan, max_value_chars(), Decimal.Delimited{';'}))
266 case ReadKey{tag, remaining, value_length, reversed}:
267 key_step(tag, remaining, value_length, reversed, ch)
268 case ReadValue{key, remaining, reversed}:
269 value_step(key, remaining, reversed, ch)
270 case NeedHash{}:
271 expected_char(Char.is_eq(ch, '#'), ReadChecksum{SNil{}}, TruncatedChecksum{})
272 case ReadChecksum{reversed}:
273 checksum_step(reversed, ch)
274
275def phase_bodies(phase: Phase) -> Bool:
276 match phase:
277 case NeedHash{}:
278 False{}
279 case ReadChecksum{reversed}:
280 False{}
281 case _:
282 True{}
283
284def select_body(keeps: Bool, body_rev: String, ch: Char) -> String:
285 match keeps:
286 case True{}:
287 SCon{ch, body_rev}
288 case False{}:
289 body_rev
290
291# --- Applying transitions enforces exact count and strict key order ---
292
293def decoder_next(phase: Phase, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder:
294 Dec{phase, remaining, previous, entries, level, declared, body_rev, consumed}
295
296def count_transition(count: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, body_rev: String, consumed: Nat) -> Decoder:
297 match count:
298 case 0n:
299 decoder_next(NeedHash{}, 0n, previous, entries, level, 0n, body_rev, consumed)
300 case 1n++rest:
301 decoder_next(ReadTag{}, 1n+rest, previous, entries, level, 1n+rest, body_rev, consumed)
302
303def ordered_entry(entry: MemTable.Entry, remaining: Nat, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder:
304 match entry:
305 case MemTable.Entry{+key, +val}:
306 match remaining:
307 case 0n:
308 DecoderRejected{TrailingData{}}
309 case 1n+rest:
310 match rest:
311 case 0n:
312 decoder_next(NeedHash{}, 0n, Some{key}, Con{MemTable.Entry{key, val}, entries}, level, declared, body_rev, consumed)
313 case 1n+more:
314 decoder_next(ReadTag{}, 1n+more, Some{key}, Con{MemTable.Entry{key, val}, entries}, level, declared, body_rev, consumed)
315
316def order_decision(cmp: Cmp, entry: MemTable.Entry, remaining: Nat, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder:
317 match cmp:
318 case LT{}:
319 ordered_entry(entry, remaining, entries, level, declared, body_rev, consumed)
320 case _:
321 DecoderRejected{UnsortedOrDuplicate{}}
322
323def entry_transition(+entry: MemTable.Entry, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder:
324 match entry:
325 case MemTable.Entry{+key, +val}:
326 match previous:
327 case None{}:
328 ordered_entry(MemTable.Entry{key, val}, remaining, entries, level, declared, body_rev, consumed)
329 case Some{+old}:
330 order_decision(Keys.cmp(old, key), MemTable.Entry{key, val}, remaining, entries, level, declared, body_rev, consumed)
331
332def apply_transition(transition: Transition, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder:
333 match transition:
334 case TransitionRejected{error}:
335 DecoderRejected{error}
336 case Next{phase}:
337 decoder_next(phase, remaining, previous, entries, level, declared, body_rev, consumed)
338 case LevelRead{new_level}:
339 decoder_next(ReadCount{Decimal.Reading{0n, False{}, False{}}}, remaining, previous, entries, new_level, declared, body_rev, consumed)
340 case CountRead{count}:
341 count_transition(count, previous, entries, level, body_rev, consumed)
342 case EntryRead{entry}:
343 entry_transition(entry, remaining, previous, entries, level, declared, body_rev, consumed)
344
345def decoder_step_live(+phase: Phase, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat, +ch: Char) -> Decoder:
346 +next_body = select_body(phase_bodies(phase), body_rev, ch)
347 apply_transition(phase_step(phase, ch), remaining, previous, entries, level, declared, next_body, Nat.add(consumed, 1n))
348
349def decoder_step_bounded(
350 within: Bool,
351 phase: Phase,
352 remaining: Nat,
353 previous: Maybe<&2, String>,
354 entries: List<&2, MemTable.Entry>,
355 level: Nat,
356 declared: Nat,
357 body_rev: String,
358 consumed: Nat,
359 ch: Char,
360) -> Decoder:
361 match within:
362 case False{}:
363 DecoderRejected{FileTooLarge{}}
364 case True{}:
365 decoder_step_live(phase, remaining, previous, entries, level, declared, body_rev, consumed, ch)
366
367def decoder_step(decoder: Decoder, ch: Char) -> Decoder:
368 match decoder:
369 case DecoderRejected{error}:
370 DecoderRejected{error}
371 case Dec{phase, remaining_entries, previous_key, reversed_entries, level, declared_count, body_rev, +consumed}:
372 decoder_step_bounded(Nat.is_lt(consumed, max_file_chars()), phase, remaining_entries, previous_key, reversed_entries, level, declared_count, body_rev, consumed, ch)
373
374# --- EOF validation and table construction ---
375
376def checksum_decision(equal: Bool, entries: List<&2, MemTable.Entry>, level: Nat) -> ParseResult:
377 match equal:
378 case False{}:
379 ParseRejected{ChecksumMismatch{}}
380 case True{}:
381 Parsed{Sstable.from_sorted_unique(List.reverse(&2, MemTable.Entry, entries), level)}
382
383def checksum_value(+claimed_rev: String, entries: List<&2, MemTable.Entry>, level: Nat, +body_rev: String) -> ParseResult:
384 checksum_decision(SstChecksum.verify(String.reverse(body_rev), String.reverse(claimed_rev)), entries, level)
385
386def finish_decoder(decoder: Decoder) -> ParseResult:
387 match decoder:
388 case DecoderRejected{error}:
389 ParseRejected{error}
390 case Dec{phase, remaining_entries, previous_key, reversed_entries, level, declared_count, body_rev, consumed}:
391 match phase:
392 case ReadChecksum{reversed}:
393 checksum_value(reversed, reversed_entries, level, body_rev)
394 case NeedS{}:
395 ParseRejected{TruncatedHeader{}}
396 case Need2{}:
397 ParseRejected{TruncatedHeader{}}
398 case NeedHeaderSemi{}:
399 ParseRejected{TruncatedHeader{}}
400 case ReadLevel{scan}:
401 ParseRejected{TruncatedHeader{}}
402 case ReadCount{scan}:
403 ParseRejected{TruncatedHeader{}}
404 case NeedHash{}:
405 ParseRejected{TruncatedChecksum{}}
406 case _:
407 ParseRejected{TruncatedEntry{}}
408
409def decode_go(rest: String, decoder: Decoder) -> ParseResult:
410 match rest:
411 case SNil{}:
412 finish_decoder(decoder)
413 case SCon{c, t}:
414 decode_go(t, decoder_step(decoder, c))
415
416def parse_result(str: String) -> ParseResult:
417 decode_go(str, Dec{NeedS{}, 0n, None{}, Nil{}, 0n, 0n, "", 0n})
418
419def parse_wrap(result: ParseResult) -> Maybe<&2, Sstable.Table>:
420 match result:
421 case ParseRejected{error}:
422 None{}
423 case Parsed{table}:
424 Some{table}
425
426def parse(str: String) -> Maybe<&2, Sstable.Table>:
427 parse_wrap(parse_result(str))
428
429# Useful pure projections for laws and streaming-state agreement.
430def parser_decides(_result: ParseResult) -> Bool:
431 True{}
432
433def error_of(result: ParseResult) -> Maybe<&2, Error>:
434 match result:
435 case ParseRejected{error}:
436 Some{error}
437 case Parsed{table}:
438 None{}