mylsm.bend

mylsm-lsm-store@0.3.2.0 · 160 lines · 36 definitions

1import Base
2import mylsm-lsm-store@0.3.2.0/src/Keys.bend as Keys
3import mylsm-lsm-store@0.3.2.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.3.2.0/src/SortedRun.bend as SortedRun
5import mylsm-lsm-store@0.3.2.0/src/Sstable.bend as Sstable
6import mylsm-lsm-store@0.3.2.0/src/SstFile.bend as SstFile
7import mylsm-lsm-store@0.3.2.0/src/Wal.bend as Wal
8import mylsm-lsm-store@0.3.2.0/src/Manifest.bend as Manifest
9import mylsm-lsm-store@0.3.2.0/src/MergeIter.bend as MergeIter
10import mylsm-lsm-store@0.3.2.0/src/Db.bend as Db
11
12# --- Level 2: part control (1:1 delegates) ---
13def cmp(+s1: String, +s2: String) -> Cmp:
14  Keys.cmp(s1, s2)
15
16# Key equality: true exactly for identical strings.
17def eq(+s1: String, +s2: String) -> Bool:
18  Keys.eq(s1, s2)
19
20# Empty MemTable: no entries, count zero.
21def mem_empty() -> MemTable.MemTable:
22  MemTable.empty()
23
24# MemTable with one live version prepended (newest-first log).
25def mem_put(tab: MemTable.MemTable, key: String, val: String) -> MemTable.MemTable:
26  MemTable.put(tab, key, val)
27
28# MemTable with one tombstone prepended; hides older versions on read.
29def mem_del(tab: MemTable.MemTable, key: String) -> MemTable.MemTable:
30  MemTable.del(tab, key)
31
32# Newest-first read; a tombstone answers None and hides older versions.
33def mem_get(tab: MemTable.MemTable, +key: String) -> Maybe<&2, String>:
34  MemTable.get(tab, key)
35
36# Number of log entries, live and tombstoned.
37def mem_count(tab: MemTable.MemTable) -> Nat:
38  MemTable.count(tab)
39
40# Canonicalize newest-first entries into a strict ordered run.
41def sort_newest(entries: List<&2, MemTable.Entry>) -> List<&2, MemTable.Entry>:
42  SortedRun.sort_newest(entries)
43
44# Range scan over a merged run: lo <= key < hi, tombstones dropped.
45def range_scan(+merged: List<&2, MemTable.Entry>, lo: String, hi: String) -> List<&2, MemTable.Entry>:
46  MergeIter.scan(merged, lo, hi)
47
48# Build an SSTable, canonicalizing entries newest-first.
49def sst_build(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat) -> Sstable.Table:
50  Sstable.build(entries, level, est_keys)
51
52# Build an SSTable trusting an already strict, unique run (no sorting).
53def sst_from_sorted_unique(+entries: List<&2, MemTable.Entry>, level: Nat) -> Sstable.Table:
54  Sstable.from_sorted_unique(entries, level)
55
56# Build an SSTable from a sorted run with an explicit key estimate.
57def sst_build_sorted(+entries: List<&2, MemTable.Entry>, level: Nat, est_keys: Nat) -> Sstable.Table:
58  Sstable.build_sorted(entries, level, est_keys)
59
60# Encode a WAL batch to its exact on-disk bytes.
61def wal_encode(batch: Wal.Batch) -> String:
62  Wal.encode(batch)
63
64# Decode WAL bytes; None on corrupt or truncated input (fail-closed).
65def wal_decode(+str: String) -> Maybe<&2, Wal.Batch>:
66  Wal.decode(str)
67
68# Serialize entries to the compact v2 table wire format.
69def sst_serialize(+entries: List<&2, MemTable.Entry>, +level: Nat) -> String:
70  SstFile.serialize(entries, level)
71
72# Parse v2 table bytes with checksum verification; None when rejected.
73def sst_parse(+encoded: String) -> Maybe<&2, Sstable.Table>:
74  SstFile.parse(encoded)
75
76# Serialize a Manifest to its exact on-disk bytes.
77def mfst_serialize(mfst: Manifest.Manifest) -> String:
78  Manifest.serialize(mfst)
79
80# Parse Manifest bytes; None on corrupt input (fail-closed).
81def mfst_parse(+str: String) -> Maybe<&2, Manifest.Manifest>:
82  Manifest.parse(str)
83
84# --- Level 1: Sess session monad (primary API) ---
85type Sess<a, -A: Kind(a)> is Kind(a <&> &1):
86  Sess{run: Db.Db -> (Db.Db & A)}
87
88def Sess.pure(a, -A: Kind(a), val: A) -> Sess<a, A>:
89  Sess{db => (db, val)}
90
91def Sess.run_go(b, -B: Kind(b), sess: Sess<b, B>, st: Db.Db) -> Db.Db & B:
92  match sess:
93    case Sess{run}: run(st)
94
95def Sess.apply(a, -A: Kind(a), b, -B: Kind(b), pair: Db.Db & A, fun: A -> Sess<b, B>) -> Db.Db & B:
96  match pair:
97    case (st, x): Sess.run_go(b, B, fun(x), st)
98
99def Sess.bind(a, -A: Kind(a), -B: Kind(a), sess: Sess<a, A>, fun: A -> Sess<a, B>) -> Sess<a, B>:
100  Sess{st => Sess.apply(a, A, a, B, Sess.run_go(a, A, sess, st), fun)}
101
102# Open a database handle over a directory label (pure; no IO touched).
103def open(+dir: String) -> Db.Db:
104  Db.open_db(dir)
105
106# Reopen a directory replaying wal.log chunks into a fresh memtable.
107# Missing file/dir answers open_db; torn tail discarded; IO errors Fail.
108def reopen(+dir: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
109  Db.reopen(dir)
110
111# Pure put transition used by the session action.
112def put_go(+db: Db.Db, +key: String, +val: String) -> Db.Db:
113  match db:
114    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
115      Db.apply_done(Db.apply_and_rotate(Con{Wal.Put{key, val}, Nil{}}, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
116
117# Pure delete transition used by the session action.
118def del_go(+db: Db.Db, +key: String) -> Db.Db:
119  match db:
120    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
121      Db.apply_done(Db.apply_and_rotate(Con{Wal.Del{key}, Nil{}}, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
122
123# Pure batch transition used by the session action.
124def batch_go(+db: Db.Db, +muts: List<&2, Wal.Mut>) -> Db.Db:
125  match db:
126    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
127      Db.apply_done(Db.apply_and_rotate(muts, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
128
129# Pure point-read transition used by the session action (cache-backed).
130def sget_go(+db: Db.Db, +key: String) -> Db.Db & Maybe<&2, String>:
131  Db.db_get_cached(db, key)
132
133# Session step writing one key.
134def sput(+key: String, +val: String) -> Sess<&2, Unit>:
135  Sess{db => (put_go(db, key, val), Unit{})}
136
137# Session step deleting one key.
138def sdel(+key: String) -> Sess<&2, Unit>:
139  Sess{db => (del_go(db, key), Unit{})}
140
141# Session step applying a whole mutation batch at once.
142def sbatch(+muts: List<&2, Wal.Mut>) -> Sess<&2, Unit>:
143  Sess{db => (batch_go(db, muts), Unit{})}
144
145# Session step reading one key.
146def sget(+key: String) -> Sess<&2, Maybe<&2, String>>:
147  Sess{db => sget_go(db, key)}
148
149# Execute a whole session against a handle.
150def run_sess(a, -A: Kind(a), st: Db.Db, sess: Sess<a, A>) -> Db.Db & A:
151  Sess.run_go(a, A, sess, st)
152
153# Project the answer out of a runner pair.
154def value_of(a, -A: Kind(a), pair: Db.Db & A) -> A:
155  match pair:
156    case (db, x): x
157
158# Render the exact WAL bytes a VFS adapter frames with wal_frame plus fsync.
159def encode_batch(batch: Wal.Batch) -> String:
160  Wal.encode(batch)