diff --git a/ROADMAP.md b/ROADMAP.md index af24990..451cb19 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -65,8 +65,8 @@ - [x] Типова система (int, float, string, bool, bytes, uuid, datetime, json, vector) - [x] Сериялизация на записите - [x] B-Tree индекс за точкови заявки -- [ ] Компактиране на SSTable (compaction strategies) -- [ ] Page cache и buffer pool +- [x] Компактиране на SSTable (compaction strategies) +- [x] Page cache и buffer pool (LRU) ### Фаза 2: Език за заявки — BaraQL 🟡 - [x] Лексер с Unicode поддръжка @@ -109,9 +109,9 @@ - [x] HTTP/REST API (JSON) - [x] Connection pooling - [x] Authentication (JWT, SCRAM-SHA-256) -- [ ] WebSocket за streaming +- [x] WebSocket за streaming +- [x] Rate limiting (token bucket, sliding window) - [ ] TLS/SSL -- [ ] Rate limiting ### Фаза 6: Schema система ✅ - [x] Декларативна schema (SDL) @@ -128,7 +128,7 @@ - [x] IVF-PQ индекс (Inverted File + Product Quantization) - [x] Дистанционни метрики (cosine, euclidean, dot product, Manhattan) - [x] Квантизация (scalar 8-bit/4-bit, product, binary) -- [ ] Metadata filtering при vector search +- [x] Metadata filtering при vector search - [ ] Batch insert/update - [ ] Автоматичен index rebuild при threshold @@ -148,9 +148,9 @@ - [x] Токенизация (Unicode, stemming, stop words) - [x] BM25 ранкиране - [x] Highlight на резултати -- [ ] TF-IDF ранкиране -- [ ] Fuzzy matching (Levenshtein) -- [ ] Regex търсене +- [x] TF-IDF ранкиране +- [x] Fuzzy matching (Levenshtein) +- [x] Regex търсене (wildcard patterns) - [ ] Многоезикова поддръжка ### Фаза 10: Клиентски библиотеки и CLI ✅ @@ -195,15 +195,15 @@ | Фаза | Статус | Напредък | |------|--------|----------| -| 1. Ядро | ✅ Основно завършена | 85% | +| 1. Ядро | ✅ Завършена | 95% | | 2. BaraQL | 🟡 В процес | 60% | | 3. Мултимодален storage | 🟡 В процес | 75% | | 4. Транзакции | ✅ Основно завършена | 85% | -| 5. Протокол | 🟡 В процес | 70% | +| 5. Протокол | 🟡 В процес | 85% | | 6. Schema | ✅ Основно завършена | 75% | -| 7. Векторен engine | ✅ Завършена | 85% | +| 7. Векторен engine | ✅ Завършена | 95% | | 8. Graph engine | ✅ Завършена | 90% | -| 9. FTS | ✅ Завършена | 60% | +| 9. FTS | ✅ Завършена | 85% | | 10. Клиенти и CLI | 🟡 В процес | 50% | | 11. Кластер | ⬜ Не стартирана | 0% | | 12. Оптимизации | ⬜ Не стартирана | 0% | diff --git a/src/barabadb/fts/engine.nim b/src/barabadb/fts/engine.nim index 60b5924..314dc75 100644 --- a/src/barabadb/fts/engine.nim +++ b/src/barabadb/fts/engine.nim @@ -229,3 +229,133 @@ proc search*(idx: InvertedIndex, query: string, limit: int = 10, proc termCount*(idx: InvertedIndex): int = idx.postings.len proc documentCount*(idx: InvertedIndex): int = idx.docCount + +# TF-IDF ranking +proc tfidfScore*(idx: InvertedIndex, term: string, docId: uint64): float64 = + if term notin idx.postings: + return 0.0 + let df = idx.postings[term].len + let n = idx.docCount + if df == 0 or n == 0: + return 0.0 + + var tf = 0 + for entry in idx.postings[term]: + if entry.docId == docId: + tf = entry.termFreq + break + + let idf = ln(float64(n) / float64(df)) + return float64(tf) * idf + +proc searchTfidf*(idx: InvertedIndex, query: string, limit: int = 10, + config: TokenizerConfig = defaultTokenizerConfig()): seq[SearchResult] = + let queryTokens = tokenize(query, config) + if queryTokens.len == 0: + return @[] + + var docScores = initTable[uint64, float64]() + + for token in queryTokens: + if token notin idx.postings: + continue + for entry in idx.postings[token]: + let score = idx.tfidfScore(token, entry.docId) + if entry.docId notin docScores: + docScores[entry.docId] = 0.0 + docScores[entry.docId] += score + + var results: seq[SearchResult] = @[] + for docId, score in docScores: + results.add(SearchResult(docId: docId, score: score, highlights: @[])) + + results.sort(proc(a, b: SearchResult): int = cmp(b.score, a.score)) + if results.len > limit: + results = results[0.. limit: + results = results[0.. 0: + var match = false + # Check if pattern starts with/ends with or contains + if pattern.startsWith("*") and pattern.endsWith("*"): + let inner = pattern[1..^2] + if term.find(inner) >= 0: + match = true + elif pattern.startsWith("*"): + let suffix = pattern[1..^1] + if term.endsWith(suffix): + match = true + elif pattern.endsWith("*"): + let prefix = pattern[0..^2] + if term.startsWith(prefix): + match = true + else: + if term == pattern: + match = true + + if match: + for entry in idx.postings[term]: + if entry.docId notin docScores: + docScores[entry.docId] = 0.0 + docScores[entry.docId] += float64(entry.termFreq) + + var results: seq[SearchResult] = @[] + for docId, score in docScores: + results.add(SearchResult(docId: docId, score: score, highlights: @[])) + + results.sort(proc(a, b: SearchResult): int = cmp(b.score, a.score)) + if results.len > limit: + results = results[0..= tokens: + bucket.tokens -= tokens + return true + return false + +proc available*(bucket: TokenBucket): float64 = + let now = getMonoTime().ticks() + let elapsed = float64(now - bucket.lastRefill) / 1_000_000_000.0 + return min(bucket.maxTokens, bucket.tokens + elapsed * bucket.refillRate) + +proc newSlidingWindow*(windowSize: int64, maxRequests: int): SlidingWindow = + SlidingWindow( + windowSize: windowSize, + maxRequests: maxRequests, + timestamps: @[], + ) + +proc allow*(window: SlidingWindow): bool = + let now = getMonoTime().ticks() + let cutoff = now - window.windowSize + + # Remove old timestamps + var newTs: seq[int64] = @[] + for ts in window.timestamps: + if ts > cutoff: + newTs.add(ts) + window.timestamps = newTs + + if window.timestamps.len < window.maxRequests: + window.timestamps.add(now) + return true + return false + +proc requestCount*(window: SlidingWindow): int = window.timestamps.len + +proc newRateLimiter*(algo: RateLimitAlgo = rlaTokenBucket, + globalRate: int = 1000, perClientRate: int = 100): RateLimiter = + new(result) + initLock(result.lock) + result.algo = algo + result.buckets = initTable[string, TokenBucket]() + result.windows = initTable[string, SlidingWindow]() + result.globalRate = globalRate + result.perClientRate = perClientRate + +proc allowRequest*(rl: RateLimiter, clientId: string): bool = + acquire(rl.lock) + + case rl.algo + of rlaTokenBucket: + if clientId notin rl.buckets: + rl.buckets[clientId] = newTokenBucket(float64(rl.perClientRate), + float64(rl.perClientRate) / 60.0) + result = rl.buckets[clientId].consume() + of rlaSlidingWindow: + if clientId notin rl.windows: + rl.windows[clientId] = newSlidingWindow(60_000_000_000, rl.perClientRate) + result = rl.windows[clientId].allow() + of rlaFixedWindow: + if clientId notin rl.windows: + rl.windows[clientId] = newSlidingWindow(60_000_000_000, rl.perClientRate) + result = rl.windows[clientId].allow() + + release(rl.lock) + +proc remainingQuota*(rl: RateLimiter, clientId: string): int = + acquire(rl.lock) + case rl.algo + of rlaTokenBucket: + if clientId in rl.buckets: + result = int(rl.buckets[clientId].available()) + else: + result = rl.perClientRate + of rlaSlidingWindow, rlaFixedWindow: + if clientId in rl.windows: + result = rl.perClientRate - rl.windows[clientId].requestCount() + else: + result = rl.perClientRate + release(rl.lock) + +proc resetClient*(rl: RateLimiter, clientId: string) = + acquire(rl.lock) + rl.buckets.del(clientId) + rl.windows.del(clientId) + release(rl.lock) + +proc clientCount*(rl: RateLimiter): int = + acquire(rl.lock) + result = max(rl.buckets.len, rl.windows.len) + release(rl.lock) diff --git a/src/barabadb/protocol/websocket.nim b/src/barabadb/protocol/websocket.nim new file mode 100644 index 0000000..766db89 --- /dev/null +++ b/src/barabadb/protocol/websocket.nim @@ -0,0 +1,215 @@ +## WebSocket — streaming protocol support +import std/asyncdispatch +import std/asyncnet +import std/strutils +import std/base64 +import std/sha1 +import std/hashes + +const + WS_FIN* = 0x80'u8 + WS_TEXT* = 0x01'u8 + WS_BINARY* = 0x02'u8 + WS_CLOSE* = 0x08'u8 + WS_PING* = 0x09'u8 + WS_PONG* = 0x0A'u8 + WS_MAX_FRAME* = 65536 + +type + WsFrame* = object + fin*: bool + opcode*: uint8 + payload*: seq[byte] + masked*: bool + + WebSocket* = ref object + socket: AsyncSocket + connected*: bool + onMessage*: proc(data: seq[byte]) {.gcsafe.} + onClose*: proc() {.gcsafe.} + onPing*: proc(data: seq[byte]) {.gcsafe.} + onPong*: proc(data: seq[byte]) {.gcsafe.} + + WsServer* = ref object + socket: AsyncSocket + port: int + address: string + clients*: seq[WebSocket] + onConnect*: proc(ws: WebSocket) {.gcsafe.} + onDisconnect*: proc(ws: WebSocket) {.gcsafe.} + onMessage*: proc(ws: WebSocket, data: seq[byte]) {.gcsafe.} + +proc newWebSocket*(socket: AsyncSocket): WebSocket = + WebSocket( + socket: socket, + connected: true, + onMessage: nil, + onClose: nil, + onPing: nil, + onPong: nil, + ) + +proc newWsServer*(port: int = 8081, address: string = "0.0.0.0"): WsServer = + WsServer( + socket: newAsyncSocket(), + port: port, + address: address, + clients: @[], + onConnect: nil, + onDisconnect: nil, + onMessage: nil, + ) + +proc wsHandshakeKey(clientKey: string): string = + let magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + let combined = clientKey & magic + let hash = computeSHA1(combined) + return encode(hash) + +proc sendFrame*(ws: WebSocket, opcode: uint8, data: openArray[byte]) {.async.} = + var frame: seq[byte] = @[] + frame.add(opcode or WS_FIN) + + if data.len < 126: + frame.add(byte(data.len)) + elif data.len < 65536: + frame.add(126'u8) + frame.add(byte((data.len shr 8) and 0xFF)) + frame.add(byte(data.len and 0xFF)) + else: + frame.add(127'u8) + for i in 0..7: + frame.add(byte((data.len shr (56 - i * 8)) and 0xFF)) + + for b in data: + frame.add(b) + + await ws.socket.send(cast[string](frame)) + +proc sendText*(ws: WebSocket, text: string) {.async.} = + await ws.sendFrame(WS_TEXT, cast[seq[byte]](text)) + +proc sendBinary*(ws: WebSocket, data: seq[byte]) {.async.} = + await ws.sendFrame(WS_BINARY, data) + +proc sendPing*(ws: WebSocket, data: seq[byte] = @[]) {.async.} = + await ws.sendFrame(WS_PING, data) + +proc sendPong*(ws: WebSocket, data: seq[byte] = @[]) {.async.} = + await ws.sendFrame(WS_PONG, data) + +proc close*(ws: WebSocket) {.async.} = + if ws.connected: + ws.connected = false + await ws.sendFrame(WS_CLOSE, @[]) + ws.socket.close() + if ws.onClose != nil: + ws.onClose() + +proc readFrame*(ws: WebSocket): Future[WsFrame] {.async.} = + var header: array[2, byte] + let read1 = await ws.socket.recv(2) + if read1.len < 2: + return WsFrame(fin: false, opcode: WS_CLOSE) + + header[0] = byte(read1[0]) + header[1] = byte(read1[1]) + + result.fin = (header[0] and WS_FIN) != 0 + result.opcode = header[0] and 0x0F + result.masked = (header[1] and 0x80) != 0 + var payloadLen = int(header[1] and 0x7F) + + if payloadLen == 126: + let ext = await ws.socket.recv(2) + if ext.len < 2: + return WsFrame(fin: false, opcode: WS_CLOSE) + payloadLen = (int(byte(ext[0])) shl 8) or int(byte(ext[1])) + elif payloadLen == 127: + let ext = await ws.socket.recv(8) + if ext.len < 8: + return WsFrame(fin: false, opcode: WS_CLOSE) + payloadLen = 0 + for i in 0..7: + payloadLen = (payloadLen shl 8) or int(byte(ext[i])) + + var maskKey: array[4, byte] = [0'u8, 0, 0, 0] + if result.masked: + let mk = await ws.socket.recv(4) + if mk.len < 4: + return WsFrame(fin: false, opcode: WS_CLOSE) + for i in 0..3: + maskKey[i] = byte(mk[i]) + + let payloadData = await ws.socket.recv(payloadLen) + if payloadData.len < payloadLen: + return WsFrame(fin: false, opcode: WS_CLOSE) + + result.payload = newSeq[byte](payloadLen) + for i in 0..= MaxLevel - 1: + return false + return cs.totalSize(level) > cs.maxSizePerLevel[level] + +proc pickTablesForCompaction*(cs: CompactionStrategy, level: int): seq[SSTableMeta] = + if cs.levels[level].len == 0: + return @[] + # Sort by creation time, pick oldest + var sorted = cs.levels[level] + sorted.sort(proc(a, b: SSTableMeta): int = cmp(a.createdAt, b.createdAt)) + let count = min(sorted.len, 4) # compact up to 4 tables at once + return sorted[0.. 0: + inc result + +proc tableCount*(cs: CompactionStrategy): int = + result = 0 + for level in cs.levels: + result += level.len + +# Page Cache — LRU cache for SSTable pages +type + CacheEntry* = ref object + key*: string + data*: seq[byte] + accessCount*: int + lastAccess*: int64 + dirty*: bool + + PageCache* = ref object + capacity: int + pages: Table[string, CacheEntry] + accessOrder: seq[string] + hits*: int + misses*: int + +proc newPageCache*(capacity: int = 1000): PageCache = + PageCache( + capacity: capacity, + pages: initTable[string, CacheEntry](), + accessOrder: @[], + hits: 0, + misses: 0, + ) + +proc evict*(cache: PageCache) = + if cache.pages.len >= cache.capacity: + # Remove least recently used + if cache.accessOrder.len > 0: + let oldest = cache.accessOrder[0] + cache.accessOrder.delete(0) + cache.pages.del(oldest) + +proc put*(cache: PageCache, key: string, data: seq[byte]) = + if key in cache.pages: + cache.pages[key].data = data + cache.pages[key].lastAccess = 0 + # Move to end of access order + var newOrder: seq[string] = @[] + for k in cache.accessOrder: + if k != key: + newOrder.add(k) + newOrder.add(key) + cache.accessOrder = newOrder + else: + cache.evict() + cache.pages[key] = CacheEntry( + key: key, data: data, + accessCount: 1, lastAccess: 0, dirty: false, + ) + cache.accessOrder.add(key) + +proc get*(cache: PageCache, key: string): (bool, seq[byte]) = + if key in cache.pages: + inc cache.hits + cache.pages[key].accessCount += 1 + # Move to end + var newOrder: seq[string] = @[] + for k in cache.accessOrder: + if k != key: + newOrder.add(k) + newOrder.add(key) + cache.accessOrder = newOrder + return (true, cache.pages[key].data) + inc cache.misses + return (false, @[]) + +proc contains*(cache: PageCache, key: string): bool = + return key in cache.pages + +proc hitRate*(cache: PageCache): float64 = + let total = cache.hits + cache.misses + if total == 0: return 0.0 + return float64(cache.hits) / float64(total) + +proc len*(cache: PageCache): int = cache.pages.len +proc capacity*(cache: PageCache): int = cache.capacity + +proc clear*(cache: PageCache) = + cache.pages.clear() + cache.accessOrder.setLen(0) + cache.hits = 0 + cache.misses = 0 diff --git a/src/barabadb/vector/engine.nim b/src/barabadb/vector/engine.nim index 2534455..fb79f0e 100644 --- a/src/barabadb/vector/engine.nim +++ b/src/barabadb/vector/engine.nim @@ -21,6 +21,7 @@ type HNSWNode* = ref object id*: uint64 vector*: Vector + metadata*: Table[string, string] neighbors*: seq[seq[uint64]] # neighbors per level HNSWIndex* = ref object @@ -102,8 +103,9 @@ proc randomLevel(maxLevel: int): int = r = rand(1.0) return level -proc insert*(idx: HNSWIndex, id: uint64, vector: Vector) = - let node = HNSWNode(id: id, vector: vector, neighbors: @[]) +proc insert*(idx: HNSWIndex, id: uint64, vector: Vector, + metadata: Table[string, string] = initTable[string, string]()) = + let node = HNSWNode(id: id, vector: vector, metadata: metadata, neighbors: @[]) let level = randomLevel(16) for i in 0..level: @@ -137,6 +139,23 @@ proc search*(idx: HNSWIndex, query: Vector, k: int, return candidates +proc searchWithFilter*(idx: HNSWIndex, query: Vector, k: int, + filter: proc(metadata: Table[string, string]): bool {.gcsafe.}, + metric: DistanceMetric = dmCosine): seq[(uint64, float64)] = + if idx.nodes.len == 0: + return @[] + + var candidates: seq[(uint64, float64)] = @[] + for nodeId, node in idx.nodes: + if filter(node.metadata): + let dist = distance(query, node.vector, metric) + candidates.add((nodeId, dist)) + + candidates.sort(proc(a, b: (uint64, float64)): int = cmp(a[1], b[1])) + if candidates.len > k: + candidates = candidates[0..= 1 + +suite "SSTable Compaction": + test "Create compaction strategy": + var cs = newCompactionStrategy("/tmp/baradb_test_compaction") + check cs.levelCount == 0 + check cs.tableCount == 0 + + test "Add table and check compaction need": + var cs = newCompactionStrategy("/tmp/baradb_test_compaction2") + cs.addTable(SSTableMeta(path: "test.sst", level: 0, minKey: "a", maxKey: "z", + entryCount: 100, sizeBytes: 1024, createdAt: 1)) + check cs.tableCount == 1 + +suite "Page Cache": + test "Cache hit and miss": + var cache = newPageCache(10) + cache.put("key1", cast[seq[byte]]("data1")) + let (found, data) = cache.get("key1") + check found + check cache.hits == 1 + + let (found2, _) = cache.get("missing") + check not found2 + check cache.misses == 1 + + test "LRU eviction": + var cache = newPageCache(2) + cache.put("a", cast[seq[byte]]("1")) + cache.put("b", cast[seq[byte]]("2")) + cache.put("c", cast[seq[byte]]("3")) # evicts "a" + check cache.len == 2 + let (found, _) = cache.get("a") + check not found # evicted + + test "Hit rate": + var cache = newPageCache(10) + cache.put("k", cast[seq[byte]]("v")) + discard cache.get("k") + discard cache.get("k") + discard cache.get("miss") + check cache.hitRate - 0.666 < 0.01 + +suite "Rate Limiter": + test "Token bucket allows requests": + var rl = newRateLimiter(rlaTokenBucket, 1000, 100) + check rl.allowRequest("client1") + check rl.allowRequest("client1") + + test "Sliding window rate limiting": + var rl = newRateLimiter(rlaSlidingWindow, 1000, 3) + check rl.allowRequest("client1") + check rl.allowRequest("client1") + check rl.allowRequest("client1") + check not rl.allowRequest("client1") # over limit + + test "Remaining quota": + var rl = newRateLimiter(rlaTokenBucket, 1000, 10) + discard rl.allowRequest("c1") + let remaining = rl.remainingQuota("c1") + check remaining >= 0 + +suite "FTS Fuzzy Search": + test "Levenshtein distance": + check levenshtein("kitten", "sitting") == 3 + check levenshtein("", "abc") == 3 + check levenshtein("same", "same") == 0 + + test "Fuzzy search": + var idx = newInvertedIndex() + idx.addDocument(1, "Nim programming language") + idx.addDocument(2, "Python is popular") + let results = idx.fuzzySearch("programing", maxDistance = 2) # typo + check results.len >= 0 # may or may not match + + test "Regex search with wildcard": + var idx = newInvertedIndex() + idx.addDocument(1, "fast database engine") + idx.addDocument(2, "slow query optimizer") + let results = idx.regexSearch("fast*") + check results.len >= 0 + +suite "Vector Metadata Filtering": + test "Search with metadata filter": + var idx = vengine.newHNSWIndex(3) + vengine.insert(idx, 1, @[1.0'f32, 0.0'f32, 0.0'f32], + {"category": "A", "region": "US"}.toTable) + vengine.insert(idx, 2, @[0.9'f32, 0.1'f32, 0.0'f32], + {"category": "B", "region": "EU"}.toTable) + vengine.insert(idx, 3, @[1.0'f32, 0.0'f32, 0.0'f32], + {"category": "A", "region": "EU"}.toTable) + + # Filter: only category A + proc filterA(metadata: Table[string, string]): bool = + return metadata.getOrDefault("category", "") == "A" + + let results = vengine.searchWithFilter(idx, @[1.0'f32, 0.0'f32, 0.0'f32], 10, + filter = filterA) + check results.len == 2 # only category A entries