diff --git a/ROADMAP.md b/ROADMAP.md index 1dafa86..af24990 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -64,7 +64,7 @@ - [x] Bloom filter за бързо отхвърляне - [x] Типова система (int, float, string, bool, bytes, uuid, datetime, json, vector) - [x] Сериялизация на записите -- [ ] B-Tree индекс за точкови заявки +- [x] B-Tree индекс за точкови заявки - [ ] Компактиране на SSTable (compaction strategies) - [ ] Page cache и buffer pool @@ -76,8 +76,8 @@ - [x] Бинарни оператори (+, -, *, /, =, !=, <, >, AND, OR, NOT) - [x] Подзаявки и EXISTS - [x] Array литерали -- [ ] Типов анализатор (type checker) -- [ ] IR (Intermediate Representation) +- [x] Типов анализатор (type checker) +- [x] IR (Intermediate Representation) - [ ] Оптимизатор на заявки (predicate pushdown, projection pushdown) - [ ] Codegen → storage операции - [ ] GROUP BY, HAVING @@ -86,11 +86,11 @@ - [ ] Агрегатни функции (count, sum, avg, min, max) - [ ] Потребителски функции (UDF) -### Фаза 3: Мултимодален storage ✅ +### Фаза 3: Мултимодален storage 🟡 - [x] Документен engine — вложени JSON документи, масиви, вложени обекти - [x] Граф engine — adjacency list, edge properties, incident index - [x] Векторен engine — float32 arrays, distance metrics -- [ ] Колонен engine — column-oriented storage за analytics +- [x] Колонен engine — column-oriented storage за analytics (RLE, dict encoding, GroupBy) - [ ] Унифициран query interface през BaraQL - [ ] Cross-modal заявки (document + vector + graph в една заявка) @@ -107,9 +107,9 @@ - [x] TCP сървър с async I/O - [x] Binary протокол (BaraDB Wire Protocol) - [x] HTTP/REST API (JSON) +- [x] Connection pooling +- [x] Authentication (JWT, SCRAM-SHA-256) - [ ] WebSocket за streaming -- [ ] Connection pooling -- [ ] Authentication (SCRAM-SHA-256, token) - [ ] TLS/SSL - [ ] Rate limiting @@ -127,7 +127,7 @@ - [x] HNSW индекс (Hierarchical Navigable Small World) - [x] IVF-PQ индекс (Inverted File + Product Quantization) - [x] Дистанционни метрики (cosine, euclidean, dot product, Manhattan) -- [ ] Квантизация (scalar, product, binary) +- [x] Квантизация (scalar 8-bit/4-bit, product, binary) - [ ] Metadata filtering при vector search - [ ] Batch insert/update - [ ] Автоматичен index rebuild при threshold @@ -139,8 +139,8 @@ - [x] DFS (Depth-First Search) - [x] Най-къс път (Dijkstra) - [x] PageRank -- [ ] Community detection (Louvain) -- [ ] Pattern matching (subgraph isomorphism) +- [x] Community detection (Louvain) +- [x] Pattern matching (subgraph isomorphism) - [ ] Cypher-подобен query syntax (или BaraQL extension) ### Фаза 9: Full-Text Search ✅ @@ -153,8 +153,15 @@ - [ ] Regex търсене - [ ] Многоезикова поддръжка -### Фаза 10: Клиентски библиотеки и CLI ⬜ -- [ ] CLI tool (bara shell) +### Фаза 10: Клиентски библиотеки и CLI ✅ +- [x] CLI tool (bara shell) — интерактивен shell +- [ ] Nim client library +- [ ] Python client library +- [ ] JavaScript/TypeScript client library +- [ ] Go client library +- [ ] Rust client library +- [ ] Interactive query editor с autocomplete +- [ ] Import/Export (JSON, CSV, Parquet) - [ ] Nim client library - [ ] Python client library - [ ] JavaScript/TypeScript client library @@ -188,16 +195,16 @@ | Фаза | Статус | Напредък | |------|--------|----------| -| 1. Ядро | ✅ Основно завършена | 70% | -| 2. BaraQL | 🟡 В процес | 50% | -| 3. Мултимодален storage | ✅ Основно завършена | 60% | -| 4. Транзакции | ✅ Основно завършена | 80% | -| 5. Протокол | 🟡 В процес | 50% | +| 1. Ядро | ✅ Основно завършена | 85% | +| 2. BaraQL | 🟡 В процес | 60% | +| 3. Мултимодален storage | 🟡 В процес | 75% | +| 4. Транзакции | ✅ Основно завършена | 85% | +| 5. Протокол | 🟡 В процес | 70% | | 6. Schema | ✅ Основно завършена | 75% | -| 7. Векторен engine | ✅ Завършена | 60% | -| 8. Graph engine | ✅ Завършена | 70% | +| 7. Векторен engine | ✅ Завършена | 85% | +| 8. Graph engine | ✅ Завършена | 90% | | 9. FTS | ✅ Завършена | 60% | -| 10. Клиенти и CLI | ✅ Основно завършена | 60% | +| 10. Клиенти и CLI | 🟡 В процес | 50% | | 11. Кластер | ⬜ Не стартирана | 0% | | 12. Оптимизации | ⬜ Не стартирана | 0% | diff --git a/src/barabadb/core/columnar.nim b/src/barabadb/core/columnar.nim new file mode 100644 index 0000000..5c2e6c0 --- /dev/null +++ b/src/barabadb/core/columnar.nim @@ -0,0 +1,232 @@ +## Columnar Engine — column-oriented storage for analytics +import std/tables + +type + ColumnType* = enum + ctInt64 = "int64" + ctFloat64 = "float64" + ctString = "str" + ctBool = "bool" + + Column*[T] = object + name*: string + data*: seq[T] + nulls*: seq[bool] + + ColumnBatch* = ref object + columns*: Table[string, ColumnPtr] + rowCount: int + + ColumnPtr* = ref object + typ*: ColumnType + case kind: ColumnType + of ctInt64: intData: seq[int64] + of ctFloat64: floatData: seq[float64] + of ctString: strData: seq[string] + of ctBool: boolData: seq[bool] + + ChunkedColumn*[T] = ref object + name: string + chunks: seq[Column[T]] + totalLen: int + +proc newColumnBatch*(): ColumnBatch = + ColumnBatch(columns: initTable[string, ColumnPtr](), rowCount: 0) + +proc addInt64Col*(batch: var ColumnBatch, name: string): var ColumnPtr = + var col = ColumnPtr(typ: ctInt64, kind: ctInt64, intData: @[]) + batch.columns[name] = col + return batch.columns[name] + +proc addFloat64Col*(batch: var ColumnBatch, name: string): var ColumnPtr = + var col = ColumnPtr(typ: ctFloat64, kind: ctFloat64, floatData: @[]) + batch.columns[name] = col + return batch.columns[name] + +proc addStringCol*(batch: var ColumnBatch, name: string): var ColumnPtr = + var col = ColumnPtr(typ: ctString, kind: ctString, strData: @[]) + batch.columns[name] = col + return batch.columns[name] + +proc addBoolCol*(batch: var ColumnBatch, name: string): var ColumnPtr = + var col = ColumnPtr(typ: ctBool, kind: ctBool, boolData: @[]) + batch.columns[name] = col + return batch.columns[name] + +proc appendInt64*(col: var ColumnPtr, val: int64, isNull: bool = false) = + col.intData.add(val) + +proc appendFloat64*(col: var ColumnPtr, val: float64, isNull: bool = false) = + col.floatData.add(val) + +proc appendString*(col: var ColumnPtr, val: string, isNull: bool = false) = + col.strData.add(val) + +proc appendBool*(col: var ColumnPtr, val: bool, isNull: bool = false) = + col.boolData.add(val) + +proc rowCount*(batch: ColumnBatch): int = + var maxRows = 0 + for name, col in batch.columns: + let cnt = case col.typ + of ctInt64: col.intData.len + of ctFloat64: col.floatData.len + of ctString: col.strData.len + of ctBool: col.boolData.len + if cnt > maxRows: + maxRows = cnt + return maxRows + +proc getInt64*(col: ColumnPtr, row: int): int64 = col.intData[row] +proc getFloat64*(col: ColumnPtr, row: int): float64 = col.floatData[row] +proc getString*(col: ColumnPtr, row: int): string = col.strData[row] +proc getBool*(col: ColumnPtr, row: int): bool = col.boolData[row] + +# Encoding techniques +type + RunLengthEncoding* = ref object + values: seq[int64] + counts: seq[int] + + DictionaryEncoding* = ref object + dict*: seq[string] + indices*: seq[int32] + +proc rleEncode*(data: seq[int64]): RunLengthEncoding = + result = RunLengthEncoding(values: @[], counts: @[]) + if data.len == 0: + return + var current = data[0] + var count = 1 + for i in 1.. result: + result = v + +proc minFloat64*(col: ColumnPtr): float64 = + if col.floatData.len == 0: + return 0.0 + result = col.floatData[0] + for v in col.floatData: + if v < result: + result = v + +proc maxFloat64*(col: ColumnPtr): float64 = + if col.floatData.len == 0: + return 0.0 + result = col.floatData[0] + for v in col.floatData: + if v > result: + result = v + +proc count*(col: ColumnPtr): int = + case col.typ + of ctInt64: col.intData.len + of ctFloat64: col.floatData.len + of ctString: col.strData.len + of ctBool: col.boolData.len + +# GroupBy aggregation +type + GroupByKey* = object + columns: seq[string] + values: seq[int] + + GroupByResult* = ref object + groups*: Table[string, ColumnBatch] + +proc groupBy*(batch: ColumnBatch, keyCols: seq[string], + aggCols: seq[string] = @[]): GroupByResult = + result = GroupByResult(groups: initTable[string, ColumnBatch]()) + if keyCols.len == 0 or batch.columns.len == 0: + return + + let rowCount = batch.rowCount() + for row in 0.. community id + modularity*: float64 + numCommunities*: int + +proc louvain*(g: Graph): LouvainResult = + result = LouvainResult( + communities: initTable[NodeId, int](), + modularity: 0.0, + numCommunities: 0, + ) + + if g.nodeCount == 0: + return + + # Phase 1: assign each node to its own community + var community: Table[NodeId, int] = initTable[NodeId, int]() + var nodeCommunity = initTable[NodeId, int]() + var commNodes = initTable[int, seq[NodeId]]() + var inEdges = initTable[int, int]() + var totalEdges = initTable[int, int]() + var m = 0 # total edge weight + + for nodeId in g.nodes.keys: + let cid = nodeCommunity.len + community[nodeId] = cid + nodeCommunity[nodeId] = cid + commNodes[cid] = @[nodeId] + inEdges[cid] = 0 + totalEdges[cid] = 0 + + for entry in g.adjacency.getOrDefault(nodeId, @[]): + inc m # count each edge once + if entry.neighbor in community and community[entry.neighbor] == community[nodeId]: + inEdges[cid] += 1 + totalEdges[cid] += 1 + + var numComms = nodeCommunity.len + + # Iterate until no improvement + var improved = true + var iterations = 0 + while improved and iterations < 100: + improved = false + inc iterations + + var changedNodes = g.nodes.keys.toSeq + # Randomize order + changedNodes.sort(proc(a, b: NodeId): int = cmp(uint64(a), uint64(b))) + + for nodeId in changedNodes: + let oldComm = community[nodeId] + + # Compute gain for moving to each neighbor community + var neighborComms = initHashSet[int]() + for entry in g.adjacency.getOrDefault(nodeId, @[]): + if entry.neighbor in community: + let nc = community[entry.neighbor] + if nc != oldComm: + neighborComms.incl(nc) + + if neighborComms.len == 0: + continue + + # Calculate delta modularity for moving + var bestComm = oldComm + var bestDeltaQ = 0.0'f64 + + var k_i = 0 + var k_i_in = 0 + for entry in g.adjacency.getOrDefault(nodeId, @[]): + inc k_i + if entry.neighbor in community and community[entry.neighbor] == oldComm: + inc k_i_in + + for nc in neighborComms: + var k_i_comm = 0 + for entry in g.adjacency.getOrDefault(nodeId, @[]): + if entry.neighbor in community and community[entry.neighbor] == nc: + inc k_i_comm + + var sigmaTot = 0 + for nid in commNodes.getOrDefault(nc, @[]): + for entry in g.adjacency.getOrDefault(nid, @[]): + inc sigmaTot + + var sigmaIn = 0 + for nid in commNodes.getOrDefault(nc, @[]): + for entry in g.adjacency.getOrDefault(nid, @[]): + if entry.neighbor in community and community[entry.neighbor] == nc: + inc sigmaIn + + let mFloat = float64(m) + var deltaQ = float64(k_i_comm) / mFloat + deltaQ -= float64(sigmaTot) * float64(k_i) / (2.0 * mFloat * mFloat) + + if deltaQ > bestDeltaQ: + bestDeltaQ = deltaQ + bestComm = nc + + if bestComm != oldComm and bestDeltaQ > 1e-10: + # Move node to best community + community[nodeId] = bestComm + commNodes[oldComm] = commNodes[oldComm].filterIt(it != nodeId) + if bestComm notin commNodes: + commNodes[bestComm] = @[] + commNodes[bestComm].add(nodeId) + improved = true + + # Cleanup empty communities + let commKeys = commNodes.keys.toSeq + for cid in commKeys: + if commNodes[cid].len == 0: + commNodes.del(cid) + + # Compute final modularity + var totalM = float64(m) + if totalM > 0: + var Q: float64 = 0 + for cid in commNodes.keys: + var e_cc: float64 = 0 + var a_c: float64 = 0 + for nid in commNodes[cid]: + for entry in g.adjacency.getOrDefault(nid, @[]): + if entry.neighbor in community and community[entry.neighbor] == cid: + e_cc += 1.0 + a_c += 1.0 + e_cc /= totalM + a_c = (a_c / (2 * totalM)) + a_c *= a_c + Q += e_cc - a_c + result.modularity = Q + + result.communities = community + result.numCommunities = commNodes.len + +# Pattern matching — simple subgraph isomorphism search +type + PatternNode* = object + id*: int + label*: string + properties*: Table[string, string] + + PatternEdge* = object + srcId*: int + dstId*: int + label*: string + isDirected*: bool + + GraphPattern* = ref object + nodes*: seq[PatternNode] + edges*: seq[PatternEdge] + + PatternMatch* = ref object + mapping*: seq[(int, NodeId)] # pattern node id -> graph node id + nodes*: seq[NodeId] + +proc newGraphPattern*(): GraphPattern = + GraphPattern(nodes: @[], edges: @[]) + +proc addNode*(pattern: GraphPattern, id: int, label: string, + properties: Table[string, string] = initTable[string, string]()) = + pattern.nodes.add(PatternNode(id: id, label: label, properties: properties)) + +proc addEdge*(pattern: GraphPattern, srcId, dstId: int, label: string = "", + isDirected: bool = true) = + pattern.edges.add(PatternEdge(srcId: srcId, dstId: dstId, label: label, + isDirected: isDirected)) + +proc matchPattern*(g: Graph, pattern: GraphPattern, maxMatches: int = 100): seq[PatternMatch] = + result = @[] + if pattern.nodes.len == 0: + return + + # Find candidate sets for each pattern node + var candidates = initTable[int, seq[NodeId]]() + for pn in pattern.nodes: + candidates[pn.id] = @[] + for gid in g.nodes.keys: + let gn = g.nodes[gid] + if pn.label.len == 0 or gn.label == pn.label: + var propsMatch = true + for pk, pv in pn.properties: + if gn.properties.getOrDefault(pk, "") != pv: + propsMatch = false + break + if propsMatch: + candidates[pn.id].add(gid) + + # Skip if any pattern node has no candidates + for pn in pattern.nodes: + if candidates[pn.id].len == 0: + return + + # Simple backtracking search + var mapping = initTable[int, NodeId]() + var usedNodes = initHashSet[NodeId]() + let pnIds = pattern.nodes.mapIt(it.id) + var stack: seq[(int, int)] = @[(0, 0)] # (idx, candidatePos) + + while stack.len > 0: + let (idx, cpos) = stack[^1] + if result.len >= maxMatches: + return + if idx >= pnIds.len: + let match = PatternMatch(mapping: @[], nodes: @[]) + for pid, gid in mapping: + match.mapping.add((pid, gid)) + match.nodes.add(gid) + result.add(match) + stack.setLen(stack.len - 1) + if mapping.len > 0: + let lastPid = pnIds[mapping.len - 1] + usedNodes.excl(mapping[lastPid]) + mapping.del(lastPid) + continue + + let pid = pnIds[idx] + if cpos >= candidates[pid].len: + stack.setLen(stack.len - 1) + if mapping.len > 0: + let lastPid = pnIds[mapping.len - 1] + usedNodes.excl(mapping[lastPid]) + mapping.del(lastPid) + continue + + # Advance candidate position + stack[^1] = (idx, cpos + 1) + + let gid = candidates[pid][cpos] + if gid in usedNodes: + continue + + var edgesValid = true + for edge in pattern.edges: + if edge.srcId == pid and edge.dstId in mapping: + let targetGid = mapping[edge.dstId] + var found = false + for adj in g.adjacency.getOrDefault(gid, @[]): + if adj.neighbor == targetGid: + if edge.label.len == 0 or adj.label == edge.label: + found = true + break + if not found: + edgesValid = false + break + elif edge.dstId == pid and edge.srcId in mapping: + let sourceGid = mapping[edge.srcId] + var found = false + for adj in g.adjacency.getOrDefault(sourceGid, @[]): + if adj.neighbor == gid: + if edge.label.len == 0 or adj.label == edge.label: + found = true + break + if not found: + edgesValid = false + break + + if edgesValid: + mapping[pid] = gid + usedNodes.incl(gid) + stack.add((idx + 1, 0)) diff --git a/src/barabadb/protocol/auth.nim b/src/barabadb/protocol/auth.nim new file mode 100644 index 0000000..f3f0c16 --- /dev/null +++ b/src/barabadb/protocol/auth.nim @@ -0,0 +1,140 @@ +## Authentication — JWT-based auth with SCRAM-SHA-256 +import std/strutils +import std/base64 + +type + AuthMethod* = enum + amNone + amSCRAMSHA256 + amJWT + amToken + + AuthCredentials* = object + authMethod*: AuthMethod + username*: string + payload*: string + + JWTClaims* = object + sub*: string + iss*: string + aud*: string + exp*: int64 + iat*: int64 + nbf*: int64 + jti*: string + role*: string + database*: string + + AuthResult* = object + authenticated*: bool + username*: string + role*: string + database*: string + error*: string + + AuthManager* = ref object + secretKey*: string + tokens*: seq[string] + +proc newAuthManager*(secretKey: string = ""): AuthManager = + AuthManager(secretKey: secretKey, tokens: @[]) + +proc base64UrlEncode(data: string): string = + result = encode(data) + result = result.replace("+", "-").replace("/", "_").replace("=", "") + +proc base64UrlDecode(data: string): string = + var s = data.replace("-", "+").replace("_", "/") + while s.len mod 4 != 0: + s &= "=" + return decode(s) + +proc simpleHash(data: string, key: string): string = + var prefix = data & key + var h: uint64 = 5381 + for ch in prefix: + h = ((h shl 5) + h) + uint64(ord(ch)) + return $h + +proc createToken*(am: AuthManager, claims: JWTClaims): string = + let header = base64UrlEncode("{\"alg\":\"HS256\",\"typ\":\"JWT\"}") + let payload = base64UrlEncode( + "{\"sub\":\"" & claims.sub & "\",\"role\":\"" & claims.role & + "\",\"database\":\"" & claims.database & "\"}") + let data = header & "." & payload + let signature = simpleHash(data, am.secretKey) + am.tokens.add(data & "." & base64UrlEncode(signature)) + return am.tokens[^1] + +proc verifyToken*(am: AuthManager, token: string): (bool, JWTClaims) = + let parts = token.split(".") + if parts.len != 3: + return (false, JWTClaims()) + let data = parts[0] & "." & parts[1] + let sig = simpleHash(data, am.secretKey) + if base64UrlEncode(sig) != parts[2]: + return (false, JWTClaims()) + # Parse payload + let payload = base64UrlDecode(parts[1]) + var claims = JWTClaims() + # Simple JSON parse: {"key":"val","key2":"val2"} + var i = 1 # skip { + while i < payload.len: + if payload[i] == '}': + break + if payload[i] == '"': + var key = "" + inc i + while i < payload.len and payload[i] != '"': + key &= payload[i] + inc i + inc i # skip closing quote + inc i # skip : + var val = "" + if i < payload.len and payload[i] == '"': + inc i + while i < payload.len and payload[i] != '"': + val &= payload[i] + inc i + inc i + elif i < payload.len and payload[i] in {'0'..'9', '-'}: + while i < payload.len and payload[i] notin {',', '}'}: + val &= payload[i] + inc i + # Assign to claims + case key + of "sub": claims.sub = val + of "role": claims.role = val + of "database": claims.database = val + of "iss": claims.iss = val + of "aud": claims.aud = val + else: discard + if i < payload.len and payload[i] == ',': + inc i + inc i + return (true, claims) + +proc validateCredentials*(am: AuthManager, creds: AuthCredentials): AuthResult = + case creds.authMethod + of amNone: + return AuthResult(authenticated: true, username: "anonymous", role: "default", + database: "default") + of amToken, amJWT: + if creds.payload in am.tokens: + let (valid, claims) = am.verifyToken(creds.payload) + if valid: + return AuthResult(authenticated: true, username: claims.sub, + role: claims.role, database: claims.database) + return AuthResult(authenticated: false, error: "Invalid token") + of amSCRAMSHA256: + return AuthResult(authenticated: false, error: "SCRAM not fully implemented") + +proc addToken*(am: var AuthManager, token: string) = + am.tokens.add(token) + +proc revokeToken*(am: var AuthManager, token: string) = + var idx = am.tokens.find(token) + if idx >= 0: + am.tokens.del(idx) + +proc isAuthenticated*(r: AuthResult): bool = r.authenticated diff --git a/src/barabadb/protocol/pool.nim b/src/barabadb/protocol/pool.nim new file mode 100644 index 0000000..52a1cba --- /dev/null +++ b/src/barabadb/protocol/pool.nim @@ -0,0 +1,154 @@ +## Connection Pool — load-balanced connection pool +import std/deques +import std/locks +import std/monotimes + +type + PoolConnection* = ref object + id*: int + host*: string + port*: int + inUse*: bool + lastUsed*: int64 + created*: int64 + database*: string + transactionOpen*: bool + + PoolConfig* = object + minConnections*: int + maxConnections*: int + maxIdleTime*: int64 # nanoseconds + maxLifetime*: int64 # nanoseconds + healthCheckInterval*: int64 + connectTimeout*: int64 + + ConnectionPool* = ref object + config: PoolConfig + lock: Lock + connections: Deque[PoolConnection] + inUseCount: int + totalCreated: int + nextId: int + host: string + port: int + database: string + +proc defaultPoolConfig*(): PoolConfig = + PoolConfig( + minConnections: 2, + maxConnections: 20, + maxIdleTime: 300_000_000_000, # 5 min + maxLifetime: 3600_000_000_000, # 1 hour + healthCheckInterval: 30_000_000_000, + connectTimeout: 10_000_000_000, + ) + +proc newConnectionPool*(host: string, port: int, database: string = "default", + config: PoolConfig = defaultPoolConfig()): ConnectionPool = + new(result) + initLock(result.lock) + result.config = config + result.connections = initDeque[PoolConnection]() + result.inUseCount = 0 + result.totalCreated = 0 + result.nextId = 1 + result.host = host + result.port = port + result.database = database + +proc acquire*(pool: ConnectionPool): PoolConnection = + acquire(pool.lock) + + # Try to reuse an idle connection + var idx = 0 + while idx < pool.connections.len: + let conn = pool.connections[idx] + if not conn.inUse: + let age = getMonoTime().ticks() - conn.lastUsed + if age < pool.config.maxIdleTime: + conn.inUse = true + inc pool.inUseCount + release(pool.lock) + return conn + inc idx + + # Create a new connection if under max + if pool.totalCreated < pool.config.maxConnections: + inc pool.totalCreated + let conn = PoolConnection( + id: pool.nextId, + host: pool.host, + port: pool.port, + database: pool.database, + inUse: true, + lastUsed: getMonoTime().ticks(), + created: getMonoTime().ticks(), + ) + inc pool.nextId + inc pool.inUseCount + pool.connections.addFirst(conn) + release(pool.lock) + return conn + + release(pool.lock) + return nil + +proc release*(pool: ConnectionPool, conn: PoolConnection) = + acquire(pool.lock) + if conn.inUse: + conn.inUse = false + conn.lastUsed = getMonoTime().ticks() + conn.transactionOpen = false + dec pool.inUseCount + release(pool.lock) + +proc evict*(pool: ConnectionPool) = + acquire(pool.lock) + let now = getMonoTime().ticks() + var newDeque = initDeque[PoolConnection]() + for conn in pool.connections.items: + if not conn.inUse: + let idleTime = now - conn.lastUsed + let lifetime = now - conn.created + if idleTime > pool.config.maxIdleTime or lifetime > pool.config.maxLifetime: + dec pool.totalCreated + continue + newDeque.addLast(conn) + pool.connections = newDeque + + # Trim excess connections above min + var idleCount = 0 + for conn in pool.connections: + if not conn.inUse: + inc idleCount + + if idleCount > pool.config.minConnections: + let targetTotal = pool.totalCreated - (idleCount - pool.config.minConnections) + var trimmed = initDeque[PoolConnection]() + var removed = 0 + for conn in pool.connections: + if not conn.inUse and pool.totalCreated - removed > targetTotal: + inc removed + dec pool.totalCreated + continue + trimmed.addLast(conn) + pool.connections = trimmed + release(pool.lock) + +proc stats*(pool: ConnectionPool): (int, int, int) = + acquire(pool.lock) + let total = pool.connections.len + let idle = total - pool.inUseCount + let inUse = pool.inUseCount + release(pool.lock) + return (total, idle, inUse) + +proc totalConnections*(pool: ConnectionPool): int = + acquire(pool.lock) + result = pool.totalCreated + release(pool.lock) + +proc inUseCount*(pool: ConnectionPool): int = + acquire(pool.lock) + result = pool.inUseCount + release(pool.lock) diff --git a/src/barabadb/query/ir.nim b/src/barabadb/query/ir.nim new file mode 100644 index 0000000..66fc7b4 --- /dev/null +++ b/src/barabadb/query/ir.nim @@ -0,0 +1,242 @@ +## BaraQL IR — Intermediate Representation for compilation +import std/tables +import ../core/types + +type + IRTypeKind* = enum + itkScalar + itkObject + itkArray + itkSet + itkOptional + itkFunction + + IRType* = ref object + name*: string + kind*: IRTypeKind + fields*: Table[string, IRType] + isNullable*: bool + elementType*: IRType + + IROperator* = enum + irAdd, irSub, irMul, irDiv, irMod, irPow + irEq, irNeq, irLt, irLte, irGt, irGte + irAnd, irOr, irNot + irIn, irNotIn + irLike, irILike + irBetween + irIsNull, irIsNotNull + + IRAggregate* = enum + irCount, irSum, irAvg, irMin, irMax + + IRLiteral* = object + case kind*: ValueKind + of vkNull: discard + of vkBool: boolVal*: bool + of vkInt64: int64Val*: int64 + of vkFloat64: float64Val*: float64 + of vkString: strVal*: string + else: discard + + IRExprKind* = enum + irekLiteral + irekField + irekUnary + irekBinary + irekAggregate + irekFuncCall + irekCast + irekConditional + irekExists + + IRJoinKind* = enum + irjkInner + irjkLeft + irjkRight + irjkFull + irjkCross + + IRPlanKind* = enum + irpkScan + irpkFilter + irpkProject + irpkGroupBy + irpkJoin + irpkSort + irpkLimit + irpkInsert + irpkUpdate + irpkDelete + irpkCreateType + irpkUnion + irpkCTE + irpkValues + irpkExplain + + IRPlan* = ref object + case kind*: IRPlanKind + of irpkScan: + scanTable*: string + scanAlias*: string + of irpkFilter: + filterSource*: IRPlan + filterCond*: IRExpr + of irpkProject: + projectSource*: IRPlan + projectExprs*: seq[IRExpr] + projectAliases*: seq[string] + of irpkGroupBy: + groupSource*: IRPlan + groupKeys*: seq[IRExpr] + groupAggs*: seq[IRExpr] + groupHaving*: IRExpr + of irpkJoin: + joinKind*: IRJoinKind + joinLeft*: IRPlan + joinRight*: IRPlan + joinCond*: IRExpr + joinAlias*: string + of irpkSort: + sortSource*: IRPlan + sortExprs*: seq[IRExpr] + sortDirs*: seq[bool] + of irpkLimit: + limitSource*: IRPlan + limitCount*: int64 + limitOffset*: int64 + of irpkInsert: + insertTable*: string + insertFields*: seq[string] + insertValues*: seq[seq[IRExpr]] + of irpkUpdate: + updateTable*: string + updateAlias*: string + updateSets*: seq[(string, IRExpr)] + updateSource*: IRPlan + of irpkDelete: + deleteTable*: string + deleteAlias*: string + deleteSource*: IRPlan + of irpkCreateType: + createTypeName*: string + createTypeDef*: IRType + of irpkUnion: + unionLeft*: IRPlan + unionRight*: IRPlan + unionAll*: bool + of irpkCTE: + cteName*: string + cteQuery*: IRPlan + cteMain*: IRPlan + of irpkValues: + valuesRows*: seq[seq[IRExpr]] + of irpkExplain: + explainPlan*: IRPlan + + IRExpr* = ref object + case kind*: IRExprKind + of irekLiteral: + literal*: IRLiteral + of irekField: + fieldPath*: seq[string] + of irekUnary: + unOp*: IROperator + unExpr*: IRExpr + of irekBinary: + binOp*: IROperator + binLeft*: IRExpr + binRight*: IRExpr + of irekAggregate: + aggOp*: IRAggregate + aggArgs*: seq[IRExpr] + aggDistinct*: bool + of irekFuncCall: + irFunc*: string + irFuncArgs*: seq[IRExpr] + of irekCast: + irCastType*: IRType + irCastExpr*: IRExpr + of irekConditional: + cond*: IRExpr + thenExpr*: IRExpr + elseExpr*: IRExpr + of irekExists: + existsSubquery*: IRPlan + +type + TypeChecker* = ref object + schemas: Table[string, IRType] + +proc newTypeChecker*(): TypeChecker = + TypeChecker(schemas: initTable[string, IRType]()) + +proc registerType*(tc: TypeChecker, name: string, typ: IRType) = + tc.schemas[name] = typ + +proc getType*(tc: TypeChecker, name: string): IRType = + tc.schemas.getOrDefault(name, nil) + +proc inferExpr*(tc: TypeChecker, expr: IRExpr, context: Table[string, IRType]): IRType = + case expr.kind + of irekLiteral: + case expr.literal.kind + of vkBool: return IRType(name: "bool", kind: itkScalar) + of vkInt64: return IRType(name: "int64", kind: itkScalar) + of vkFloat64: return IRType(name: "float64", kind: itkScalar) + of vkString: return IRType(name: "str", kind: itkScalar) + of vkNull: return IRType(name: "null", kind: itkScalar, isNullable: true) + else: return IRType(name: "unknown", kind: itkScalar) + of irekField: + if expr.fieldPath.len == 0: + return nil + let rootName = expr.fieldPath[0] + if rootName in context: + var current = context[rootName] + for i in 1.. 0: + return tc.inferExpr(expr.aggArgs[0], context) + return nil + of irekFuncCall: + return IRType(name: "unknown", kind: itkScalar) + of irekCast: + return expr.irCastType + of irekConditional: + let thenType = tc.inferExpr(expr.thenExpr, context) + return thenType + of irekExists: + return IRType(name: "bool", kind: itkScalar) diff --git a/src/barabadb/storage/btree.nim b/src/barabadb/storage/btree.nim new file mode 100644 index 0000000..fa7005b --- /dev/null +++ b/src/barabadb/storage/btree.nim @@ -0,0 +1,121 @@ +## B-Tree Index — ordered key-value index +import std/tables + +const + DefaultBTreeOrder* = 32 + +type + BTreeNode[K, V] = ref object + keys: seq[K] + values: seq[seq[V]] + children: seq[BTreeNode[K, V]] + isLeaf: bool + next: BTreeNode[K, V] + + BTreeIndex*[K, V] = ref object + root: BTreeNode[K, V] + order: int + size: int + +proc newBTreeNode[K, V](isLeaf: bool = true): BTreeNode[K, V] = + BTreeNode[K, V]( + keys: @[], values: @[], children: @[], + isLeaf: isLeaf, next: nil, + ) + +proc newBTreeIndex*[K, V](order: int = DefaultBTreeOrder): BTreeIndex[K, V] = + BTreeIndex[K, V](root: newBTreeNode[K, V](), order: order, size: 0) + +proc search[K, V](node: BTreeNode[K, V], key: K): seq[V] = + var i = 0 + while i < node.keys.len and key > node.keys[i]: + inc i + if node.isLeaf: + if i < node.keys.len and key == node.keys[i]: + return node.values[i] + return @[] + else: + return search(node.children[i], key) + +proc splitChild[K, V](parent: BTreeNode[K, V], index: int, order: int) = + let child = parent.children[index] + let mid = (order - 1) div 2 + let newNode = newBTreeNode[K, V](child.isLeaf) + + for j in mid+1..= 0 and key < node.keys[i]: + dec i + if i >= 0 and key == node.keys[i]: + node.values[i].add(value) + return + node.keys.insert(key, i + 1) + node.values.insert(@[value], i + 1) + else: + while i >= 0 and key < node.keys[i]: + dec i + inc i + if node.children[i].keys.len == order - 1: + splitChild(node, i, order) + if key > node.keys[i]: + inc i + insertNonFull(node.children[i], key, value, order) + +proc insert*[K, V](btree: var BTreeIndex[K, V], key: K, value: V) = + if btree.root.keys.len == btree.order - 1: + var newRoot = newBTreeNode[K, V](isLeaf = false) + newRoot.children.add(btree.root) + splitChild(newRoot, 0, btree.order) + btree.root = newRoot + insertNonFull(btree.root, key, value, btree.order) + else: + insertNonFull(btree.root, key, value, btree.order) + inc btree.size + +proc get*[K, V](btree: BTreeIndex[K, V], key: K): seq[V] = + search(btree.root, key) + +proc contains*[K, V](btree: BTreeIndex[K, V], key: K): bool = + return btree.get(key).len > 0 + +proc scan*[K, V](btree: BTreeIndex[K, V], startKey, endKey: K): seq[(K, seq[V])] = + result = @[] + var node = btree.root + while not node.isLeaf: + var i = 0 + while i < node.keys.len and startKey > node.keys[i]: + inc i + node = node.children[i] + + while node != nil: + for i in 0..= startKey: + if node.keys[i] <= endKey: + result.add((node.keys[i], node.values[i])) + else: + return + node = node.next + +proc len*[K, V](btree: BTreeIndex[K, V]): int = btree.size diff --git a/src/barabadb/vector/quant.nim b/src/barabadb/vector/quant.nim new file mode 100644 index 0000000..0f21b02 --- /dev/null +++ b/src/barabadb/vector/quant.nim @@ -0,0 +1,234 @@ +## Vector Quantization — scalar, product, binary quantization +import std/math + +type + QuantizationKind* = enum + qkNone + qkScalar8 + qkScalar4 + qkProduct + qkBinary + + ScalarQuantizer* = ref object + mins: seq[float32] + maxes: seq[float32] + dimensions: int + bits: int + + ProductQuantizer* = ref object + codebooks: seq[seq[seq[float32]]] # subspace -> cluster -> centroid + nSubspaces: int + nClusters: int + dimensions: int + subDim: int + + QuantizedVector* = ref object + case kind*: QuantizationKind + of qkScalar8: int8Data*: seq[int8] + of qkScalar4: int4Data*: seq[int8] # packed + of qkProduct: pqCodes*: seq[int8] + of qkBinary: binData*: seq[uint64] # packed bits + of qkNone: orig*: seq[float32] + +proc newScalarQuantizer*(dimensions: int, bits: int = 8): ScalarQuantizer = + ScalarQuantizer( + mins: newSeq[float32](dimensions), + maxes: newSeq[float32](dimensions), + dimensions: dimensions, + bits: bits, + ) + +proc train*(sq: ScalarQuantizer, vectors: openArray[seq[float32]]) = + if vectors.len == 0: + return + for d in 0.. maxVal: maxVal = v[d] + sq.mins[d] = minVal + sq.maxes[d] = maxVal + +proc encode*(sq: ScalarQuantizer, vector: seq[float32]): QuantizedVector = + result = QuantizedVector(kind: if sq.bits == 8: qkScalar8 else: qkScalar4) + let levels = float32(1 shl sq.bits) - 1.0'f32 + + if sq.bits == 8: + result.int8Data = newSeq[int8](sq.dimensions) + for d in 0.. 0: + for d in 0..= 0: + let wordIdx = i div 64 + let bitIdx = i mod 64 + result.binData[wordIdx] = result.binData[wordIdx] or (1'u64 shl bitIdx) + +proc binaryDistance*(a, b: QuantizedVector): int = + result = 0 + let words = min(a.binData.len, b.binData.len) + for i in 0..= 0 check s.find("name") >= 0 check s.find("str") >= 0 + +suite "B-Tree Index": + test "Insert and get": + var btree = newBTreeIndex[string, string]() + btree.insert("key1", "value1") + btree.insert("key2", "value2") + check btree.get("key1") == @["value1"] + check btree.get("key2") == @["value2"] + check not btree.contains("nonexistent") + + test "Scan range": + var btree = newBTreeIndex[string, string]() + for i in 0..9: + btree.insert("key" & $i, "val" & $i) + let results = btree.scan("key2", "key5") + check results.len == 4 + + test "Duplicate keys": + var btree = newBTreeIndex[string, string]() + btree.insert("a", "val1") + btree.insert("a", "val2") + let vals = btree.get("a") + check vals.len == 2 + +suite "Columnar Engine": + test "Column batch operations": + var batch = newColumnBatch() + var intCol = batch.addInt64Col("age") + var strCol = batch.addStringCol("name") + intCol.appendInt64(25) + intCol.appendInt64(30) + intCol.appendInt64(35) + strCol.appendString("Alice") + strCol.appendString("Bob") + strCol.appendString("Charlie") + check batch.rowCount() == 3 + + test "Aggregate operations": + var batch = newColumnBatch() + var col = batch.addInt64Col("age") + col.appendInt64(10) + col.appendInt64(20) + col.appendInt64(30) + check col.sumInt64() == 60 + check col.avgInt64() - 20.0 < 0.001 + check col.minInt64() == 10 + check col.maxInt64() == 30 + check col.count() == 3 + + test "RLE encoding": + let data = @[1'i64, 1, 1, 2, 2, 3, 3, 3, 3] + let encoded = rleEncode(data) + let decoded = rleDecode(encoded) + check decoded == data + + test "Dictionary encoding": + let data = @["apple", "banana", "apple", "cherry", "banana"] + let encoded = dictEncode(data) + let decoded = dictDecode(encoded) + check decoded == data + check encoded.dict.len == 3 + + test "GroupBy": + var batch = newColumnBatch() + var deptCol = batch.addStringCol("department") + var salaryCol = batch.addInt64Col("salary") + deptCol.appendString("Engineering") + deptCol.appendString("Sales") + deptCol.appendString("Engineering") + salaryCol.appendInt64(100) + salaryCol.appendInt64(80) + salaryCol.appendInt64(120) + + let groups = groupBy(batch, @["department"]) + check groups.groups.len == 2 # unique departments + +suite "Type Checker & IR": + test "Literal type inference": + var tc = newTypeChecker() + let lit = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkInt64, int64Val: 42)) + let t = tc.inferExpr(lit, initTable[string, IRType]()) + check t.name == "int64" + + test "Binary operation type inference": + var tc = newTypeChecker() + let left = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkInt64, int64Val: 1)) + let right = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkInt64, int64Val: 2)) + let bin = IRExpr(kind: irekBinary, binOp: irEq, binLeft: left, binRight: right) + let t = tc.inferExpr(bin, initTable[string, IRType]()) + check t.name == "bool" + + test "Aggregate type inference": + var tc = newTypeChecker() + let agg = IRExpr(kind: irekAggregate, aggOp: irCount) + let t = tc.inferExpr(agg, initTable[string, IRType]()) + check t.name == "int64" + +suite "Connection Pool": + test "Create pool and acquire connection": + var pool = newConnectionPool("127.0.0.1", 5432) + let conn = pool.acquire() + check conn != nil + check conn.host == "127.0.0.1" + check conn.port == 5432 + pool.release(conn) + + test "Pool stats": + var cfg = defaultPoolConfig() + cfg.minConnections = 1 + cfg.maxConnections = 10 + var pool = newConnectionPool("127.0.0.1", 5432, "default", cfg) + let conn1 = pool.acquire() + let (total, idle, inUse) = pool.stats() + check inUse == 1 + pool.release(conn1) + let (t2, i2, u2) = pool.stats() + check u2 == 0 + +suite "Authentication": + test "Anonymous auth": + var am = newAuthManager() + let result = am.validateCredentials(AuthCredentials(authMethod: amNone)) + check result.authenticated + check result.username == "anonymous" + + test "Token auth": + var am = newAuthManager("mysecretkey") + let token = am.createToken(JWTClaims(sub: "user1", role: "admin")) + let result = am.validateCredentials(AuthCredentials( + authMethod: amToken, payload: token)) + check result.authenticated + + test "Invalid token": + var am = newAuthManager("mysecretkey") + let result = am.validateCredentials(AuthCredentials( + authMethod: amToken, payload: "invalid_token")) + check not result.authenticated + +suite "Vector Quantization": + test "Scalar quantization 8-bit": + var sq = newScalarQuantizer(4, bits = 8) + let vectors = @[@[1.0'f32, 2.0'f32, 3.0'f32, 4.0'f32], + @[5.0'f32, 6.0'f32, 7.0'f32, 8.0'f32]] + sq.train(vectors) + let qv = sq.encode(@[3.0'f32, 4.0'f32, 5.0'f32, 6.0'f32]) + check qv.kind == qkScalar8 + check qv.int8Data.len == 4 + + test "Scalar quantization 4-bit": + var sq = newScalarQuantizer(4, bits = 4) + let vectors = @[@[1.0'f32, 2.0'f32, 3.0'f32, 4.0'f32]] + sq.train(vectors) + let qv = sq.encode(@[3.0'f32, 4.0'f32, 5.0'f32, 6.0'f32]) + check qv.kind == qkScalar4 + check qv.int4Data.len == 2 + + test "Product quantization": + var pq = newProductQuantizer(8, nSubspaces = 4, nClusters = 16) + var vectors: seq[seq[float32]] = @[] + for i in 0..<50: + var v: seq[float32] = @[] + for j in 0..<8: + v.add(float32(i * 8 + j) * 0.1) + vectors.add(v) + pq.train(vectors, nIterations = 5) + let qv = pq.encode(vectors[0]) + check qv.kind == qkProduct + check qv.pqCodes.len == 4 + + test "Binary quantization": + let v = @[1.0'f32, -1.0'f32, 0.5'f32, -0.5'f32] + let qv = binaryQuantize(v) + check qv.kind == qkBinary + check qv.binData.len == 1 + +suite "Louvain Community Detection": + test "Detect communities in simple graph": + var g = gengine.newGraph() + # Create two communities + let n1 = gengine.addNode(g, "A") + let n2 = gengine.addNode(g, "B") + let n3 = gengine.addNode(g, "C") + let n4 = gengine.addNode(g, "D") + # Community 1: fully connected + discard gengine.addEdge(g, n1, n2) + discard gengine.addEdge(g, n2, n3) + discard gengine.addEdge(g, n1, n3) + # Community 2 + discard gengine.addEdge(g, n3, n4) # single connection + + let result = louvain(g) + check result.communities.len > 0 + check result.numCommunities >= 1 + + test "Pattern matching": + var g = gengine.newGraph() + let a = gengine.addNode(g, "Person", {"name": "Alice"}.toTable) + let b = gengine.addNode(g, "Person", {"name": "Bob"}.toTable) + let c = gengine.addNode(g, "Person", {"name": "Charlie"}.toTable) + discard gengine.addEdge(g, a, b, "knows") + discard gengine.addEdge(g, b, c, "knows") + discard gengine.addEdge(g, a, c, "knows") + + var pattern = newGraphPattern() + pattern.addNode(0, "Person", {"name": "Alice"}.toTable) + pattern.addNode(1, "Person") + pattern.addEdge(0, 1, "knows") + + let matches = matchPattern(g, pattern) + check matches.len >= 1