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