1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ../Keys.bend as Keys
4	import ./ByteRange.bend as ByteRange
5	import ./RangeCursor.bend as RangeCursor
6	import ./ScanMerge.bend as ScanMerge
7	import ./SstCursor.bend as SstCursor
8	import ./VisibleSeekPolicy.bend as VisibleSeekPolicy
9	import ./CursorPreviewPolicy.bend as PreviewPolicy
10	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
11	
12	# Retains one SST head and one ordered delta source between reads.
13	type 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.
22	type 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.
29	type 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.
35	type 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.
40	type 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.
45	type 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.
52	type 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.
58	type 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.
64	def 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	
71	def 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	
76	def 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	
83	def 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	
93	def 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	
99	def 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.
111	def 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	
119	def 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	
128	def 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	
136	def 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	
147	def 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	
154	def 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	
159	def 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	
166	def 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	
182	def 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	
196	def 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	
208	def 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	
219	def 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	
231	def 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	
241	def 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	
266	def compare.order.only(result: Bytes.Bytes & Bytes.Bytes & Cmp) -> Cmp:
267	  match result:
268	    case (_, _, order): order
269	
270	def 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	
273	def 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	
285	def 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	
298	def 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.
311	def 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	
321	def 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	
332	def 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.
339	def 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	
348	def 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	
373	def 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.
384	def 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	
392	def 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	
406	def 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.
414	def 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.
430	def 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	
436	def 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	
445	def 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	
456	def 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	
465	def 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	
474	def 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	
487	def preview.item(entry: ByteMemTable.Entry, cursor: Cursor) -> Preview:
488	  preview.item.result(preview.item.length(entry), cursor)
489	
490	def 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.
523	def preview(cursor: Cursor) -> IO(Preview):
524	  preview.loop(4096n, PreviewNeed{cursor})