src/SstReadPolicy.bend

mylsm-lsm-store@0.6.0.1 · 46 lines · 7 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
3
4# Classifies a single SST size against format and caller byte limits.
5type Decision is Data:
6  TooSmall{}
7  WithinBudget{}
8  OverBudget{}
9
10# Carries the next aggregate byte total only when admission succeeds.
11type TotalDecision is Data:
12  TotalAdmitted{next_total: Nat}
13  TotalRejected{}
14
15# Accepts a table only when its size is within the configured file budget.
16def within_budget(size: Nat, max_bytes: Nat) -> Bool:
17  Nat.is_le(size, max_bytes)
18
19def classify.copies(pair: Nat & Nat, max_bytes: Nat) -> Decision:
20  match pair:
21    case (minimum_check, budget_check):
22      Bool.pick(Decision, Nat.is_le(60n, minimum_check),
23        Bool.pick(Decision, within_budget(budget_check, max_bytes),
24          WithinBudget{}, OverBudget{}), TooSmall{})
25
26# Separates undersized corruption from a valid but oversized table.
27def classify(size: Nat, max_bytes: Nat) -> Decision:
28  classify.copies(QueuePolicy.nat.clone.pair(size), max_bytes)
29
30def total.copies(
31  totals: Nat & Nat,
32  sizes: Nat & Nat,
33  max_bytes: Nat
34) -> TotalDecision:
35  match totals:
36    case (checked_total, next_total):
37      match sizes:
38        case (checked_size, next_size):
39          Bool.pick(TotalDecision,
40            Nat.is_le(Nat.add(checked_total, checked_size), max_bytes),
41            TotalAdmitted{Nat.add(next_total, next_size)}, TotalRejected{})
42
43# Checks a file against the remaining aggregate budget before decoding.
44def total_decision(total: Nat, size: Nat, max_bytes: Nat) -> TotalDecision:
45  total.copies(QueuePolicy.nat.clone.pair(total),
46    QueuePolicy.nat.clone.pair(size), max_bytes)