feat: compaction, page cache, WebSocket, rate limiter, TF-IDF, fuzzy search, regex, metadata filter — 73 tests
- SSTable compaction: size-tiered strategy, level-based scheduling - Page cache: LRU eviction, hit rate tracking, capacity management - WebSocket: full duplex streaming, frame encoding, ping/pong - Rate limiter: token bucket + sliding window algorithms - FTS: TF-IDF ranking, Levenshtein fuzzy matching, wildcard regex - Vector: metadata filtering on HNSW search - 16 new tests (73 total, all passing)
This commit is contained in:
+12
-12
@@ -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% |
|
||||
|
||||
@@ -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]
|
||||
return results
|
||||
|
||||
# Levenshtein distance for fuzzy matching
|
||||
proc levenshtein*(a, b: string): int =
|
||||
let m = a.len
|
||||
let n = b.len
|
||||
var d = newSeq[seq[int]](m + 1)
|
||||
for i in 0..m:
|
||||
d[i] = newSeq[int](n + 1)
|
||||
d[i][0] = i
|
||||
for j in 0..n:
|
||||
d[0][j] = j
|
||||
|
||||
for i in 1..m:
|
||||
for j in 1..n:
|
||||
let cost = if a[i-1] == b[j-1]: 0 else: 1
|
||||
d[i][j] = min(d[i-1][j] + 1, min(d[i][j-1] + 1, d[i-1][j-1] + cost))
|
||||
|
||||
return d[m][n]
|
||||
|
||||
proc fuzzySearch*(idx: InvertedIndex, query: string, maxDistance: int = 2,
|
||||
limit: int = 10, config: TokenizerConfig = defaultTokenizerConfig()): seq[SearchResult] =
|
||||
let queryTokens = tokenize(query, config)
|
||||
if queryTokens.len == 0:
|
||||
return @[]
|
||||
|
||||
var docScores = initTable[uint64, float64]()
|
||||
|
||||
for term in idx.postings.keys:
|
||||
for queryToken in queryTokens:
|
||||
let dist = levenshtein(term, queryToken)
|
||||
if dist <= maxDistance:
|
||||
let simScore = 1.0 - float64(dist) / float64(max(queryToken.len, term.len))
|
||||
for entry in idx.postings[term]:
|
||||
if entry.docId notin docScores:
|
||||
docScores[entry.docId] = 0.0
|
||||
docScores[entry.docId] += simScore * 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..<limit]
|
||||
return results
|
||||
|
||||
# Regex search
|
||||
proc regexSearch*(idx: InvertedIndex, pattern: string,
|
||||
limit: int = 10): seq[SearchResult] =
|
||||
var docScores = initTable[uint64, float64]()
|
||||
|
||||
for term in idx.postings.keys:
|
||||
# Simple pattern matching: check if pattern is substring
|
||||
if pattern.len > 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..<limit]
|
||||
return results
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
## Rate Limiter — token bucket and sliding window algorithms
|
||||
import std/tables
|
||||
import std/monotimes
|
||||
import std/locks
|
||||
|
||||
type
|
||||
RateLimitAlgo* = enum
|
||||
rlaTokenBucket
|
||||
rlaSlidingWindow
|
||||
rlaFixedWindow
|
||||
|
||||
TokenBucket* = ref object
|
||||
tokens: float64
|
||||
maxTokens: float64
|
||||
refillRate: float64 # tokens per second
|
||||
lastRefill: int64
|
||||
|
||||
SlidingWindow* = ref object
|
||||
windowSize: int64 # nanoseconds
|
||||
maxRequests: int
|
||||
timestamps: seq[int64]
|
||||
|
||||
RateLimiter* = ref object
|
||||
lock: Lock
|
||||
algo: RateLimitAlgo
|
||||
buckets: Table[string, TokenBucket]
|
||||
windows: Table[string, SlidingWindow]
|
||||
globalRate*: int
|
||||
perClientRate*: int
|
||||
|
||||
proc newTokenBucket*(maxTokens: float64, refillRate: float64): TokenBucket =
|
||||
TokenBucket(
|
||||
tokens: maxTokens,
|
||||
maxTokens: maxTokens,
|
||||
refillRate: refillRate,
|
||||
lastRefill: getMonoTime().ticks(),
|
||||
)
|
||||
|
||||
proc consume*(bucket: TokenBucket, tokens: float64 = 1.0): bool =
|
||||
let now = getMonoTime().ticks()
|
||||
let elapsed = float64(now - bucket.lastRefill) / 1_000_000_000.0
|
||||
bucket.tokens = min(bucket.maxTokens, bucket.tokens + elapsed * bucket.refillRate)
|
||||
bucket.lastRefill = now
|
||||
|
||||
if bucket.tokens >= 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)
|
||||
@@ -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..<payloadLen:
|
||||
if result.masked:
|
||||
result.payload[i] = byte(payloadData[i]) xor maskKey[i mod 4]
|
||||
else:
|
||||
result.payload[i] = byte(payloadData[i])
|
||||
|
||||
proc handleUpgrade*(client: AsyncSocket, requestHeaders: Table[string, string]): Future[WebSocket] {.async.} =
|
||||
let wsKey = requestHeaders.getOrDefault("Sec-WebSocket-Key", "")
|
||||
if wsKey.len == 0:
|
||||
return nil
|
||||
|
||||
let acceptKey = wsHandshakeKey(wsKey)
|
||||
let response = "HTTP/1.1 101 Switching Protocols\r\L" &
|
||||
"Upgrade: websocket\r\L" &
|
||||
"Connection: Upgrade\r\L" &
|
||||
"Sec-WebSocket-Accept: " & acceptKey & "\r\L\r\L"
|
||||
await client.send(response)
|
||||
return newWebSocket(client)
|
||||
|
||||
proc run*(server: WsServer) {.async.} =
|
||||
server.socket.setSockOpt(OptReuseAddr, true)
|
||||
server.socket.bindAddr(Port(server.port), server.address)
|
||||
server.socket.listen()
|
||||
|
||||
while true:
|
||||
let client = await server.socket.accept()
|
||||
let ws = newWebSocket(client)
|
||||
server.clients.add(ws)
|
||||
|
||||
if server.onConnect != nil:
|
||||
server.onConnect(ws)
|
||||
|
||||
# Read loop
|
||||
try:
|
||||
while ws.connected:
|
||||
let frame = await ws.readFrame()
|
||||
case frame.opcode
|
||||
of WS_TEXT, WS_BINARY:
|
||||
if server.onMessage != nil:
|
||||
server.onMessage(ws, frame.payload)
|
||||
of WS_PING:
|
||||
await ws.sendPong(frame.payload)
|
||||
of WS_CLOSE:
|
||||
ws.connected = false
|
||||
of WS_PONG:
|
||||
discard
|
||||
else:
|
||||
discard
|
||||
except:
|
||||
discard
|
||||
finally:
|
||||
ws.connected = false
|
||||
server.clients = server.clients.filterIt(it != ws)
|
||||
if server.onDisconnect != nil:
|
||||
server.onDisconnect(ws)
|
||||
|
||||
proc broadcast*(server: WsServer, data: seq[byte]) {.async.} =
|
||||
for client in server.clients:
|
||||
if client.connected:
|
||||
await client.sendBinary(data)
|
||||
|
||||
proc broadcastText*(server: WsServer, text: string) {.async.} =
|
||||
for client in server.clients:
|
||||
if client.connected:
|
||||
await client.sendText(text)
|
||||
|
||||
proc clientCount*(server: WsServer): int = server.clients.len
|
||||
@@ -0,0 +1,205 @@
|
||||
## SSTable Compaction — size-tiered and leveled strategies
|
||||
import std/tables
|
||||
import std/algorithm
|
||||
import std/os
|
||||
import std/math
|
||||
|
||||
const
|
||||
MaxLevel* = 7
|
||||
LevelMultiplier* = 10 # each level is 10x the previous
|
||||
|
||||
type
|
||||
SSTableMeta* = object
|
||||
path*: string
|
||||
level*: int
|
||||
minKey*: string
|
||||
maxKey*: string
|
||||
entryCount*: int
|
||||
sizeBytes*: int
|
||||
createdAt*: int64
|
||||
|
||||
CompactionResult* = object
|
||||
inputTables*: seq[SSTableMeta]
|
||||
outputTables*: seq[SSTableMeta]
|
||||
entriesRead*: int
|
||||
entriesWritten*: int
|
||||
|
||||
CompactionStrategy* = ref object
|
||||
levels*: seq[seq[SSTableMeta]]
|
||||
dataDir*: string
|
||||
maxSizePerLevel*: seq[int]
|
||||
|
||||
proc newCompactionStrategy*(dataDir: string): CompactionStrategy =
|
||||
result = CompactionStrategy(
|
||||
levels: newSeq[seq[SSTableMeta]](MaxLevel),
|
||||
dataDir: dataDir,
|
||||
maxSizePerLevel: newSeq[int](MaxLevel),
|
||||
)
|
||||
for i in 0..<MaxLevel:
|
||||
result.levels[i] = @[]
|
||||
result.maxSizePerLevel[i] = int(float64(1024 * 1024) * pow(float64(LevelMultiplier), float64(i))) # 1MB, 10MB, 100MB...
|
||||
|
||||
proc addTable*(cs: CompactionStrategy, meta: SSTableMeta) =
|
||||
if meta.level < MaxLevel:
|
||||
cs.levels[meta.level].add(meta)
|
||||
|
||||
proc totalSize*(cs: CompactionStrategy, level: int): int =
|
||||
result = 0
|
||||
for t in cs.levels[level]:
|
||||
result += t.sizeBytes
|
||||
|
||||
proc needsCompaction*(cs: CompactionStrategy, level: int): bool =
|
||||
if level >= 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..<count]
|
||||
|
||||
proc compact*(cs: CompactionStrategy, level: int): CompactionResult =
|
||||
let tables = cs.pickTablesForCompaction(level)
|
||||
if tables.len == 0:
|
||||
return CompactionResult()
|
||||
|
||||
var entriesRead = 0
|
||||
var allEntries: seq[(string, seq[byte], uint64, bool)] = @[]
|
||||
|
||||
for t in tables:
|
||||
entriesRead += t.entryCount
|
||||
# In real impl, would read SSTable file and merge
|
||||
# For now, simulate the merge
|
||||
|
||||
let outputPath = cs.dataDir / "sstables" / ("level_" & $level & "_" & $tables[0].createdAt & ".sst")
|
||||
let outputMeta = SSTableMeta(
|
||||
path: outputPath,
|
||||
level: level + 1,
|
||||
minKey: tables[0].minKey,
|
||||
maxKey: tables[^1].maxKey,
|
||||
entryCount: entriesRead,
|
||||
sizeBytes: entriesRead * 64, # estimate
|
||||
createdAt: tables[^1].createdAt,
|
||||
)
|
||||
|
||||
# Remove old tables from level
|
||||
var newTables: seq[SSTableMeta] = @[]
|
||||
for t in cs.levels[level]:
|
||||
var found = false
|
||||
for picked in tables:
|
||||
if t.path == picked.path:
|
||||
found = true
|
||||
break
|
||||
if not found:
|
||||
newTables.add(t)
|
||||
cs.levels[level] = newTables
|
||||
|
||||
# Add to next level
|
||||
if level + 1 < MaxLevel:
|
||||
cs.levels[level + 1].add(outputMeta)
|
||||
|
||||
return CompactionResult(
|
||||
inputTables: tables,
|
||||
outputTables: @[outputMeta],
|
||||
entriesRead: entriesRead,
|
||||
entriesWritten: entriesRead,
|
||||
)
|
||||
|
||||
proc levelCount*(cs: CompactionStrategy): int =
|
||||
result = 0
|
||||
for level in cs.levels:
|
||||
if level.len > 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
|
||||
@@ -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..<k]
|
||||
return candidates
|
||||
|
||||
proc newIVFPQIndex*(dimensions: int, nClusters: int = 100,
|
||||
nSubquantizers: int = 8, nBits: int = 8,
|
||||
metric: DistanceMetric = dmCosine): IVFPQIndex =
|
||||
|
||||
@@ -11,6 +11,7 @@ import barabadb/storage/bloom
|
||||
import barabadb/storage/wal
|
||||
import barabadb/storage/lsm
|
||||
import barabadb/storage/btree
|
||||
import barabadb/storage/compaction
|
||||
import barabadb/query/lexer as lex
|
||||
import barabadb/query/ast
|
||||
import barabadb/query/parser
|
||||
@@ -23,6 +24,7 @@ import barabadb/fts/engine as fts
|
||||
import barabadb/protocol/wire
|
||||
import barabadb/protocol/pool
|
||||
import barabadb/protocol/auth
|
||||
import barabadb/protocol/ratelimit
|
||||
import barabadb/schema/schema as schema
|
||||
|
||||
suite "Core Types":
|
||||
@@ -682,3 +684,101 @@ suite "Louvain Community Detection":
|
||||
|
||||
let matches = matchPattern(g, pattern)
|
||||
check matches.len >= 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
|
||||
|
||||
Reference in New Issue
Block a user