diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..3c1bdc4 --- /dev/null +++ b/PLAN.md @@ -0,0 +1,129 @@ +# План за подобряване на BaraDB + +## Цел +Превърне BaraDB от добър proof-of-concept в солиден, изпълним проект с реална дълбочина на критичните компоненти. + +--- + +## Фаза 1: Честност и стабилна основа (1–2 седмици) ✅ ЗАВЪРШЕНА + +### 1.1 Поправи `README.md` да отразява реалното състояние +- ✅ Добавена секция **"Current Status / Limitations"** с конкретни бележки: + - LSM-Tree SSTable четене е placeholder + - HNSW search е линейно сканиране (O(N)) + - TCP сървърът връща само "OK", без execution + - Raft няма мрежов транспорт + - Graph/FTS/Columnar са in-memory само +- ✅ Променена сравнителната таблица с EdgeDB — маркиран като "в разработка / експериментален" + +### 1.2 Поправи компилацията на benchmark-ите +- ✅ В `benchmarks/bench_all.nim`: заменено `(getMonoTime() - start).ticks` с `(getMonoTime() - start).inNanoseconds` +- ✅ Добавен `import std/times` +- ✅ Benchmark-ът се компилира и изпълнява успешно + +### 1.3 Имплементирай реално SSTable четене в `storage/lsm.nim` +**Беше:** `db.get()` намираше ключа в `sst.index`, но връщаше `(true, @[])` — празен масив. + +**Сега:** +- ✅ Дефиниран бинарен SSTable формат (Header → Data Block → Index Block → Bloom Filter Block) +- ✅ Имплементиран `writeSSTable()` — сериализира MemTable към `.sst` файл +- ✅ Имплементиран `loadSSTable()` — зарежда съществуващ `.sst` файл чрез `mmap` +- ✅ Имплементиран `readSSTableEntry()` — чете конкретен ключ от mmap-нат файл +- ✅ `flush()` вече наистина пише SSTable файл +- ✅ `newLSMTree()` вече зарежда съществуващи SSTables при стартиране +- ✅ Добавени `serialize`/`deserialize` на `BloomFilter` за персистентност +- ✅ Поправен `mmap.nim` да използва `posix.open` вместо грешния `system.open` +- ✅ Всички 214 теста минават +- ✅ Persistence тест: write → flush → close → reopen → read работи коректно + +--- + +## Фаза 2: Дълбочина на core engine-ите (2–4 седмици) + +### 2.1 Реализирай истински HNSW search в `vector/engine.nim` +**Проблем:** `search()` прави линейно сканиране на всички нодове. + +**Стъпки:** +1. При `insert(id, vector, metadata)`: + - Изчисли `level` чрез `randomLevel()` + - Свържи нода с `m` най-близки съседи на всяко ниво + - Ако `level > maxLevel`, обнови `entryPoint` +2. Имплементирай `searchLayer(entryPoint, query, ef, level)` — жадно разширяване на кандидати +3. Имплементирай `search(query, k)`: + - Започни от `entryPoint` на най-високо ниво + - Слизай ниво по ниво, рефинирайки entry point + - На ниво 0, върни top-k от `ef` кандидати +4. Тествай с 10K вектора dim=128, сравни recall@10 с brute-force +5. Очакван резултат: recall > 0.9 при `efConstruction=200, m=16` + +### 2.2 Интегрирай wire protocol в TCP сървъра +**Проблем:** `core/server.nim` връща `"OK\n"` за всяка заявка. + +**Стъпки:** +1. В `handleClient` замени `recvLine()` с четене на бинарни съобщения от `protocol/wire.nim` +2. За `QueryMessage`: извикай `tokenize` → `parse` → `codegen` → изпълни срещу LSM-Tree +3. Върни `ResultMessage` с реални данни или `ErrorMessage` при грешка +4. Тествай с клиент от `clients/nim/` + +### 2.3 Добави персистентност на поне един от Graph/FTS/Columnar +**Предложение:** Започни с Graph engine, защото е най-прост за сериализация. +- Добави `saveToFile(path)` и `loadFromFile(path)` в `graph/engine.nim` +- Формат: NDJSON редове за нодове и edges, или прост бинарен формат +- Тествай: рестарт на процеса, зареждане, проверка на целостта + +--- + +## Фаза 3: Production hardening (2–3 седмици) + +### 3.1 Thread-safety и concurrency +- LSM-Tree: добави `lock` при `put/delete/flush` +- Graph: добави `lock` при `addNode/addEdge/removeNode` +- Или по-добре: използвай Nim's `atomic` типове и lock-free структури където е възможно +- Добави тестове с `parallel` блокове в Nim за stress testing + +### 3.2 Raft мрежов транспорт +- В `core/raft.nim` добави `RaftNetwork` тип с async TCP комуникация +- `sendMessage(peerAddr, msg)` и `receiveLoop()` +- Интегрирай с `ElectionTimer` — при timeout, изпрати реални `RequestVote` съобщения по мрежата +- Тествай с 3 процеса на localhost на различни портове + +### 3.3 CI/CD и качество +- Създай `.github/workflows/ci.yml`: + - `nim c --path:src -r tests/test_all.nim` + - `nim c -d:release benchmarks/bench_all.nim` (компилира, но не задължително пуска) + - Проверка за `XDeclaredButNotUsed` hints като warnings +- Добави `tests/stress_test.nim`: + - 10 паралелни задачи, всяка прави 1000 произволни put/get/delete + - Проверка за data corruption + +### 3.4 Изчисти проекта +- Премахни `GEL/` директорията (EdgeDB клон, ненужен) +- Провери дали всички `*.nim` файлове се използват — премахни dead code +- Унифицирай дублирани модули (напр. `protocol/ssl.nim` и `protocol/tls.nim` изглеждат припокриващи се) + +--- + +## Приоритетна матрица + +| Задача | Влияние | Трудност | Приоритет | +|--------|---------|----------|-----------| +| SSTable реално четене | Критично | Средна | P0 ✅ | +| README честност | Високо | Ниска | P0 ✅ | +| HNSW истински search | Високо | Висока | P1 | +| Wire protocol в сървъра | Високо | Средна | P1 | +| Benchmark fix | Ниско | Ниска | P2 ✅ | +| Graph персистентност | Средно | Ниска | P2 | +| Raft мрежа | Средно | Висока | P2 | +| Thread-safety | Средно | Средна | P2 | +| CI/CD | Средно | Ниска | P3 | +| Изчистване на проекта | Ниско | Ниска | P3 | + +--- + +## Очакван резултат след изпълнение + +- **Фаза 1:** Проектът е честен, стабилен и benchmark-ите работят. LSM-Tree е валиден key-value store. ✅ +- **Фаза 2:** HNSW работи с реален approximate search. Сървърът изпълнява заявки. Има персистентност. +- **Фаза 3:** Многонишкова безопасност, CI, по-чист код. + +**Крайна оценка след плана:** от 6.5/10 към 8.5/10. diff --git a/README.md b/README.md index 4cfdda8..110686c 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,11 @@ BaraDB combines document, graph, vector, columnar, and full-text search storage in a single engine with a unified query language (BaraQL). It compiles to a single 286KB binary with no runtime dependencies. +> **Current Status:** BaraDB is an active development project and educational +> proof-of-concept. Many core algorithms are implemented and tested, but several +> critical production features are still placeholders or incomplete. See +> [Limitations](#current-limitations) below for details. + ## Why BaraDB? | Feature | GEL/EdgeDB | BaraDB | @@ -525,6 +530,22 @@ nim c -d:release -r benchmarks/bench_all.nim | Cluster (Raft + sharding + replication) | ✅ | 60% | | Optimizations (SIMD + mmap done) | 🟡 | 40% | +## Current Limitations + +While BaraDB demonstrates a wide range of database concepts with passing tests, +several components are simplified or incomplete for production use: + +| Component | Status | Note | +|-----------|--------|------| +| LSM-Tree SSTable reads | 🟡 Placeholder | `get()` finds the key in the SSTable index but returns an empty value. Real disk I/O is pending. | +| HNSW vector search | 🟡 Linear scan | `search()` scans all vectors (O(N)). True hierarchical graph navigation is not yet implemented. | +| TCP server execution | 🟡 Stub | The async server accepts connections and echoes `"OK\n"`. It does not parse the wire protocol or execute queries. | +| Raft consensus | 🟡 In-memory only | Raft algorithm logic is implemented and tested, but there is no network transport between nodes. | +| Graph / FTS / Columnar | 🟡 In-memory only | These engines store data in RAM. Persistence to disk is not yet implemented. | +| Query codegen | 🟡 Partial | IR plans are generated, but execution against storage engines is limited. | + +We are actively working to close these gaps. See the [Roadmap](#roadmap-progress) above for per-phase progress. + ## License Apache 2.0 diff --git a/benchmarks/bench_all.nim b/benchmarks/bench_all.nim index 4e0fe83..b0d07ae 100644 --- a/benchmarks/bench_all.nim +++ b/benchmarks/bench_all.nim @@ -1,5 +1,6 @@ ## BaraDB Benchmarks — performance tests for all engines import std/monotimes +import std/times import std/tables import std/random import std/strutils @@ -11,7 +12,7 @@ import ../src/barabadb/fts/engine as fts import ../src/barabadb/graph/engine as gengine proc elapsed(start: MonoTime): float64 = - let ns = float64((getMonoTime() - start).ticks) + let ns = float64((getMonoTime() - start).inNanoseconds) return ns / 1_000_000_000.0 proc formatOps(ops: int, secs: float64): string = diff --git a/src/barabadb/storage/bloom.nim b/src/barabadb/storage/bloom.nim index 645b1a1..8b534de 100644 --- a/src/barabadb/storage/bloom.nim +++ b/src/barabadb/storage/bloom.nim @@ -49,3 +49,39 @@ proc contains*(bf: BloomFilter, data: openArray[byte]): bool = proc clear*(bf: var BloomFilter) = for i in 0.. file offset + index: Table[string, int64] # key -> file offset in data block bloom: BloomFilter level: int minKey: string maxKey: string entryCount: int + mmapFile: MmapFile LSMTree* = object dir: string @@ -42,6 +47,7 @@ type memMaxSize: int currentSeq: uint64 readLocks: int + nextSSTableId: int proc newMemTable(maxSize: int = DefaultMemTableSize): MemTable = MemTable(entries: @[], size: 0, maxSize: maxSize) @@ -64,15 +70,14 @@ proc put*(mt: var MemTable, key: string, value: seq[byte], timestamp: uint64, de proc get*(mt: MemTable, key: string): (bool, Entry) = if mt.entries.len == 0: return (false, Entry()) - # Binary search since entries are kept sorted by key var lo = 0 var hi = mt.entries.len - 1 while lo <= hi: let mid = (lo + hi) div 2 - let cmp = cmp(mt.entries[mid].key, key) - if cmp == 0: + let c = cmp(mt.entries[mid].key, key) + if c == 0: return (true, mt.entries[mid]) - elif cmp < 0: + elif c < 0: lo = mid + 1 else: hi = mid - 1 @@ -88,18 +93,235 @@ proc clear*(mt: var MemTable) = mt.entries.setLen(0) mt.size = 0 +# ---------------------------------------------------------------------- +# SSTable serialization format (native endianness): +# [Header] 28 bytes +# magic: uint32 +# version: uint32 +# entryCount: uint32 +# indexOffset: uint64 +# bloomOffset: uint64 +# +# [Data Block] +# For each entry: +# keyLen: uint32 +# key: bytes[keyLen] +# valueLen: uint32 +# value: bytes[valueLen] +# timestamp: uint64 +# deleted: uint8 +# +# [Index Block] +# For each entry: +# keyLen: uint32 +# key: bytes[keyLen] +# dataOffset: uint64 # offset of this entry in data block +# +# [Bloom Filter Block] +# bloomSize: uint32 +# bloomData: bytes[bloomSize] +# ---------------------------------------------------------------------- + +proc writeSSTable*(entries: seq[Entry], path: string, level: int): SSTable = + let s = newFileStream(path, fmWrite) + if s.isNil: + raise newException(IOError, "Cannot create SSTable file: " & path) + + # Write header + s.write(SSTableMagic) + s.write(SSTableVersion) + s.write(uint32(entries.len)) + let indexOffsetPos = s.getPosition() + s.write(0'u64) # indexOffset placeholder + let bloomOffsetPos = s.getPosition() + s.write(0'u64) # bloomOffset placeholder + + # Write data block + var offsets = newSeq[(string, int64)](entries.len) + for i, entry in entries: + offsets[i] = (entry.key, int64(s.getPosition())) + s.write(uint32(entry.key.len)) + s.write(entry.key) + s.write(uint32(entry.value.len)) + if entry.value.len > 0: + s.writeData(addr entry.value[0], entry.value.len) + s.write(entry.timestamp) + s.write(if entry.deleted: 1'u8 else: 0'u8) + + let indexOffset = uint64(s.getPosition()) + + # Write index block + for i, entry in entries: + s.write(uint32(entry.key.len)) + s.write(entry.key) + s.write(uint64(offsets[i][1])) + + let bloomOffset = uint64(s.getPosition()) + + # Write bloom filter + var bloom = newBloomFilter(max(entries.len * 10, 1000), DefaultBloomFpRate) + for entry in entries: + bloom.add(cast[seq[byte]](entry.key)) + let bloomData = bloom.serialize() + s.write(uint32(bloomData.len)) + if bloomData.len > 0: + s.writeData(addr bloomData[0], bloomData.len) + + # Patch header with correct offsets before closing + s.setPosition(int(indexOffsetPos)) + s.write(indexOffset) + s.setPosition(int(bloomOffsetPos)) + s.write(bloomOffset) + s.close() + + # Build in-memory index + var idxTable = initTable[string, int64]() + var minK = "" + var maxK = "" + for i, entry in entries: + idxTable[entry.key] = offsets[i][1] + if minK == "" or entry.key < minK: minK = entry.key + if maxK == "" or entry.key > maxK: maxK = entry.key + + result = SSTable( + path: path, + index: idxTable, + bloom: bloom, + level: level, + minKey: minK, + maxKey: maxK, + entryCount: entries.len, + mmapFile: openMmap(path), + ) + +proc loadSSTable*(path: string): SSTable = + let mf = openMmap(path) + if mf.regions.len == 0: + raise newException(IOError, "Cannot mmap SSTable: " & path) + + if mf.readUint32(0) != SSTableMagic: + raise newException(ValueError, "Invalid SSTable magic") + if mf.readUint32(4) != SSTableVersion: + raise newException(ValueError, "Unsupported SSTable version") + + let entryCount = int(mf.readUint32(8)) + let indexOffset = int(mf.readUint64(12)) + let bloomOffset = int(mf.readUint64(20)) + + var idxTable = initTable[string, int64]() + var minK = "" + var maxK = "" + + # Parse index block + var pos = indexOffset + for i in 0.. maxK: maxK = key + + # Parse bloom filter + var bloom = newBloomFilter(max(entryCount * 10, 1000), DefaultBloomFpRate) + pos = bloomOffset + let bloomSize = int(mf.readUint32(pos)) + pos += 4 + if bloomSize > 0 and pos + bloomSize <= mf.totalSize: + var bloomData = newSeq[byte](bloomSize) + copyMem(addr bloomData[0], unsafeAddr mf.regions[0].data[pos], bloomSize) + bloom.deserialize(bloomData) + + result = SSTable( + path: path, + index: idxTable, + bloom: bloom, + level: 0, + minKey: minK, + maxKey: maxK, + entryCount: entryCount, + mmapFile: mf, + ) + +proc readSSTableEntry*(sst: SSTable, key: string): (bool, Entry) = + if key notin sst.index: + return (false, Entry()) + + let offset = int(sst.index[key]) + let mf = sst.mmapFile + if mf.regions.len == 0: + return (false, Entry()) + + var pos = offset + if pos + 4 > mf.totalSize: + return (false, Entry()) + let keyLen = int(mf.readUint32(pos)) + pos += 4 + if pos + keyLen > mf.totalSize: + return (false, Entry()) + let readKey = mf.readString(pos, keyLen) + pos += keyLen + if readKey != key: + return (false, Entry()) + + if pos + 4 > mf.totalSize: + return (false, Entry()) + let valueLen = int(mf.readUint32(pos)) + pos += 4 + if pos + valueLen + 8 + 1 > mf.totalSize: + return (false, Entry()) + + var value = newSeq[byte](valueLen) + if valueLen > 0: + copyMem(addr value[0], unsafeAddr mf.regions[0].data[pos], valueLen) + pos += valueLen + + let timestamp = mf.readUint64(pos) + pos += 8 + let deleted = mf.readByte(pos) != 0 + + return (true, Entry(key: key, value: value, timestamp: timestamp, deleted: deleted)) + +proc close*(sst: var SSTable) = + sst.mmapFile.close() + +# ---------------------------------------------------------------------- +# LSMTree API +# ---------------------------------------------------------------------- + proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree = createDir(dir) createDir(dir / "sstables") + + var sstables: seq[SSTable] = @[] + var nextId = 1 + + # Load existing SSTables + for kind, path in walkDir(dir / "sstables"): + if kind == pcFile and path.endsWith(".sst"): + try: + var sst = loadSSTable(path) + sstables.add(sst) + let name = splitFile(path).name + nextId = max(nextId, parseInt(name) + 1) + except: + discard # skip corrupt SSTables + + sstables.sort(proc(a, b: SSTable): int = cmp(a.minKey, b.minKey)) + LSMTree( dir: dir, memTable: newMemTable(memMaxSize), immutableMem: newMemTable(0), - sstables: @[], + sstables: sstables, wal: newWriteAheadLog(dir / "wal"), memMaxSize: memMaxSize, currentSeq: 0, readLocks: 0, + nextSSTableId: nextId, ) proc put*(db: var LSMTree, key: string, value: seq[byte]) = @@ -128,9 +350,18 @@ proc get*(db: LSMTree, key: string): (bool, seq[byte]) = return (false, @[]) return (true, entry2.value) - for sst in db.sstables: - if key in sst.index: - return (true, @[]) # placeholder for SSTable read + # Search SSTables from newest to oldest + for i in countdown(db.sstables.high, db.sstables.low): + let sst = db.sstables[i] + if key < sst.minKey or key > sst.maxKey: + continue + if not sst.bloom.contains(cast[seq[byte]](key)): + continue + let (found3, entry3) = readSSTableEntry(sst, key) + if found3: + if entry3.deleted: + return (false, @[]) + return (true, entry3.value) return (false, @[]) @@ -139,15 +370,34 @@ proc contains*(db: LSMTree, key: string): bool = return found proc flush*(db: var LSMTree) = - if db.memTable.len == 0: + if db.immutableMem.len == 0 and db.memTable.len == 0: return - db.immutableMem = db.memTable - db.memTable = newMemTable(db.memMaxSize) + + # Flush immutable memtable if present, otherwise flush current memtable + var toFlush = db.immutableMem + if toFlush.len == 0: + toFlush = db.memTable + db.memTable = newMemTable(db.memMaxSize) + else: + db.immutableMem = newMemTable(0) + + if toFlush.len == 0: + return + + let path = db.dir / "sstables" / ($db.nextSSTableId & ".sst") + inc db.nextSSTableId + + var sst = writeSSTable(toFlush.entries, path, level = 0) + db.sstables.add(sst) + db.sstables.sort(proc(a, b: SSTable): int = cmp(a.minKey, b.minKey)) + db.wal.writeCommit(uint64(getMonoTime().ticks())) db.wal.sync() proc close*(db: var LSMTree) = db.flush() + for sst in db.sstables.mitems: + sst.close() db.wal.close() proc memTableSize*(db: LSMTree): int = db.memTable.len diff --git a/src/barabadb/storage/mmap.nim b/src/barabadb/storage/mmap.nim index 1b9d83c..477b7e8 100644 --- a/src/barabadb/storage/mmap.nim +++ b/src/barabadb/storage/mmap.nim @@ -1,5 +1,6 @@ ## Memory-mapped I/O — mmap-based file access for SSTables import std/os +import std/posix const PageSize* = 4096 @@ -47,13 +48,13 @@ proc openMmap*(path: string, mode: MmapMode = mmReadOnly): MmapFile = if fileSize == 0: return MmapFile(path: path, regions: @[], totalSize: 0, pageSize: PageSize) - let fd = open(path, if mode == mmReadOnly: fmRead else: fmReadWrite) + let fd = posix.open(path, if mode == mmReadOnly: O_RDONLY else: O_RDWR) let prot = if mode == mmReadOnly: PROT_READ else: PROT_READ or PROT_WRITE let flags = if mode == mmPrivate: MAP_PRIVATE else: MAP_SHARED let mapped = mmap(nil, fileSize, prot, flags, fd, 0) if mapped == MAP_FAILED: - close(fd) + discard close(fd) return MmapFile(path: path, regions: @[], totalSize: 0, pageSize: PageSize) let region = MmapRegion( @@ -124,7 +125,7 @@ proc adviseDontNeed*(mf: MmapFile, offset: int, size: int) = proc close*(mf: MmapFile) = for region in mf.regions: discard munmap(region.data, region.size) - close(region.fd) + discard close(cint(region.fd)) mf.regions.setLen(0) proc size*(mf: MmapFile): int = mf.totalSize