src/Compact.bend source

src/Compact.bend on the hub · documented module

import Baseimport ./Keys.bend as Keysimport ./MemTable.bend as MemTableimport ./Sstable.bend as Sstableimport ./SortedRun.bend as SortedRunimport ./SstFile.bend as SstFileimport ./Wal.bend as Walimport ./Manifest.bend as Manifestimport ./Db.bend as Db# Tiered compaction (Task 9): L0-drain + L1 overlap-closure.## Invariants (Task 7/8 read path depends on them):# - L0-DRAIN: every compaction consumes ALL of L0 (post L0 = []), so every#   L0 table is strictly newer than every L1+ table.# - L1-DISJOINT: output is disjoint from every remaining L1 table, via#   overlap-closure over the final hull (rounds = len(l1)+1 unconditionally;#   rounds past fixpoint are idempotent, so no done-flag is needed).# - READ-ORDER MERGE: inputs merge in exact read order (L0 stored order ++#   absorbed-L1 stored order), so first-wins == read first-match.# - TOMBSTONE-SAFE: all winning tombstones are retained at this checkpoint.#   This conservative policy prevents resurrection without requiring zero-output#   publication support.# Trigger: L0 count > 4 (spec: backpressure stall at 8 = 2T, Task 10).# CrashPoint calls are test-only host effects. Unset or unequal checkpoints are# successful no-ops; signal delivery and filesystem ordering are not Bend proofs.# Soundness sketch (spec law 11): keys resolving outside the closure are# untouched (a key in both closure-inputs and a remainder table would put# it in the hull and in a hull-disjoint range — contradiction); keys inside# resolve to the same newest version (subset order == read order).# Handle ov and in the level compaction.def ov_and(lhs: Bool, rhs: Bool) -> Bool:  match lhs:    case True{}:      rhs    case False{}:      False{}# Handle le cmp in the level compaction.def le_cmp(ord: Cmp) -> Bool:  match ord:    case LT{}:      True{}    case EQ{}:      True{}    case GT{}:      False{}# Handle ov12 in the level compaction.def ov12(fa: Maybe<&2, String>, lb: Maybe<&2, String>) -> Bool:  match fa lb:    case None{} _:      False{}    case _ None{}:      False{}    case Some{x} Some{y}:      le_cmp(Keys.cmp(x, y))# Return the table overlap for the level compaction.def table_overlap(+ta: Sstable.Table, +tb: Sstable.Table) -> Bool:  ov_and(ov12(Sstable.table_smallest(ta), Sstable.table_largest(tb)), ov12(Sstable.table_smallest(tb), Sstable.table_largest(ta)))# Hull bound folds (pump+fuel+leaf; single-Maybe state, no pairs).def pick_lo(ord: Cmp, k1: String, k2: String) -> Maybe<&2, String>:  match ord:    case LT{}:      Some{k1}    case EQ{}:      Some{k1}    case GT{}:      Some{k2}# Handle min step in the level compaction.def min_step(fm: Maybe<&2, String>, cur: Maybe<&2, String>) -> Maybe<&2, String>:  match fm cur:    case None{} _:      cur    case _ None{}:      fm    case Some{+a} Some{+b}:      pick_lo(Keys.cmp(a, b), a, b)# Handle min go in the level compaction.def min_go(fuel: Nat, +cur: Maybe<&2, String>, tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:  match fuel:    case 0n:      cur    case 1n+f:      match tabs:        case Nil{}:          cur        case Con{h, t}:          min_go(f, min_step(Sstable.table_smallest(h), cur), t)# Handle min first in the level compaction.def min_first(+tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:  min_go(List.length(&2, Sstable.Table, tabs), None{}, tabs)# Select hi for the level compaction.def pick_hi(ord: Cmp, k1: String, k2: String) -> Maybe<&2, String>:  match ord:    case LT{}:      Some{k2}    case EQ{}:      Some{k2}    case GT{}:      Some{k1}# Handle max step in the level compaction.def max_step(lm: Maybe<&2, String>, cur: Maybe<&2, String>) -> Maybe<&2, String>:  match lm cur:    case None{} _:      cur    case _ None{}:      lm    case Some{+a} Some{+b}:      pick_hi(Keys.cmp(a, b), a, b)# Handle max go in the level compaction.def max_go(fuel: Nat, +cur: Maybe<&2, String>, tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:  match fuel:    case 0n:      cur    case 1n+f:      match tabs:        case Nil{}:          cur        case Con{h, t}:          max_go(f, max_step(Sstable.table_largest(h), cur), t)# Handle max last in the level compaction.def max_last(+tabs: List<&2, Sstable.Table>) -> Maybe<&2, String>:  max_go(List.length(&2, Sstable.Table, tabs), None{}, tabs)# Overlap of one table against a hull (bounds as params).def hull_step(keep: Bool, +tab: Sstable.Table, +acc: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:  match keep:    case True{}:      Con{tab, acc}    case False{}:      acc# Handle hull neg in the level compaction.def hull_neg(keep: Bool, +tab: Sstable.Table, +acc: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:  match keep:    case True{}:      acc    case False{}:      Con{tab, acc}# Handle ov hull in the level compaction.def ov_hull(+tab: Sstable.Table, +lo: Maybe<&2, String>, +hi: Maybe<&2, String>) -> Bool:  ov_and(ov12(Sstable.table_smallest(tab), hi), ov12(lo, Sstable.table_largest(tab)))# Handle filt go in the level compaction.def filt_go(  fuel: Nat,  +lo: Maybe<&2, String>,  +hi: Maybe<&2, String>,  +acc: List<&2, Sstable.Table>,  rest: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:  match fuel:    case 0n:      List.reverse(&2, Sstable.Table, acc)    case 1n+f:      match rest:        case Nil{}:          List.reverse(&2, Sstable.Table, acc)        case Con{+h, t}:          filt_go(f, lo, hi, hull_step(ov_hull(h, lo, hi), h, acc), t)# Handle filter pos in the level compaction.def filter_pos(  +rest: List<&2, Sstable.Table>,  +lo: Maybe<&2, String>,  +hi: Maybe<&2, String>) -> List<&2, Sstable.Table>:  filt_go(List.length(&2, Sstable.Table, rest), lo, hi, Nil{}, rest)# Handle filt neg go in the level compaction.def filt_neg_go(  fuel: Nat,  +lo: Maybe<&2, String>,  +hi: Maybe<&2, String>,  +acc: List<&2, Sstable.Table>,  rest: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:  match fuel:    case 0n:      List.reverse(&2, Sstable.Table, acc)    case 1n+f:      match rest:        case Nil{}:          List.reverse(&2, Sstable.Table, acc)        case Con{+h, t}:          filt_neg_go(f, lo, hi, hull_neg(ov_hull(h, lo, hi), h, acc), t)# Handle filter neg in the level compaction.def filter_neg(  +rest: List<&2, Sstable.Table>,  +lo: Maybe<&2, String>,  +hi: Maybe<&2, String>) -> List<&2, Sstable.Table>:  filt_neg_go(List.length(&2, Sstable.Table, rest), lo, hi, Nil{}, rest)# The production policy is conservative: every winning tombstone is retained# regardless of the lower-level shadow list.def drop_none(  +key: String,  +_shadow: List<&2, MemTable.Entry>,  +acc: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:  Con{MemTable.Entry{key, None{}}, acc}# Handle drop go in the level compaction.def drop_go(  fuel: Nat,  +shadow: List<&2, MemTable.Entry>,  +acc: List<&2, MemTable.Entry>,  xs: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:  match fuel:    case 0n:      List.reverse(&2, MemTable.Entry, acc)    case 1n+f:      match xs:        case Nil{}:          List.reverse(&2, MemTable.Entry, acc)        case Con{MemTable.Entry{k, v}, t}:          match v:            case None{}:              drop_go(f, shadow, drop_none(k, shadow, acc), t)            case Some{s}:              drop_go(f, shadow, Con{MemTable.Entry{k, Some{s}}, acc}, t)# Handle shadow drop in the level compaction.def shadow_drop(+xs: List<&2, MemTable.Entry>, +shadow: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:  drop_go(List.length(&2, MemTable.Entry, xs), shadow, Nil{}, xs)# Preserve each table's strict sorted-run boundary and stored read order.def table_runs(tabs: List<&2, Sstable.Table>) -> List<&2, List<&2, MemTable.Entry>>:  match tabs:    case Nil{}:      Nil{}    case Con{h, rest}:      Con{Db.table_entries(h), table_runs(rest)}# Production output: merge L0 in chronological stored order, merge absorbed L1# in its original stored order, then merge L0 as strictly newer than L1.# Tombstones are retained, so a triggered compaction publishes exactly one table.def finish_runs(  +l0_runs: List<&2, List<&2, MemTable.Entry>>,  +l1_runs: List<&2, List<&2, MemTable.Entry>>,  +rest: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:  merged_l0 merged_l1 = SortedRun.merge_many_newest(l0_runs) SortedRun.merge_many_newest(l1_runs)  +merged = SortedRun.merge_many_newest(Con{merged_l0, Con{merged_l1, Nil{}}})  Con{Sstable.from_sorted_unique(shadow_drop(merged, Nil{}), 1n), rest}# Closure rounds retain the original L1 list so the final absorbed runs are# selected in exact stored read order, independent of the round that found them.def closed_runs_go(  major: Nat,  +lo: Maybe<&2, String>,  +hi: Maybe<&2, String>,  +l0_runs: List<&2, List<&2, MemTable.Entry>>,  +all_l1: List<&2, Sstable.Table>,  +rest: List<&2, Sstable.Table>,) -> List<&2, Sstable.Table>:  match major:    case 0n:      finish_runs(l0_runs, table_runs(filter_pos(all_l1, lo, hi)), rest)    case 1n+m:      +abs_new +rem_new = filter_pos(rest, lo, hi) filter_neg(rest, lo, hi)      closed_runs_go(m, min_step(min_first(abs_new), lo), max_step(max_last(abs_new), hi), l0_runs, all_l1, rem_new)# Top dispatch (all matches on params; count gate inside).def compact_snd(  +l0: List<&2, Sstable.Table>,  rest1: List<&2, List<&2, Sstable.Table>>) -> List<&2, List<&2, Sstable.Table>>:  match rest1:    case Nil{}:      Con{Nil{}, Con{finish_runs(table_runs(l0), Nil{}, Nil{}), Nil{}}}    case Con{+l1, +below}:      Con{Nil{}, Con{closed_runs_go(Nat.add(List.length(&2, Sstable.Table, l1), 1n), min_first(l0), max_last(l0), table_runs(l0), l1, l1), below}}# Compact dec for the level compaction.def compact_dec(  gt: Bool,  +l0: List<&2, Sstable.Table>,  +rest1: List<&2, List<&2, Sstable.Table>>) -> List<&2, List<&2, Sstable.Table>>:  match gt:    case True{}:      compact_snd(l0, rest1)    case False{}:      Con{l0, rest1}# Compact levels for the level compaction.def compact_levels(levels: List<&2, List<&2, Sstable.Table>>) -> List<&2, List<&2, Sstable.Table>>:  match levels:    case Nil{}:      Nil{}    case Con{+l0, rest1}:      compact_dec(Nat.is_lt(4n, List.length(&2, Sstable.Table, l0)), l0, rest1)# --- IO helpers (Task 9b): remainder-membership, name splits, level surgery.# Remainder test via ranges (sound by old-L1 disjointness, spec law 12):# in a disjoint family, overlap with a remainder table means identity.# Combine leaf for the level compaction.def or_leaf(+hit: Bool, +acc: Bool) -> Bool:  match hit:    case True{}:      True{}    case False{}:      acc# Handle in go in the level compaction.def in_go(fuel: Nat, +tab: Sstable.Table, +acc: Bool, rem: List<&2, Sstable.Table>) -> Bool:  match fuel:    case 0n:      acc    case 1n+f:      match rem:        case Nil{}:          acc        case Con{h, tt}:          in_go(f, tab, or_leaf(table_overlap(tab, h), acc), tt)# Handle in rem in the level compaction.def in_rem(+tab: Sstable.Table, +rem: List<&2, Sstable.Table>) -> Bool:  in_go(List.length(&2, Sstable.Table, rem), tab, False{}, rem)# Handle abs pick in the level compaction.def abs_pick(+inr: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>:  match inr:    case True{}:      acc    case False{}:      Con{name, acc}# Handle abs go in the level compaction.def abs_go(  fuel: Nat,  +rem: List<&2, Sstable.Table>,  +acc: List<&2, String>,  tabs: List<&2, Sstable.Table>,  names: List<&2, String>) -> List<&2, String>:  match fuel:    case 0n:      List.reverse(&2, String, acc)    case 1n+f:      match tabs names:        case Nil{} Nil{}:          List.reverse(&2, String, acc)        case Nil{} Con{nh, nt}:          List.reverse(&2, String, acc)        case Con{th, tt} Nil{}:          List.reverse(&2, String, acc)        case Con{th, tt} Con{nh, nt}:          abs_go(f, rem, abs_pick(in_rem(th, rem), nh, acc), tt, nt)# Handle abs names in the level compaction.def abs_names(  +tabs: List<&2, Sstable.Table>,  +names: List<&2, String>,  +rem: List<&2, Sstable.Table>) -> List<&2, String>:  abs_go(List.length(&2, Sstable.Table, tabs), rem, Nil{}, tabs, names)# Handle rem pick in the level compaction.def rem_pick(+inr: Bool, +name: String, +acc: List<&2, String>) -> List<&2, String>:  match inr:    case True{}:      Con{name, acc}    case False{}:      acc# Handle rem go in the level compaction.def rem_go(  fuel: Nat,  +rem: List<&2, Sstable.Table>,  +acc: List<&2, String>,  tabs: List<&2, Sstable.Table>,  names: List<&2, String>) -> List<&2, String>:  match fuel:    case 0n:      List.reverse(&2, String, acc)    case 1n+f:      match tabs names:        case Nil{} Nil{}:          List.reverse(&2, String, acc)        case Nil{} Con{nh, nt}:          List.reverse(&2, String, acc)        case Con{th, tt} Nil{}:          List.reverse(&2, String, acc)        case Con{th, tt} Con{nh, nt}:          rem_go(f, rem, rem_pick(in_rem(th, rem), nh, acc), tt, nt)# Handle rem names in the level compaction.def rem_names(  +tabs: List<&2, Sstable.Table>,  +names: List<&2, String>,  +rem: List<&2, Sstable.Table>) -> List<&2, String>:  rem_go(List.length(&2, Sstable.Table, tabs), rem, Nil{}, tabs, names)# --- Closed-vector laws live in laws/Compact.bend (spec laws 6, 11, 12) ---# --- IO chain (Task 9b): output file + manifest rewrite + input removal.# Crash discipline mirrors flush: output durable, then manifest (tmp +# rename + dir fsync), then input files become orphans (recovery ignores# files absent from the manifest; Task 12 sweeps them). Manifest/table# drift (exact serialized Manifest identity mismatch) aborts fail-closed;# names stay positionally aligned with tables (output head, remainder tail).# Handle l1hd in the level compaction.def l1hd(r1: List<&2, List<&2, Sstable.Table>>) -> List<&2, Sstable.Table>:  match r1:    case Nil{}:      Nil{}    case Con{l1, below}:      l1# Handle l1list in the level compaction.def l1list(lvls: List<&2, List<&2, Sstable.Table>>) -> List<&2, Sstable.Table>:  match lvls:    case Nil{}:      Nil{}    case Con{l0, r1}:      l1hd(r1)# Handle l0list in the level compaction.def l0list(lvls: List<&2, List<&2, Sstable.Table>>) -> List<&2, Sstable.Table>:  match lvls:    case Nil{}:      Nil{}    case Con{l0, r1}:      l0# Handle hd tbl in the level compaction.def hd_tbl(tabs: List<&2, Sstable.Table>) -> Sstable.Table:  match tabs:    case Nil{}:      Sstable.from_sorted_unique(Nil{}, 1n)    case Con{h, t}:      h# Process the remaining of for the level compaction.def tail_of(tabs: List<&2, Sstable.Table>) -> List<&2, Sstable.Table>:  match tabs:    case Nil{}:      Nil{}    case Con{h, t}:      t# Handle ml0 in the level compaction.def ml0(mfst: Manifest.Manifest) -> List<&2, String>:  match mfst:    case Manifest.M{lvs}:      match lvs:        case Nil{}:          Nil{}        case Con{l0n, r1}:          l0n# Handle ml1b in the level compaction.def ml1b(r1: List<&2, List<&2, String>>) -> List<&2, String>:  match r1:    case Nil{}:      Nil{}    case Con{l1n, below}:      l1n# Handle ml1 in the level compaction.def ml1(mfst: Manifest.Manifest) -> List<&2, String>:  match mfst:    case Manifest.M{lvs}:      match lvs:        case Nil{}:          Nil{}        case Con{l0n, r1}:          ml1b(r1)# Handle mbelb in the level compaction.def mbelb(r1: List<&2, List<&2, String>>) -> List<&2, List<&2, String>>:  match r1:    case Nil{}:      Nil{}    case Con{l1n, below}:      below# Handle mbelow in the level compaction.def mbelow(mfst: Manifest.Manifest) -> List<&2, List<&2, String>>:  match mfst:    case Manifest.M{lvs}:      match lvs:        case Nil{}:          Nil{}        case Con{l0n, r1}:          mbelb(r1)# Handle mfst new in the level compaction.def mfst_new(+mfst: Manifest.Manifest, +remnames: List<&2, String>, +outname: String) -> Manifest.Manifest:  Manifest.M{Con{Nil{}, Con{Con{outname, remnames}, mbelow(mfst)}}}# Handle prefix go in the level compaction.def prefix_go(names: List<&2, String>, +dir: String, +acc: List<&2, String>) -> List<&2, String>:  match names:    case Nil{}:      List.reverse(&2, String, acc)    case Con{h, t}:      prefix_go(t, dir, Con{dir ++ "/" ++ h, acc})# Handle prefix in the level compaction.def prefix(+names: List<&2, String>, +dir: String) -> List<&2, String>:  prefix_go(names, dir, Nil{})