diff --git a/BARADB_BUGS_REPORT.md b/BARADB_BUGS_REPORT.md new file mode 100644 index 0000000..56d682f --- /dev/null +++ b/BARADB_BUGS_REPORT.md @@ -0,0 +1,181 @@ +# BaraDB Bug Report — NimForum Integration (Stress Test) + +**Date:** 2026-05-19 +**Reporter:** NimForum integration team +**BaraDB Version:** Latest (docker image `nimforum-baradb`) +**Severity:** High — prevents real-world multi-table applications from working + +--- + +## Summary + +During integration of the NimForum application (a real-world web forum) with BaraDB via the native TCP wire protocol, we hit **three distinct SQL-layer bugs** in BaraDB's query executor/parser. All of them are related to multi-table (implicit join) queries. The bugs are severe enough that any application using standard SQL patterns (table aliases, duplicate column names, or `IN (list)` predicates) will receive incorrect results or crash. + +--- + +## Bug 1: Duplicate column names in multi-table implicit joins cause 0 rows returned + +### Description +When a `SELECT` from multiple tables uses implicit join syntax (`FROM a, b WHERE ...`) and the selected columns include the **same column name from both tables** (e.g. `a.id` and `b.id`, or `a.name` and `b.name`) **without explicit `AS` aliases**, BaraDB returns **0 rows** even when matching data exists. + +### Reproducer + +```sql +-- Returns 0 rows (BUG) +SELECT thread.id, thread.name, category.id, category.name +FROM thread, category +WHERE thread.id = 3 + AND thread.isDeleted = 0 + AND thread.category = category.id; + +-- Returns 1 correct row (WORKAROUND) +SELECT t.id AS thread_id, t.name AS thread_name, + c.id AS cat_id, c.name AS cat_name +FROM thread t, category c +WHERE t.id = 3 + AND t.isDeleted = 0 + AND t.category = c.id; +``` + +### Expected behavior +Both queries should return 1 row with the thread and category data. + +### Actual behavior +The first query (without aliases) returns **0 rows**. + +### Root cause hypothesis +BaraDB's executor or result-set builder appears to use column names as keys internally. When two selected columns have the same name (`id`, `name`), the second overwrites the first, corrupting the row metadata and causing the executor to discard the row. + +### Impact +Any standard SQL query joining two tables on `id` columns (extremely common) will silently fail. + +### Workaround used +Add explicit `AS` aliases to every column in multi-table selects. + +--- + +## Bug 2: Column metadata corruption when duplicate names are present + +### Description +Even when rows *are* returned (e.g. in a multi-row list query without `WHERE id = ?`), if duplicate column names exist, the **column metadata and data values are scrambled**. Values from one column appear in another column's position. + +### Reproducer + +```sql +SELECT t.id, t.name, c.id, c.name, c.description, c.color +FROM thread t, category c +WHERE t.isDeleted = 0 AND t.category = c.id +ORDER BY modified DESC LIMIT 1; +``` + +### Expected behavior +| id | name | id | name | description | color | +|---|---|---|---|---|---| +| 3 | Test Thread2 | 1 | baradb | multimodal database engine written in Nim | 1a465b | + +### Actual behavior +Column metadata returned by BaraDB: +``` +Columns: ['id', 'name', 'views', "strftime('%s', \"modified\")", 'isLocked', 'isPinned', 'id', 'name', 'description', 'color'] +Row[0]: ['3', 'Test Thread2', 0, '1779200191', , None, , None, '1', 'baradb'] +``` +Values are shifted/missing for the second table's columns. + +### Impact +Client code receives corrupted data. In our case `category.description` became `"1"` and `category.color` became `"baradb"`, which would break UI rendering. + +### Workaround used +Same as Bug 1: explicit `AS` aliases for every column. + +--- + +## Bug 3: `IN (val1, val2, ...)` list syntax not supported + +### Description +BaraDB's SQL parser does not accept the standard `IN` predicate with a comma-separated value list. It throws a parse error. + +### Reproducer + +```sql +-- ERROR: Expected tkRParen but got tkComma at line 1 +SELECT id, name, email +FROM person +WHERE id IN (2, 1); +``` + +### Expected behavior +Should return rows for `id = 2` and `id = 1`. + +### Actual behavior +Parser error: `Expected tkRParen but got tkComma at line 1` + +### Impact +Any application that dynamically builds `IN (...)` lists (e.g. looking up multiple users by ID) cannot function without rewriting every query. + +### Workaround used +Replace `IN (2, 1)` with `id = 2 OR id = 1` generated dynamically in the application code. + +--- + +## Bug 4: `person.id` / `post.id` in three-table join produces `nkPath` column names + +### Description +In a three-table implicit join, when column references use `table.column` syntax, BaraDB's parser sometimes produces corrupted column names like `"strftime('%s', nkPath)"` instead of `"strftime('%s', post.creation)"`. + +### Reproducer + +```sql +SELECT post.id, strftime('%s', post.creation), post.thread, + person.id, person.name, person.email, + strftime('%s', person.lastOnline), + strftime('%s', person.previousVisitAt), person.usrStatus, + person.isDeleted, + thread.name +FROM post, person, thread +WHERE post.thread = thread.id + AND post.author = person.id + AND post.id = 10; +``` + +### Actual column metadata returned +``` +Columns: ['id', "strftime('%s', nkPath)", 'thread', 'id', 'name', 'email', + "strftime('%s', nkPath)", "strftime('%s', nkPath)", + 'usrStatus', 'isDeleted', 'name'] +``` + +### Impact +Result set metadata is unreadable. Client code cannot map columns to fields. + +### Workaround used +Use `AS` aliases for every expression and column: `p.id AS post_id`, `strftime('%s', p.creation) AS post_creation`, etc. + +--- + +## General Observations + +### What works reliably +- Single-table `SELECT`, `INSERT`, `UPDATE`, `DELETE` +- `SELECT` with sub-queries (`WHERE id IN (SELECT author FROM post WHERE thread = ?)`) +- `LIMIT`, `ORDER BY`, `WHERE` with single-table conditions +- `COUNT(*)`, `MIN()`, `MAX()` aggregates +- `strftime('%s', column)` expressions + +### What fails or is fragile +- Multi-table implicit joins (`FROM a, b WHERE a.id = b.id`) **without** `AS` aliases +- `IN (list)` with literal value lists +- `table.column` syntax inside `strftime()` or similar functions in multi-table queries + +### Recommendation for BaraDB team +1. **Fix column deduplication in result sets** — the executor should not use raw column names as internal keys; it should use ordinal positions or alias names. +2. **Support `IN (val1, val2, ...)`** — extend the parser to accept comma-separated expressions inside `IN (...)`. +3. **Investigate `nkPath` leak in parser** — `post.creation` inside `strftime()` should not be tokenized into `nkPath` in the column metadata output. + +--- + +## Appendix: Test Environment + +- **Client:** Custom synchronous Nim TCP client (`baradb_sync_client.nim`) using `net.Socket` +- **Wire protocol:** Binary protocol with `mkQuery` (0x02), `mkData` (0x82), `mkComplete` (0x83) +- **Result format:** `rfBinary` (0x00) +- **Connection:** Fresh socket per query (to rule out interleaving) diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 3a5e8fc..7f51768 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -6,13 +6,13 @@ import hunos/context import json import tables import strutils -import os import times import std/asyncdispatch import config import ../query/lexer import ../query/parser import ../query/executor +import ../core/types import ../storage/lsm import ../core/mvcc import ../protocol/wire @@ -20,6 +20,7 @@ import ../core/websocket import jwt as jwtlib import ../protocol/auth import ../protocol/ratelimit +import ../core/registry type HttpServer* = ref object @@ -27,6 +28,7 @@ type running: bool db*: LSMTree ctx: ExecutionContext + registry*: DatabaseRegistry metrics*: Metrics secretKey*: string authManager*: AuthManager @@ -40,8 +42,10 @@ type selectCount*: int activeConnections*: int -proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer = - let ctx = newExecutionContext(db) +proc newHttpServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): HttpServer = + let dbInfo = getOrCreateDatabase(registry, "default") + let db = dbInfo.db + let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx)) ctx.txnManager = newTxnManager() let secret = config.getEffectiveJwtSecret() let ws = newWsServer(config, secret) @@ -51,15 +55,27 @@ proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer = asyncCheck ws.broadcastToTable(ev.table, msg) let am = newAuthManager(secret) HttpServer(config: config, running: false, db: db, ctx: ctx, + registry: registry, secretKey: secret, authManager: am, rateLimiter: rl, metrics: Metrics(), ws: ws) +proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer = + let registry = newDatabaseRegistry(config) + registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} = + cast[ContextRef](cast[pointer](newExecutionContext(d, r)))) + let ctx = newExecutionContext(db, registry) + registry.setDatabase("default", db, cast[ContextRef](cast[pointer](ctx))) + return newHttpServerWithRegistry(config, registry) + proc newHttpServer*(config: BaraConfig): HttpServer = - let dataDir = config.dataDir / "server" - let db = newLSMTree(dataDir) - return newHttpServerWithDb(config, db) + let registry = newDatabaseRegistry(config) + registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} = + cast[ContextRef](cast[pointer](newExecutionContext(d, r)))) + registry.loadExistingDatabases() + registry.ensureDefaultDatabase() + return newHttpServerWithRegistry(config, registry) # ---------------------------------------------------------------------- # JWT helpers @@ -178,8 +194,8 @@ proc queryHandler(server: HttpServer): RequestHandler = var jsonRow = newJObject() for col in res.columns: let key = col - if key in row and not isNull(row[key]): - jsonRow[key] = %row[key] + if key in row and row[key].kind != vkNull: + jsonRow[key] = %valueToString(row[key]) else: jsonRow[key] = newJNull() jsonRows.add(jsonRow) @@ -564,4 +580,7 @@ proc run*(server: HttpServer, port: int = 9470) = proc stop*(server: HttpServer) = server.running = false server.ws.stop() - server.db.close() + if server.registry != nil: + server.registry.closeAll() + else: + server.db.close() diff --git a/src/barabadb/core/registry.nim b/src/barabadb/core/registry.nim new file mode 100644 index 0000000..4c28528 --- /dev/null +++ b/src/barabadb/core/registry.nim @@ -0,0 +1,192 @@ +## BaraDB Database Registry — manages per-database LSMTree instances +import std/tables +import std/os +import std/strutils +import std/locks +import std/algorithm +import logging +import config +import ../storage/lsm + +type + ContextRef* = ref RootObj + + ContextFactory* = proc(db: LSMTree, reg: DatabaseRegistry): ContextRef {.closure.} + + DatabaseInfo* = ref object + name*: string + db*: LSMTree + ctx*: ContextRef + activeConnections*: int + + DatabaseRegistry* = ref object + config*: BaraConfig + databases: Table[string, DatabaseInfo] + lock*: Lock + defaultDbName*: string + dataRoot*: string + ctxFactory*: ContextFactory + +const reservedDbNames* = ["system", "information_schema", "pg_catalog"] + +proc isValidDbName*(name: string): bool = + if name.len == 0: return false + if '/' in name or '\\' in name: return false + if name in reservedDbNames: return false + if name.startsWith("_"): return false + if name == ".." or name == ".": return false + true + +proc newDatabaseRegistry*(config: BaraConfig, defaultDbName = "default"): DatabaseRegistry = + new(result) + result.config = config + result.databases = initTable[string, DatabaseInfo]() + result.defaultDbName = defaultDbName + result.dataRoot = config.dataDir / "databases" + initLock(result.lock) + + # Create root directory + if not dirExists(result.dataRoot): + createDir(result.dataRoot) + +proc setContextFactory*(reg: DatabaseRegistry, factory: ContextFactory) = + reg.ctxFactory = factory + +proc loadExistingDatabases*(reg: DatabaseRegistry) = + if reg.ctxFactory == nil: + raise newException(ValueError, "Context factory not set. Call setContextFactory first.") + + # Scan for existing databases + for kind, path in walkDir(reg.dataRoot): + if kind == pcDir: + let dbName = path.splitPath().tail + if dbName.len > 0 and isValidDbName(dbName): + let dbDir = reg.dataRoot / dbName + info("Loading database '" & dbName & "' from " & dbDir) + let db = newLSMTree(dbDir) + let ctx = reg.ctxFactory(db, reg) + reg.databases[dbName] = DatabaseInfo( + name: dbName, db: db, ctx: ctx, activeConnections: 0 + ) + +proc setDatabase*(reg: DatabaseRegistry, name: string, db: LSMTree, ctx: ContextRef) = + acquire(reg.lock) + defer: release(reg.lock) + reg.databases[name] = DatabaseInfo( + name: name, db: db, ctx: ctx, activeConnections: 0) + +proc ensureDefaultDatabase*(reg: DatabaseRegistry) = + if reg.ctxFactory == nil: + raise newException(ValueError, "Context factory not set. Call setContextFactory first.") + + let defaultDbName = reg.defaultDbName + if defaultDbName notin reg.databases: + let dbDir = reg.dataRoot / defaultDbName + info("Creating default database at " & dbDir) + let db = newLSMTree(dbDir) + let ctx = reg.ctxFactory(db, reg) + reg.databases[defaultDbName] = DatabaseInfo( + name: defaultDbName, db: db, ctx: ctx, activeConnections: 0 + ) + +proc getOrCreateDatabase*(reg: DatabaseRegistry, name: string): DatabaseInfo = + if not isValidDbName(name): + raise newException(ValueError, "Invalid database name: " & name) + + if reg.ctxFactory == nil: + raise newException(ValueError, "Context factory not set. Call setContextFactory first.") + + acquire(reg.lock) + defer: release(reg.lock) + + if name in reg.databases: + return reg.databases[name] + + # Create new database + let dbDir = reg.dataRoot / name + info("Creating database '" & name & "' at " & dbDir) + let db = newLSMTree(dbDir) + let ctx = reg.ctxFactory(db, reg) + let info = DatabaseInfo(name: name, db: db, ctx: ctx, activeConnections: 0) + reg.databases[name] = info + info + +proc getConnectionCount*(reg: DatabaseRegistry, name: string): int = + acquire(reg.lock) + defer: release(reg.lock) + if name in reg.databases: + return reg.databases[name].activeConnections + return 0 + +proc incrementConnections*(reg: DatabaseRegistry, name: string) = + acquire(reg.lock) + defer: release(reg.lock) + if name in reg.databases: + inc reg.databases[name].activeConnections + +proc decrementConnections*(reg: DatabaseRegistry, name: string) = + acquire(reg.lock) + defer: release(reg.lock) + if name in reg.databases and reg.databases[name].activeConnections > 0: + dec reg.databases[name].activeConnections + +proc dropDatabase*(reg: DatabaseRegistry, name: string): bool = + if not isValidDbName(name): + return false + + acquire(reg.lock) + defer: release(reg.lock) + + if name notin reg.databases: + return false + + if name == reg.defaultDbName: + raise newException(ValueError, "Cannot drop the default database") + + let info = reg.databases[name] + if info.activeConnections > 0: + raise newException(ValueError, + "Cannot drop database '" & name & "': " & + $info.activeConnections & " active connections") + + # Close LSMTree + info.db.close() + + # Remove data directory + let dbDir = reg.dataRoot / name + if dirExists(dbDir): + removeDir(dbDir) + + reg.databases.del(name) + true + +proc listDatabases*(reg: DatabaseRegistry): seq[string] = + acquire(reg.lock) + defer: release(reg.lock) + result = @[] + for name in reg.databases.keys: + result.add(name) + result.sort() + +proc databaseExists*(reg: DatabaseRegistry, name: string): bool = + acquire(reg.lock) + defer: release(reg.lock) + name in reg.databases + +proc getDatabaseInfo*(reg: DatabaseRegistry, name: string): DatabaseInfo = + acquire(reg.lock) + defer: release(reg.lock) + if name in reg.databases: + return reg.databases[name] + return nil + +proc closeAll*(reg: DatabaseRegistry) = + acquire(reg.lock) + defer: release(reg.lock) + for name, info in reg.databases.pairs: + try: + info.db.close() + info("Database '" & name & "' closed") + except CatchableError as e: + warn("Error closing database '" & name & "': " & e.msg) + reg.databases.clear() diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index e4fe9e1..ff90d5d 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -4,7 +4,6 @@ import std/asyncnet import std/strutils import std/sequtils import std/tables -import std/os import std/endians import std/monotimes import std/locks @@ -28,6 +27,7 @@ import ../core/replication import ../core/sharding import ../core/gossip import ../protocol/ratelimit +import ../core/registry import jwt as jwtlib type @@ -36,6 +36,7 @@ type running*: bool db*: LSMTree ctx*: ExecutionContext + registry*: DatabaseRegistry txnManager*: TxnManager distTxnManager*: DistTxnManager replicationManager*: ReplicationManager @@ -47,8 +48,10 @@ type activeConnections*: int activeConnectionsLock*: Lock -proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server = - let ctx = newExecutionContext(db) +proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Server = + let dbInfo = getOrCreateDatabase(registry, "default") + let db = dbInfo.db + let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx)) ctx.txnManager = newTxnManager() var tls: TLSContext = nil if config.tlsEnabled and config.certFile.len > 0 and config.keyFile.len > 0: @@ -60,7 +63,7 @@ proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server = let localId = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port let cm = newClusterMembership(shardRouter, localId) - # Wire shard migration callbacks to LSM + # Wire shard migration callbacks to LSM (use default database) shardRouter.iterateKeys = proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.} = var entries: seq[(string, seq[byte])] = @[] for (key, value) in db.scanAll(): @@ -94,6 +97,7 @@ proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server = let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient) result = Server(config: config, running: false, db: db, ctx: ctx, + registry: registry, txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), replicationManager: newReplicationManager(), shardRouter: shardRouter, @@ -103,10 +107,22 @@ proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server = rateLimiter: rl) initLock(result.activeConnectionsLock) +proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server = + let registry = newDatabaseRegistry(config) + let ctx = newExecutionContext(db, registry) + registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} = + cast[ContextRef](cast[pointer](newExecutionContext(d, r)))) + # Use the existing db for default + registry.setDatabase("default", db, cast[ContextRef](cast[pointer](ctx))) + return newServerWithRegistry(config, registry) + proc newServer*(config: BaraConfig): Server = - let dataDir = config.dataDir / "server" - let db = newLSMTree(dataDir) - return newServerWithDb(config, db) + let registry = newDatabaseRegistry(config) + registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} = + cast[ContextRef](cast[pointer](newExecutionContext(d, r)))) + registry.loadExistingDatabases() + registry.ensureDefaultDatabase() + return newServerWithRegistry(config, registry) # ---------------------------------------------------------------------- # Wire Protocol Helpers @@ -230,7 +246,7 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq for row in res.rows: var wireRow: seq[WireValue] = @[] for i, col in res.columns: - let val = if col in row: row[col] else: "\\N" + let val = if col in row: valueToString(row[col]) else: "\\N" let cType = if i < colTypes.len: colTypes[i] else: "" wireRow.add(valueToWire(val, cType)) qr.rows.add(wireRow) @@ -312,16 +328,17 @@ proc slowQueryLog(logPath: string, query: string, durationMs: int, clientId: int f.write(line) except IOError: discard -proc verifyToken(secret, tokenStr: string): (bool, string, string) = +proc verifyToken(secret, tokenStr: string): (bool, string, string, string) = try: let token = tokenStr.toJWT() if not token.verify(secret, HS256): - return (false, "", "") + return (false, "", "", "") let userId = token.claims["sub"].node.str let role = if "role" in token.claims: token.claims["role"].node.str else: "user" - return (true, userId, role) + let database = if "database" in token.claims: token.claims["database"].node.str else: "" + return (true, userId, role, database) except ValueError, KeyError: - return (false, "", "") + return (false, "", "", "") proc recvWithTimeout(client: AsyncSocket, size: int, timeoutMs: int): Future[string] {.async.} = if timeoutMs <= 0: @@ -475,7 +492,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} case header.kind of mkAuth: let tokenStr = parseAuthMessage(stringToBytes(payload)) - let (valid, userId, role) = verifyToken(secret, tokenStr) + let (valid, userId, role, jwtDatabase) = verifyToken(secret, tokenStr) if valid: authenticated = true connCtx.currentUser = userId @@ -483,6 +500,24 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} let okMsg = makeAuthOkMessage(header.requestId) await client.send(bytesToString(okMsg)) info("Client " & $clientId & " authenticated as " & userId) + # Switch to database from JWT claim if provided + if jwtDatabase.len > 0 and server.registry != nil and isValidDbName(jwtDatabase): + let info = getDatabaseInfo(server.registry, jwtDatabase) + if info != nil: + let targetCtx = cast[ExecutionContext](cast[pointer](info.ctx)) + connCtx.db = info.db + connCtx.tables = targetCtx.tables + connCtx.btrees = targetCtx.btrees + connCtx.views = targetCtx.views + connCtx.ftsIndexes = targetCtx.ftsIndexes + connCtx.vectorIndexes = targetCtx.vectorIndexes + connCtx.users = targetCtx.users + connCtx.policies = targetCtx.policies + connCtx.graphs = targetCtx.graphs + connCtx.autoIncCounters = targetCtx.autoIncCounters + connCtx.sequences = targetCtx.sequences + connCtx.currentDatabase = jwtDatabase + incrementConnections(server.registry, jwtDatabase) else: let err = makeErrorMessage(header.requestId, 403, "Invalid token") await client.send(bytesToString(err)) @@ -510,7 +545,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} if shardCheck: let startTicks = getMonoTime().ticks() - let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, replication=server.replicationManager) + let (success, result, errorMsg) = executeQuery(connCtx.db, connCtx, queryStr, replication=server.replicationManager) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: @@ -530,7 +565,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)") let startTicks = getMonoTime().ticks() - let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, params, replication=server.replicationManager) + let (success, result, errorMsg) = executeQuery(connCtx.db, connCtx, queryStr, params, replication=server.replicationManager) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: @@ -562,6 +597,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} except Exception as e: errorMsg("Client " & $clientId & " error: " & e.msg) finally: + # Decrement database connection counter + if server.registry != nil and connCtx.currentDatabase.len > 0: + decrementConnections(server.registry, connCtx.currentDatabase) acquire(server.activeConnectionsLock) try: if server.activeConnections > 0: @@ -629,4 +667,7 @@ proc stop*(server: Server) = server.running = false if server.gossipProtocol != nil: server.gossipProtocol.stop() - server.db.close() + if server.registry != nil: + server.registry.closeAll() + else: + server.db.close() diff --git a/src/barabadb/core/types.nim b/src/barabadb/core/types.nim index 41fdf03..1c7737f 100644 --- a/src/barabadb/core/types.nim +++ b/src/barabadb/core/types.nim @@ -1,6 +1,7 @@ import std/times import std/oids import std/monotimes +import std/strutils type ValueKind* = enum @@ -124,3 +125,60 @@ proc newRecordId*(): RecordId = proc `==`*(a, b: RecordId): bool {.borrow.} proc `$`*(r: RecordId): string = $uint64(r) + +proc `==`*(a, b: Value): bool {.noSideEffect.} = + if a.kind != b.kind: return false + case a.kind + of vkNull: return true + of vkBool: return a.boolVal == b.boolVal + of vkInt8: return a.int8Val == b.int8Val + of vkInt16: return a.int16Val == b.int16Val + of vkInt32: return a.int32Val == b.int32Val + of vkInt64: return a.int64Val == b.int64Val + of vkFloat32: return a.float32Val == b.float32Val + of vkFloat64: return a.float64Val == b.float64Val + of vkString: return a.strVal == b.strVal + of vkBytes: return a.bytesVal == b.bytesVal + of vkUuid: return a.uuidVal == b.uuidVal + of vkDateTime: return false # DateTime comparison not supported without side effects + of vkJson: return a.jsonVal == b.jsonVal + of vkArray: return false # Recursive comparison not supported + of vkObject: return false # Recursive comparison not supported + of vkVector: return a.vecVal == b.vecVal + +proc `!=`*(a, b: Value): bool {.noSideEffect.} = + return not (a == b) + +proc `$`*(v: Value): string = + case v.kind + of vkNull: return "\\N" + of vkBool: return $v.boolVal + of vkInt8: return $v.int8Val + of vkInt16: return $v.int16Val + of vkInt32: return $v.int32Val + of vkInt64: return $v.int64Val + of vkFloat32: return $v.float32Val + of vkFloat64: return $v.float64Val + of vkString: return v.strVal + of vkBytes: return "" + of vkUuid: return $v.uuidVal + of vkDateTime: return $v.dtVal + of vkJson: return v.jsonVal + of vkArray: return $v.arrayVal + of vkObject: return $v.objVal + of vkVector: return $v.vecVal + +proc `==`*(a: Value, b: string): bool = + if a.kind == vkString: return a.strVal == b + if a.kind == vkNull: return b == "\\N" + return $a == b + +proc `==`*(a: string, b: Value): bool = + return b == a + +proc `in`*(v: Value, s: seq[string]): bool = + return $v in s + +proc `in`*(s: string, v: Value): bool = + if v.kind == vkString: return v.strVal.contains(s) + return ($v).contains(s) diff --git a/src/barabadb/query/ast.nim b/src/barabadb/query/ast.nim index e5931cf..7e91d2d 100644 --- a/src/barabadb/query/ast.nim +++ b/src/barabadb/query/ast.nim @@ -43,6 +43,10 @@ type nkSetVar nkCreateGraph nkDropGraph + nkCreateDatabase + nkDropDatabase + nkUseDatabase + nkShowDatabases # Clauses nkFrom @@ -336,6 +340,16 @@ type of nkDropGraph: dgName*: string dgIfExists*: bool + of nkCreateDatabase: + cdDbName*: string + cdIfNotExists*: bool + of nkDropDatabase: + ddDbName*: string + ddIfExists*: bool + of nkUseDatabase: + udDbName*: string + of nkShowDatabases: + discard of nkApplyMigration: amName*: string of nkMigrationStatus: diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index 52f295d..bf97d9d 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -158,7 +158,7 @@ type fkOnUpdate*: string autoIncrement*: bool - Row* = Table[string, string] + Row* = Table[string, Value] ExecResult* = object success*: bool @@ -501,6 +501,29 @@ proc getValue(values: seq[string], fields: seq[string], colName: string): string proc isNull*(value: string): bool = value == "\\N" or value.toLower() == "null" +proc valueToString*(v: Value): string = + case v.kind + of vkNull: return "\\N" + of vkString: return v.strVal + of vkInt64: return $v.int64Val + of vkFloat64: return $v.float64Val + of vkBool: return $v.boolVal + else: return "" + +proc `%`*(v: Value): JsonNode = + case v.kind + of vkNull: return newJNull() + of vkString: return %v.strVal + of vkInt64: return %v.int64Val + of vkFloat64: return %v.float64Val + of vkBool: return %v.boolVal + else: return newJNull() + +proc toString*(v: Value): string = valueToString(v) + +proc `[]=`*(t: var Row, key: string, val: string) = + t[key] = Value(kind: vkString, strVal: val) + proc escapeRowVal(v: string): string = v.replace("\\", "\\\\").replace(",", "\\,").replace("=", "\\=") @@ -546,6 +569,11 @@ proc parseRowData(valStr: string): Table[string, string] = let v = unescapeRowVal(part[eqPos+1..^1].strip()) result[k] = v +proc parseRowDataToValueRow(valStr: string): Row = + result = initTable[string, Value]() + for k, v in parseRowData(valStr): + result[k] = v + proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] proc extractJoinEquality*(expr: IRExpr): (string, string) = @@ -716,12 +744,12 @@ proc realIdFromKey(key: string): string = proc findRealIdByDocId(ctx: ExecutionContext, table: string, docId: uint64): string = for row in execScan(ctx, table): if "$key" in row: - let docKey = table & "." & row["$key"] + let docKey = table & "." & valueToString(row["$key"]) var hash: uint64 = 0 for ch in docKey: hash = hash * 31 + uint64(ord(ch)) if hash == docId: - return realIdFromKey(row["$key"]) + return realIdFromKey(valueToString(row["$key"])) return "" proc doHybridSearch(ctx: ExecutionContext, table: string, vecCol: string, textCol: string, @@ -809,8 +837,8 @@ proc doHybridSearchFiltered(ctx: ExecutionContext, table: string, vecCol: string if filterCol.len > 0: passesFilter = false for row in execScan(ctx, table): - if realIdFromKey(row.getOrDefault("$key", "")) == realId: - if filterCol in row and row[filterCol] == filterVal: + if realIdFromKey(valueToString(row.getOrDefault("$key", Value(kind: vkNull)))) == realId: + if filterCol in row and valueToString(row[filterCol]) == filterVal: passesFilter = true break if passesFilter: @@ -829,9 +857,11 @@ proc doHybridSearchFiltered(ctx: ExecutionContext, table: string, vecCol: string sorted = sorted[0.. 0: let fullPath = expr.fieldPath.join(".") var s = "" - if fullPath in row: s = row[fullPath] + var found = false + if fullPath in row: + s = valueToString(row[fullPath]) + found = true else: let colName = expr.fieldPath[^1] - if colName in row: s = row[colName] - elif "$key" in row and row["$key"].startsWith(colName & "="): - s = row["$key"][colName.len+1..^1] + if colName in row: + s = valueToString(row[colName]) + found = true + elif "$key" in row and valueToString(row["$key"]).startsWith(colName & "="): + s = valueToString(row["$key"])[colName.len+1..^1] + found = true elif "$value" in row: - let parsed = parseRowData(row["$value"]) - if colName in parsed: s = parsed[colName] + let parsed = parseRowData(valueToString(row["$value"])) + if colName in parsed: + s = parsed[colName] + found = true if s == "\\N": return Value(kind: vkNull) + if not found: return Value(kind: vkNull) case expr.valueKind of vkInt64: try: return Value(kind: vkInt64, int64Val: parseInt(s)) @@ -881,8 +920,8 @@ proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionCont of irekBinary: case expr.binOp of irAdd: - let lv = evalExprValue(expr.binLeft, row, ctx) - let rv = evalExprValue(expr.binRight, row, ctx) + let lv = evalExpr(expr.binLeft, row, ctx) + let rv = evalExpr(expr.binRight, row, ctx) if lv.kind == vkInt64 and rv.kind == vkInt64: return Value(kind: vkInt64, int64Val: lv.int64Val + rv.int64Val) elif lv.kind == vkFloat64 and rv.kind == vkFloat64: @@ -896,8 +935,8 @@ proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionCont else: return Value(kind: vkNull) of irSub: - let lv = evalExprValue(expr.binLeft, row, ctx) - let rv = evalExprValue(expr.binRight, row, ctx) + let lv = evalExpr(expr.binLeft, row, ctx) + let rv = evalExpr(expr.binRight, row, ctx) if lv.kind == vkInt64 and rv.kind == vkInt64: return Value(kind: vkInt64, int64Val: lv.int64Val - rv.int64Val) elif lv.kind == vkFloat64 and rv.kind == vkFloat64: @@ -909,8 +948,8 @@ proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionCont else: return Value(kind: vkNull) of irMul: - let lv = evalExprValue(expr.binLeft, row, ctx) - let rv = evalExprValue(expr.binRight, row, ctx) + let lv = evalExpr(expr.binLeft, row, ctx) + let rv = evalExpr(expr.binRight, row, ctx) if lv.kind == vkInt64 and rv.kind == vkInt64: return Value(kind: vkInt64, int64Val: lv.int64Val * rv.int64Val) elif lv.kind == vkFloat64 and rv.kind == vkFloat64: @@ -922,8 +961,8 @@ proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionCont else: return Value(kind: vkNull) of irDiv: - let lv = evalExprValue(expr.binLeft, row, ctx) - let rv = evalExprValue(expr.binRight, row, ctx) + let lv = evalExpr(expr.binLeft, row, ctx) + let rv = evalExpr(expr.binRight, row, ctx) var rvf = 0.0 if rv.kind == vkFloat64: rvf = rv.float64Val elif rv.kind == vkInt64: rvf = float(rv.int64Val) @@ -935,16 +974,16 @@ proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionCont else: return Value(kind: vkNull) return Value(kind: vkFloat64, float64Val: lvf / rvf) of irMod: - let lv = evalExprValue(expr.binLeft, row, ctx) - let rv = evalExprValue(expr.binRight, row, ctx) + let lv = evalExpr(expr.binLeft, row, ctx) + let rv = evalExpr(expr.binRight, row, ctx) if lv.kind == vkInt64 and rv.kind == vkInt64: if rv.int64Val == 0: return Value(kind: vkNull) return Value(kind: vkInt64, int64Val: lv.int64Val mod rv.int64Val) else: return Value(kind: vkNull) of irPow: - let lv = evalExprValue(expr.binLeft, row, ctx) - let rv = evalExprValue(expr.binRight, row, ctx) + let lv = evalExpr(expr.binLeft, row, ctx) + let rv = evalExpr(expr.binRight, row, ctx) var lvf = 0.0 if lv.kind == vkFloat64: lvf = lv.float64Val elif lv.kind == vkInt64: lvf = float(lv.int64Val) @@ -955,24 +994,43 @@ proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionCont else: return Value(kind: vkNull) return Value(kind: vkFloat64, float64Val: pow(lvf, rvf)) else: - let s = evalExpr(expr, row, ctx) + let s = evalExprLegacy(expr, row, ctx) return Value(kind: vkString, strVal: s) of irekUnary: case expr.unOp of irNeg: - let v = evalExprValue(expr.unExpr, row, ctx) + let v = evalExpr(expr.unExpr, row, ctx) case v.kind of vkInt64: return Value(kind: vkInt64, int64Val: -v.int64Val) of vkFloat64: return Value(kind: vkFloat64, float64Val: -v.float64Val) else: return Value(kind: vkNull) else: - let s = evalExpr(expr, row, ctx) + let s = evalExprLegacy(expr, row, ctx) return Value(kind: vkString, strVal: s) else: - let s = evalExpr(expr, row, ctx) + let s = evalExprLegacy(expr, row, ctx) return Value(kind: vkString, strVal: s) -proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string = +proc rowToStringTable(row: Row): Table[string, string] = + result = initTable[string, string]() + for k, v in row: + if v.kind == vkNull: + result[k] = "\\N" + else: + result[k] = valueToString(v) + +proc stringTableToValueRow(row: Table[string, string]): Row = + result = initTable[string, Value]() + for k, v in row: + result[k] = Value(kind: vkString, strVal: v) + +proc evalExprLegacy*(expr: IRExpr, row: Row, ctx: ExecutionContext = nil): string = + return evalExprOld(expr, rowToStringTable(row), ctx) + +proc evalExprLegacy*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string = + return evalExprOld(expr, row) + +proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string = if expr == nil: return "" case expr.kind of irekLiteral: @@ -1002,7 +1060,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = of irekStar: return "*" of irekJsonPath: - let srcVal = evalExpr(expr.jpExpr, row, ctx) + let srcVal = evalExprOld(expr.jpExpr, row, ctx) if srcVal.len == 0 or isNull(srcVal): return "\\N" try: let node = parseJson(srcVal) @@ -1025,8 +1083,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = except: return "" of irekBinary: - let left = evalExpr(expr.binLeft, row, ctx) - let right = evalExpr(expr.binRight, row, ctx) + let left = evalExprOld(expr.binLeft, row, ctx) + let right = evalExprOld(expr.binRight, row, ctx) case expr.binOp of irEq: if left == right: return "true" @@ -1064,7 +1122,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if left == "true" or right == "true": return "true" return "false" of irAdd, irSub, irMul, irDiv, irMod, irPow: - let v = evalExprValue(expr, row, ctx) + let v = evalExpr(expr, stringTableToValueRow(row), ctx) case v.kind of vkInt64: return $v.int64Val of vkFloat64: @@ -1112,7 +1170,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = var found = false for k, v in row: if k.startsWith("$"): continue - firstVal = v + firstVal = valueToString(v) found = true break if found and firstVal == left: return "true" @@ -1132,7 +1190,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = var found = false for k, v in row: if k.startsWith("$"): continue - firstVal = v + firstVal = valueToString(v) found = true break if found and firstVal == left: return "false" @@ -1260,16 +1318,16 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = of irekUnary: case expr.unOp of irNot: - let v = evalExpr(expr.unExpr, row, ctx) + let v = evalExprOld(expr.unExpr, row, ctx) return if v == "true": "false" else: "true" of irIsNull: - let v = evalExpr(expr.unExpr, row, ctx) + let v = evalExprOld(expr.unExpr, row, ctx) return if isNull(v): "true" else: "false" of irIsNotNull: - let v = evalExpr(expr.unExpr, row, ctx) + let v = evalExprOld(expr.unExpr, row, ctx) return if not isNull(v): "true" else: "false" of irNeg: - let v = evalExprValue(expr.unExpr, row, ctx) + let v = evalExpr(expr.unExpr, stringTableToValueRow(row), ctx) case v.kind of vkInt64: return $(-v.int64Val) of vkFloat64: @@ -1284,8 +1342,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = of "cosine_distance", "euclidean_distance", "inner_product", "l2_distance", "l1_distance": if expr.irFuncArgs.len < 2: return "0" - let left = evalExpr(expr.irFuncArgs[0], row, ctx) - let right = evalExpr(expr.irFuncArgs[1], row, ctx) + let left = evalExprOld(expr.irFuncArgs[0], row, ctx) + let right = evalExprOld(expr.irFuncArgs[1], row, ctx) let vecA = parseVectorString(left) let vecB = parseVectorString(right) if vecA.len == 0 or vecB.len == 0: @@ -1301,13 +1359,13 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = of "vector_dims", "vector_dimension": if expr.irFuncArgs.len < 1: return "0" - let arg = evalExpr(expr.irFuncArgs[0], row, ctx) + let arg = evalExprOld(expr.irFuncArgs[0], row, ctx) return $parseVectorString(arg).len of "json_has_key": if expr.irFuncArgs.len < 2: return "false" - let jsonStr = evalExpr(expr.irFuncArgs[0], row, ctx) - let key = evalExpr(expr.irFuncArgs[1], row, ctx) + let jsonStr = evalExprOld(expr.irFuncArgs[0], row, ctx) + let key = evalExprOld(expr.irFuncArgs[1], row, ctx) try: let node = parseJson(jsonStr) if node.kind == JObject: @@ -1324,7 +1382,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = of "current_setting": if expr.irFuncArgs.len < 1: return "" - let key = evalExpr(expr.irFuncArgs[0], row, ctx) + let key = evalExprOld(expr.irFuncArgs[0], row, ctx) if ctx != nil and key in ctx.sessionVars: return ctx.sessionVars[key] return "" @@ -1336,12 +1394,12 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return "" of "hybrid_search": if expr.irFuncArgs.len < 6: return "[]" - let table = evalExpr(expr.irFuncArgs[0], row, ctx) - let vecCol = evalExpr(expr.irFuncArgs[1], row, ctx) - let textCol = evalExpr(expr.irFuncArgs[2], row, ctx) - let queryText = evalExpr(expr.irFuncArgs[3], row, ctx) - let queryVec = evalExpr(expr.irFuncArgs[4], row, ctx) - let k = try: parseInt(evalExpr(expr.irFuncArgs[5], row, ctx)) except ValueError: 10 + let table = evalExprOld(expr.irFuncArgs[0], row, ctx) + let vecCol = evalExprOld(expr.irFuncArgs[1], row, ctx) + let textCol = evalExprOld(expr.irFuncArgs[2], row, ctx) + let queryText = evalExprOld(expr.irFuncArgs[3], row, ctx) + let queryVec = evalExprOld(expr.irFuncArgs[4], row, ctx) + let k = try: parseInt(evalExprOld(expr.irFuncArgs[5], row, ctx)) except ValueError: 10 let results = doHybridSearch(ctx, table, vecCol, textCol, queryText, queryVec, k) var jsonArr = newJArray() for (id, score) in results: @@ -1352,12 +1410,12 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return $jsonArr of "hybrid_search_ids": if expr.irFuncArgs.len < 6: return "" - let table = evalExpr(expr.irFuncArgs[0], row, ctx) - let vecCol = evalExpr(expr.irFuncArgs[1], row, ctx) - let textCol = evalExpr(expr.irFuncArgs[2], row, ctx) - let queryText = evalExpr(expr.irFuncArgs[3], row, ctx) - let queryVec = evalExpr(expr.irFuncArgs[4], row, ctx) - let k = try: parseInt(evalExpr(expr.irFuncArgs[5], row, ctx)) except ValueError: 10 + let table = evalExprOld(expr.irFuncArgs[0], row, ctx) + let vecCol = evalExprOld(expr.irFuncArgs[1], row, ctx) + let textCol = evalExprOld(expr.irFuncArgs[2], row, ctx) + let queryText = evalExprOld(expr.irFuncArgs[3], row, ctx) + let queryVec = evalExprOld(expr.irFuncArgs[4], row, ctx) + let k = try: parseInt(evalExprOld(expr.irFuncArgs[5], row, ctx)) except ValueError: 10 let results = doHybridSearch(ctx, table, vecCol, textCol, queryText, queryVec, k) var ids: seq[string] = @[] for (id, score) in results: @@ -1365,14 +1423,14 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return ids.join(",") of "hybrid_search_filtered": if expr.irFuncArgs.len < 8: return "[]" - let table = evalExpr(expr.irFuncArgs[0], row, ctx) - let vecCol = evalExpr(expr.irFuncArgs[1], row, ctx) - let textCol = evalExpr(expr.irFuncArgs[2], row, ctx) - let queryText = evalExpr(expr.irFuncArgs[3], row, ctx) - let queryVec = evalExpr(expr.irFuncArgs[4], row, ctx) - let k = try: parseInt(evalExpr(expr.irFuncArgs[5], row, ctx)) except ValueError: 10 - let filterCol = evalExpr(expr.irFuncArgs[6], row, ctx) - let filterVal = evalExpr(expr.irFuncArgs[7], row, ctx) + let table = evalExprOld(expr.irFuncArgs[0], row, ctx) + let vecCol = evalExprOld(expr.irFuncArgs[1], row, ctx) + let textCol = evalExprOld(expr.irFuncArgs[2], row, ctx) + let queryText = evalExprOld(expr.irFuncArgs[3], row, ctx) + let queryVec = evalExprOld(expr.irFuncArgs[4], row, ctx) + let k = try: parseInt(evalExprOld(expr.irFuncArgs[5], row, ctx)) except ValueError: 10 + let filterCol = evalExprOld(expr.irFuncArgs[6], row, ctx) + let filterVal = evalExprOld(expr.irFuncArgs[7], row, ctx) let results = doHybridSearchFiltered(ctx, table, vecCol, textCol, queryText, queryVec, k, filterCol, filterVal) var parts: seq[string] = @[] for (id, score) in results: @@ -1380,8 +1438,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return "[" & parts.join(",") & "]" of "rerank": if expr.irFuncArgs.len < 2: return "[]" - let queryText = evalExpr(expr.irFuncArgs[0], row, ctx) - let resultsJson = evalExpr(expr.irFuncArgs[1], row, ctx) + let queryText = evalExprOld(expr.irFuncArgs[0], row, ctx) + let resultsJson = evalExprOld(expr.irFuncArgs[1], row, ctx) # Simple rerank: boost results that contain query terms try: let arr = parseJson(resultsJson) @@ -1408,12 +1466,12 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return resultsJson of "chunk": if expr.irFuncArgs.len < 1: return "[]" - let text = evalExpr(expr.irFuncArgs[0], row, ctx) + let text = evalExprOld(expr.irFuncArgs[0], row, ctx) let maxSize = if expr.irFuncArgs.len >= 2: - try: parseInt(evalExpr(expr.irFuncArgs[1], row, ctx)) except ValueError: 1024 + try: parseInt(evalExprOld(expr.irFuncArgs[1], row, ctx)) except ValueError: 1024 else: 1024 let overlap = if expr.irFuncArgs.len >= 3: - try: parseInt(evalExpr(expr.irFuncArgs[2], row, ctx)) except ValueError: 128 + try: parseInt(evalExprOld(expr.irFuncArgs[2], row, ctx)) except ValueError: 128 else: 128 let cfg = chunkmod.ChunkConfig(maxChunkSize: maxSize, chunkOverlap: overlap, strategy: chunkmod.csRecursive, minChunkSize: 64) @@ -1424,7 +1482,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return $(jsonChunks) of "embed_text": if expr.irFuncArgs.len < 1: return "[]" - let text = evalExpr(expr.irFuncArgs[0], row, ctx) + let text = evalExprOld(expr.irFuncArgs[0], row, ctx) if ctx.embedder == nil or not ctx.embedder.config.enabled: return "[]" let vec = embedmod.embed(ctx.embedder, text) @@ -1432,8 +1490,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return embedmod.vectorToJson(vec) of "nl_to_sql": if expr.irFuncArgs.len < 1: return "" - let question = evalExpr(expr.irFuncArgs[0], row, ctx) - let table = if expr.irFuncArgs.len >= 2: evalExpr(expr.irFuncArgs[1], row, ctx) else: "" + let question = evalExprOld(expr.irFuncArgs[0], row, ctx) + let table = if expr.irFuncArgs.len >= 2: evalExprOld(expr.irFuncArgs[1], row, ctx) else: "" if ctx.llmClient == nil or not ctx.llmClient.config.enabled: return "" @@ -1503,7 +1561,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return sql of "schema_prompt": if expr.irFuncArgs.len < 1: return "" - let table = evalExpr(expr.irFuncArgs[0], row, ctx) + let table = evalExprOld(expr.irFuncArgs[0], row, ctx) if table notin ctx.tables: return "Table '" & table & "' not found" @@ -1530,7 +1588,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = ddl.add("-- ") var parts: seq[string] = @[] for col in tbl.columns: - parts.add(col.name & "=" & rows[i].getOrDefault(col.name, "")) + parts.add(col.name & "=" & valueToString(rows[i].getOrDefault(col.name, Value(kind: vkNull)))) ddl.add(parts.join(", ")) ddl.add("\n") else: @@ -1561,8 +1619,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return ddl of "similarity_nodes": if expr.irFuncArgs.len < 1: return "[]" - let graphName = evalExpr(expr.irFuncArgs[0], row, ctx) - let metric = if expr.irFuncArgs.len >= 2: evalExpr(expr.irFuncArgs[1], row, ctx).toLower() else: "jaccard" + let graphName = evalExprOld(expr.irFuncArgs[0], row, ctx) + let metric = if expr.irFuncArgs.len >= 2: evalExprOld(expr.irFuncArgs[1], row, ctx).toLower() else: "jaccard" if graphName notin ctx.graphs: return "[]" let g = ctx.graphs[graphName] @@ -1574,9 +1632,9 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return $(arr) of "node2vec_embed": if expr.irFuncArgs.len < 1: return "[]" - let graphName = evalExpr(expr.irFuncArgs[0], row, ctx) + let graphName = evalExprOld(expr.irFuncArgs[0], row, ctx) let dims = if expr.irFuncArgs.len >= 2: - try: parseInt(evalExpr(expr.irFuncArgs[1], row, ctx)) except ValueError: 64 + try: parseInt(evalExprOld(expr.irFuncArgs[1], row, ctx)) except ValueError: 64 else: 64 if graphName notin ctx.graphs: return "[]" @@ -1593,7 +1651,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return $(obj) of "cypher": if expr.irFuncArgs.len < 1: return "[]" - let cypherQuery = evalExpr(expr.irFuncArgs[0], row, ctx) + let cypherQuery = evalExprOld(expr.irFuncArgs[0], row, ctx) let sql = cyphermod.cypherToSql(cypherQuery) if sql.len == 0: return "[]" let tokens = qlex.tokenize(sql) @@ -1611,7 +1669,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return $(jsonRows) of "datetime": if expr.irFuncArgs.len > 0: - let arg = evalExpr(expr.irFuncArgs[0], row, ctx).toLower() + let arg = evalExprOld(expr.irFuncArgs[0], row, ctx).toLower() if arg == "now": return $now().format("yyyy-MM-dd HH:mm:ss") return arg @@ -1635,7 +1693,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if expr.irFuncArgs.len < 1: return "0" if ctx == nil: return "0" - let seqName = evalExpr(expr.irFuncArgs[0], row, ctx) + let seqName = evalExprOld(expr.irFuncArgs[0], row, ctx) var val: int64 = 0 acquire(ctx.sharedLock.lock) try: @@ -1650,7 +1708,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if expr.irFuncArgs.len < 1: return "0" if ctx == nil: return "0" - let seqName = evalExpr(expr.irFuncArgs[0], row, ctx) + let seqName = evalExprOld(expr.irFuncArgs[0], row, ctx) acquire(ctx.sharedLock.lock) try: if seqName in ctx.sequences: @@ -1662,7 +1720,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = # Snowflake ID: timestamp_ms(41 bits) | node_id(10 bits) | sequence(12 bits) var nodeId: int64 = 0 if expr.irFuncArgs.len > 0: - try: nodeId = parseInt(evalExpr(expr.irFuncArgs[0], row, ctx)) + try: nodeId = parseInt(evalExprOld(expr.irFuncArgs[0], row, ctx)) except ValueError: nodeId = 0 nodeId = nodeId and 0x3FF # 10 bits let ts = int64(epochTime() * 1000) and 0x1FFFFFFFFFF # 41 bits @@ -1671,8 +1729,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return $snowflakeId of "strftime": if expr.irFuncArgs.len >= 2: - let fmt = evalExpr(expr.irFuncArgs[0], row, ctx) - let val = evalExpr(expr.irFuncArgs[1], row, ctx) + let fmt = evalExprOld(expr.irFuncArgs[0], row, ctx) + let val = evalExprOld(expr.irFuncArgs[1], row, ctx) if fmt == "%s": try: let dt = parse(val, "yyyy-MM-dd HH:mm:ss") @@ -1689,10 +1747,10 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = else: # Unknown function: try to evaluate args and return first arg as fallback if expr.irFuncArgs.len > 0: - return evalExpr(expr.irFuncArgs[0], row, ctx) + return evalExprOld(expr.irFuncArgs[0], row, ctx) return "" of irekCast: - let val = evalExpr(expr.irCastExpr, row, ctx) + let val = evalExprOld(expr.irCastExpr, row, ctx) let castType = expr.irCastType.name.toLower() if castType.startsWith("vector"): let vec = parseVectorString(val) @@ -1723,7 +1781,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if subRows.len > 0: for k, v in subRows[0]: if not k.startsWith("$"): - return v + return valueToString(v) return "" else: return "" @@ -1759,7 +1817,7 @@ proc passesPolicy(ctx: ExecutionContext, tableName, command: string, row: Row): continue if pol.usingExpr != nil: let expr = lowerExpr(pol.usingExpr) - if evalExpr(expr, row, ctx) != "true": + if valueToString(evalExpr(expr, row, ctx)) != "true": return false return true @@ -1774,7 +1832,7 @@ proc checkInsertPolicy(ctx: ExecutionContext, tableName: string, row: Row): bool continue if pol.withCheckExpr != nil: let expr = lowerExpr(pol.withCheckExpr) - if evalExpr(expr, row, ctx) != "true": + if valueToString(evalExpr(expr, row, ctx)) != "true": return false return true @@ -1791,7 +1849,7 @@ proc execScan(ctx: ExecutionContext, table: string): seq[Row] = for (key, value) in ctx.db.scanAll(): if not key.startsWith(prefix): continue let rest = key[prefix.len..^1] - var row: Table[string, string] + var row: Row row["$key"] = rest let valStr = cast[string](value) row["$value"] = valStr @@ -1827,7 +1885,7 @@ proc execPointRead(ctx: ExecutionContext, table: string, key: string): seq[Row] let fullKey = table & "." & key let (found, val) = ctx.db.get(fullKey) if found: - var row: Table[string, string] + var row: Row row["$key"] = key let valStr = cast[string](val) row["$value"] = valStr @@ -1871,7 +1929,7 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu let fullKey = table & "." & key # Build row for RLS WITH CHECK - var row = initTable[string, string]() + var row = initTable[string, Value]() for i, f in fields: if i < rowVals.len: row[f] = rowVals[i] @@ -1920,7 +1978,7 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu meta["key"] = fullKey for col, val in row: if col.len > 0 and col != "$key" and col != "$value": - meta[col] = val + meta[col] = valueToString(val) vengine.insert(vecIdx, docId, vec, meta) # Auto-embed: if table has VECTOR column with null value but TEXT column @@ -1966,7 +2024,7 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu meta["key"] = fullKey for col, val in row: if col.len > 0 and col != "$key" and col != "$value": - meta[col] = val + meta[col] = valueToString(val) meta[vecCol] = vecStr2 vengine.insert(ctx.vectorIndexes[vecKey], docId, vec, meta) @@ -2023,7 +2081,7 @@ proc execDelete*(ctx: ExecutionContext, table: string, key: string, let (found, existingVal) = ctx.db.get(fullKey) if found: # RLS USING check on existing row - var oldRow = parseRowData(cast[string](existingVal)) + var oldRow = parseRowDataToValueRow(cast[string](existingVal)) let eqPos = key.find('=') if eqPos >= 0: oldRow[key[0..= 0: oldRow[key[0.. 0 and not isNull(newText): - ftsIdx.addDocument(docId, newText) + let newText = if colName in parsed: parsed[colName] else: Value(kind: vkNull) + if newText.kind == vkString and newText.strVal.len > 0: + ftsIdx.addDocument(docId, newText.strVal) # Update Vector indexes: add new vector (no remove support in current HNSW) for vecKey, vecIdx in ctx.vectorIndexes: if vecKey.startsWith(table & "."): let colName = vecKey[table.len + 1..^1] - let vecStr = if colName in parsed: parsed[colName] else: "\\N" - if not isNull(vecStr): - let vec = parseVectorString(vecStr) + let vecStr = if colName in parsed: parsed[colName] else: Value(kind: vkNull) + if vecStr.kind == vkString and vecStr.strVal.len > 0: + let vec = parseVectorString(vecStr.strVal) if vec.len > 0: var docId: uint64 = 0 for ch in fullKey: @@ -2143,7 +2201,7 @@ proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Tab proc findReferencingRows(ctx: ExecutionContext, childTable: string, fkCol: string, fkValue: string): seq[Row] = result = @[] for row in execScan(ctx, childTable): - if fkCol in row and row[fkCol] == fkValue: + if fkCol in row and valueToString(row[fkCol]) == fkValue: result.add(row) proc enforceFkOnDelete(ctx: ExecutionContext, parentTable: string, parentCol: string, parentVal: string): (bool, string) = @@ -2158,14 +2216,14 @@ proc enforceFkOnDelete(ctx: ExecutionContext, parentTable: string, parentCol: st for refRow in refs: if "$key" in refRow: var dummy: seq[(string, seq[byte])] = @[] - discard execDelete(ctx, childTblName, refRow["$key"], dummy) + discard execDelete(ctx, childTblName, valueToString(refRow["$key"]), dummy) of "SET NULL": for refRow in refs: if "$key" in refRow: var sets = initTable[string, string]() sets[col.name] = "\\N" var dummy: seq[(string, seq[byte])] = @[] - discard execUpdateRow(ctx, childTblName, refRow["$key"], sets, dummy) + discard execUpdateRow(ctx, childTblName, valueToString(refRow["$key"]), sets, dummy) of "RESTRICT", "NO ACTION": return (false, "FOREIGN KEY violation: row is referenced by " & childTblName & "." & col.name) return (true, "") @@ -2184,14 +2242,14 @@ proc enforceFkOnUpdate(ctx: ExecutionContext, parentTable: string, parentCol: st var sets = initTable[string, string]() sets[col.name] = newVal var dummy: seq[(string, seq[byte])] = @[] - discard execUpdateRow(ctx, childTblName, refRow["$key"], sets, dummy) + discard execUpdateRow(ctx, childTblName, valueToString(refRow["$key"]), sets, dummy) of "SET NULL": for refRow in refs: if "$key" in refRow: var sets = initTable[string, string]() sets[col.name] = "\\N" var dummy: seq[(string, seq[byte])] = @[] - discard execUpdateRow(ctx, childTblName, refRow["$key"], sets, dummy) + discard execUpdateRow(ctx, childTblName, valueToString(refRow["$key"]), sets, dummy) of "RESTRICT", "NO ACTION": return (false, "FOREIGN KEY violation: row is referenced by " & childTblName & "." & col.name) return (true, "") @@ -2270,7 +2328,7 @@ proc validateType*(colType: string, value: string): (bool, string) = proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult proc executeMigrationSql(ctx: ExecutionContext, sql: string): ExecResult -proc fireTriggers*(ctx: ExecutionContext, tableName: string, timing: string, event: string, row: Table[string, string]) = +proc fireTriggers*(ctx: ExecutionContext, tableName: string, timing: string, event: string, row: Row) = let tbl = ctx.getTableDef(tableName) for trig in tbl.triggers: if trig.timing == timing and trig.event == event: @@ -2332,15 +2390,15 @@ proc validateConstraints*(ctx: ExecutionContext, tableName: string, # CHECK constraints for check in tbl.checks: if check.checkNode != nil: - var row = initTable[string, string]() + var row = initTable[string, Value]() for i, f in fields: if i < rowVals.len: row[f] = rowVals[i] else: - row[f] = "\\N" + row[f] = Value(kind: vkNull) let checkExpr = lowerExpr(check.checkNode) let checkResult = evalExpr(checkExpr, row, ctx) - if checkResult != "true": + if valueToString(checkResult) != "true": return (false, "CHECK constraint '" & check.name & "' violated") return (true, "") @@ -2594,7 +2652,7 @@ proc lowerExpr*(node: Node): IRExpr = proc evalNodeToString(node: Node): string = ## Evaluate a simple AST node to a string value for INSERT/UPDATE. let ir = lowerExpr(node) - return evalExpr(ir, initTable[string, string](), nil) + return valueToString(evalExpr(ir, initTable[string, Value](), nil)) proc lowerSelect*(node: Node): IRPlan = result = IRPlan(kind: irpkScan) @@ -2775,7 +2833,7 @@ proc partitionKey(row: Row, partExprs: seq[IRExpr], ctx: ExecutionContext = nil) ## Compute a string partition key for a row result = "" for expr in partExprs: - result &= evalExpr(expr, row, ctx) & "|" + result &= valueToString(evalExpr(expr, row, ctx)) & "|" proc compareRowsByOrder(a, b: Row, orderExprs: seq[IRExpr], orderDirs: seq[bool], ctx: ExecutionContext = nil): int = ## Compare two rows by their ORDER BY expressions @@ -2784,12 +2842,12 @@ proc compareRowsByOrder(a, b: Row, orderExprs: seq[IRExpr], orderDirs: seq[bool] let vb = evalExpr(expr, b, ctx) var cmpRes = 0 try: - let fa = parseFloat(va) - let fb = parseFloat(vb) + let fa = parseFloat(valueToString(va)) + let fb = parseFloat(valueToString(vb)) if fa < fb: cmpRes = -1 elif fa > fb: cmpRes = 1 except: - cmpRes = cmp(va, vb) + cmpRes = cmp(valueToString(va), valueToString(vb)) if cmpRes != 0: return if orderDirs.len > i and orderDirs[i]: -cmpRes else: cmpRes return 0 @@ -2884,7 +2942,7 @@ proc computeWindowValues*(rows: seq[Row], expr: IRExpr, ctx: ExecutionContext = of "ntile": var n = 1 if expr.wfArgs.len > 0: - try: n = parseInt(evalExpr(expr.wfArgs[0], rows[sortedIdxs[0]], ctx)) except: n = 1 + try: n = parseInt(valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[0]], ctx))) except: n = 1 if n < 1: n = 1 let groupSize = sortedIdxs.len div n let remainder = sortedIdxs.len mod n @@ -2903,36 +2961,36 @@ proc computeWindowValues*(rows: seq[Row], expr: IRExpr, ctx: ExecutionContext = var offset = 1 var defaultVal = "" if expr.wfArgs.len > 1: - try: offset = parseInt(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]], ctx)) except: offset = 1 + try: offset = parseInt(valueToString(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]], ctx))) except: offset = 1 if expr.wfArgs.len > 2: - defaultVal = evalExpr(expr.wfArgs[2], rows[sortedIdxs[0]], ctx) + defaultVal = valueToString(evalExpr(expr.wfArgs[2], rows[sortedIdxs[0]], ctx)) for pos, rowIdx in sortedIdxs: let targetPos = pos + offset if targetPos < sortedIdxs.len: - result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[targetPos]], ctx) + result[rowIdx] = valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[targetPos]], ctx)) else: result[rowIdx] = defaultVal of "lag": var offset = 1 var defaultVal = "" if expr.wfArgs.len > 1: - try: offset = parseInt(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]], ctx)) except: offset = 1 + try: offset = parseInt(valueToString(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]], ctx))) except: offset = 1 if expr.wfArgs.len > 2: - defaultVal = evalExpr(expr.wfArgs[2], rows[sortedIdxs[0]], ctx) + defaultVal = valueToString(evalExpr(expr.wfArgs[2], rows[sortedIdxs[0]], ctx)) for pos, rowIdx in sortedIdxs: let targetPos = pos - offset if targetPos >= 0: - result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[targetPos]], ctx) + result[rowIdx] = valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[targetPos]], ctx)) else: result[rowIdx] = defaultVal of "first_value": for pos, rowIdx in sortedIdxs: let (fStart, _) = resolveFrameBounds(pos, sortedIdxs.len, frameStart, frameEnd) - result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[fStart]], ctx) + result[rowIdx] = valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[fStart]], ctx)) of "last_value": for pos, rowIdx in sortedIdxs: let (_, fEnd) = resolveFrameBounds(pos, sortedIdxs.len, frameStart, frameEnd) - result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[fEnd]], ctx) + result[rowIdx] = valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[fEnd]], ctx)) else: # Unknown window function — fill with null for rowIdx in sortedIdxs: @@ -2955,7 +3013,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = result = @[] for row in sourceRows: let evalResult = evalExpr(plan.filterCond, row, ctx) - if evalResult == "true": + if valueToString(evalResult) == "true": result.add(row) of irpkProject: @@ -2964,7 +3022,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = # Scalar SELECT (no FROM): create a dummy row so expressions can be evaluated if sourceRows.len == 0 and plan.projectSource != nil and plan.projectSource.kind == irpkScan and plan.projectSource.scanTable.len == 0: - sourceRows = @[initTable[string, string]()] + sourceRows = @[initTable[string, Value]()] # Check if this projection contains aggregates without GROUP BY var hasAggs = false @@ -2991,7 +3049,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = winValues[i] = computeWindowValues(sourceRows, expr) result = @[] for rowIdx, row in sourceRows: - var newRow: Table[string, string] + var newRow: Row for i, alias in plan.projectAliases: if i < plan.projectExprs.len: let expr = plan.projectExprs[i] @@ -3013,7 +3071,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if hasAggs: # Produce exactly one row with aggregate values - var newRow: Table[string, string] + var newRow: Row for i, alias in plan.projectAliases: if i < plan.projectExprs.len: let expr = plan.projectExprs[i] @@ -3028,7 +3086,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if expr.aggFilter != nil: filteredRows = @[] for row in sourceRows: - if evalExpr(expr.aggFilter, row, ctx) == "true": + if valueToString(evalExpr(expr.aggFilter, row, ctx)) == "true": filteredRows.add(row) case expr.aggOp of irCount: @@ -3038,50 +3096,50 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var count = 0 for row in filteredRows: let v = evalExpr(expr.aggArgs[0], row, ctx) - if v.len > 0: count += 1 + if valueToString(v).len > 0: count += 1 newRow[alias] = $count of irSum: var sum = 0.0 for row in filteredRows: let v = evalExpr(expr.aggArgs[0], row, ctx) - try: sum += parseFloat(v) except: discard + try: sum += parseFloat(valueToString(v)) except: discard newRow[alias] = $sum of irAvg: var sum = 0.0 var count = 0 for row in filteredRows: let v = evalExpr(expr.aggArgs[0], row, ctx) - try: sum += parseFloat(v); count += 1 except: discard + try: sum += parseFloat(valueToString(v)); count += 1 except: discard newRow[alias] = if count > 0: $(sum / float(count)) else: "0" of irMin: var minVal = "" for row in filteredRows: let v = evalExpr(expr.aggArgs[0], row, ctx) - if v == "\\N": continue - if minVal == "" or cmpMin(v, minVal): minVal = v + if v.kind == vkNull: continue + if minVal == "" or cmpMin(valueToString(v), minVal): minVal = valueToString(v) newRow[alias] = minVal of irMax: var maxVal = "" for row in filteredRows: let v = evalExpr(expr.aggArgs[0], row, ctx) - if v == "\\N": continue - if maxVal == "" or cmpMax(v, maxVal): maxVal = v + if v.kind == vkNull: continue + if maxVal == "" or cmpMax(valueToString(v), maxVal): maxVal = valueToString(v) newRow[alias] = maxVal of irArrayAgg: var arr: seq[string] for row in filteredRows: if expr.aggArgs.len > 0: - arr.add(evalExpr(expr.aggArgs[0], row, ctx)) + arr.add(valueToString(evalExpr(expr.aggArgs[0], row, ctx))) newRow[alias] = "[" & arr.join(", ") & "]" of irStringAgg: var parts: seq[string] - let delim = if expr.aggArgs.len > 1: evalExpr(expr.aggArgs[1], initTable[string, string](), ctx) else: "," + let delim = if expr.aggArgs.len > 1: evalExpr(expr.aggArgs[1], initTable[string, Value](), ctx) else: Value(kind: vkString, strVal: ",") for row in filteredRows: if expr.aggArgs.len > 0: - parts.add(evalExpr(expr.aggArgs[0], row, ctx)) - newRow[alias] = parts.join(delim) + parts.add(valueToString(evalExpr(expr.aggArgs[0], row, ctx))) + newRow[alias] = parts.join(valueToString(delim)) else: - let val = evalExpr(expr, if sourceRows.len > 0: sourceRows[0] else: initTable[string, string](), ctx) + let val = evalExpr(expr, if sourceRows.len > 0: sourceRows[0] else: initTable[string, Value](), ctx) if alias.len > 0: newRow[alias] = val else: newRow["col" & $i] = val result = @[newRow] @@ -3089,7 +3147,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = result = @[] for row in sourceRows: - var newRow: Table[string, string] + var newRow: Row for i, alias in plan.projectAliases: if i < plan.projectExprs.len: let expr = plan.projectExprs[i] @@ -3130,12 +3188,12 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let vb = evalExpr(sortExpr, b, ctx) var cmpRes = 0 try: - let fa = parseFloat(va) - let fb = parseFloat(vb) + let fa = parseFloat(valueToString(va)) + let fb = parseFloat(valueToString(vb)) if fa < fb: cmpRes = -1 elif fa > fb: cmpRes = 1 except: - cmpRes = cmp(va, vb) + cmpRes = cmp(valueToString(va), valueToString(vb)) if not ascending: cmpRes = -cmpRes if cmpRes != 0: return cmpRes return 0 @@ -3186,12 +3244,12 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for row in sourceRows: var groupKey = "" for gk in gkeys: - groupKey &= evalExpr(gk, row, ctx) & "|" + groupKey &= valueToString(evalExpr(gk, row, ctx)) & "|" if groupKey notin groups: groups[groupKey] = @[] groups[groupKey].add(row) for gk, groupRows in groups: - var aggRow: Table[string, string] + var aggRow: Row # Populate GROUP BY key columns for gkExpr in gkeys: if gkExpr.kind == irekField and gkExpr.fieldPath.len > 0: @@ -3208,7 +3266,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if aggExpr.aggFilter != nil: filteredRows = @[] for row in groupRows: - if evalExpr(aggExpr.aggFilter, row, ctx) == "true": + if valueToString(evalExpr(aggExpr.aggFilter, row, ctx)) == "true": filteredRows.add(row) case aggExpr.aggOp of irCount: @@ -3218,51 +3276,51 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var count = 0 for row in filteredRows: let v = evalExpr(aggExpr.aggArgs[0], row, ctx) - if v.len > 0: count += 1 + if valueToString(v).len > 0: count += 1 aggRow[aggKey] = $count of irSum: var sum = 0.0 for row in filteredRows: let v = evalExpr(aggExpr.aggArgs[0], row, ctx) - try: sum += parseFloat(v) except: discard + try: sum += parseFloat(valueToString(v)) except: discard aggRow[aggKey] = $sum of irAvg: var sum = 0.0 var count = 0 for row in filteredRows: let v = evalExpr(aggExpr.aggArgs[0], row, ctx) - try: sum += parseFloat(v); count += 1 except: discard + try: sum += parseFloat(valueToString(v)); count += 1 except: discard aggRow[aggKey] = if count > 0: $(sum / float(count)) else: "0" of irMin: var minVal = "" for row in filteredRows: let v = evalExpr(aggExpr.aggArgs[0], row, ctx) - if v == "\\N": continue - if minVal == "" or cmpMin(v, minVal): minVal = v + if v.kind == vkNull: continue + if minVal == "" or cmpMin(valueToString(v), minVal): minVal = valueToString(v) aggRow[aggKey] = minVal of irMax: var maxVal = "" for row in filteredRows: let v = evalExpr(aggExpr.aggArgs[0], row, ctx) - if v == "\\N": continue - if maxVal == "" or cmpMax(v, maxVal): maxVal = v + if v.kind == vkNull: continue + if maxVal == "" or cmpMax(valueToString(v), maxVal): maxVal = valueToString(v) aggRow[aggKey] = maxVal of irArrayAgg: var arr: seq[string] for row in filteredRows: if aggExpr.aggArgs.len > 0: - arr.add(evalExpr(aggExpr.aggArgs[0], row, ctx)) + arr.add(valueToString(evalExpr(aggExpr.aggArgs[0], row, ctx))) aggRow[aggKey] = "[" & arr.join(", ") & "]" of irStringAgg: var parts: seq[string] - let delim = if aggExpr.aggArgs.len > 1: evalExpr(aggExpr.aggArgs[1], initTable[string, string](), ctx) else: "," + let delim = if aggExpr.aggArgs.len > 1: evalExpr(aggExpr.aggArgs[1], initTable[string, Value](), ctx) else: Value(kind: vkString, strVal: ",") for row in filteredRows: if aggExpr.aggArgs.len > 0: - parts.add(evalExpr(aggExpr.aggArgs[0], row, ctx)) - aggRow[aggKey] = parts.join(delim) + parts.add(valueToString(evalExpr(aggExpr.aggArgs[0], row, ctx))) + aggRow[aggKey] = parts.join(valueToString(delim)) # Apply HAVING filter if plan.groupHaving != nil: - if evalExpr(plan.groupHaving, aggRow, ctx) != "true": + if valueToString(evalExpr(plan.groupHaving, aggRow, ctx)) != "true": continue result.add(aggRow) return result @@ -3286,7 +3344,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = else: "" proc mergeRow(left, right: Row, leftAlias, rightAlias, leftTable, rightTable: string): Row = - result = initTable[string, string]() + result = initTable[string, Value]() for k, v in left: if not k.startsWith("$"): result[k] = v @@ -3351,9 +3409,9 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var mergedRows: seq[Row] for r in rawRightRows: let merged = mergeRow(l, r, leftAlias, rightAlias, leftTable, getScanTable(plan.joinRight)) - if rightFilter != nil and evalExpr(rightFilter, merged, ctx) != "true": + if rightFilter != nil and valueToString(evalExpr(rightFilter, merged, ctx)) != "true": continue - if plan.joinCond != nil and evalExpr(plan.joinCond, merged, ctx) != "true": + if plan.joinCond != nil and valueToString(evalExpr(plan.joinCond, merged, ctx)) != "true": continue mergedRows.add(merged) @@ -3365,8 +3423,8 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let bVal = evalExpr(sExpr, b, ctx) let asc = if i < rightSortDirs.len: rightSortDirs[i] else: true var cmp = 0 - let aNum = parseFloat(aVal) - let bNum = parseFloat(bVal) + let aNum = parseFloat(valueToString(aVal)) + let bNum = parseFloat(valueToString(bVal)) if aNum < bNum: cmp = -1 elif aNum > bNum: cmp = 1 if cmp != 0: @@ -3385,23 +3443,32 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = result.add(row) elif plan.joinKind == irjkLeft or plan.joinKind == irjkFull: var rightCols: seq[string] - for r in rawRightRows: + # Build rightCols from filtered rows if available, otherwise from raw rows + let sourceForCols = if mergedRows.len > 0: mergedRows else: rawRightRows + for r in sourceForCols: for k, _ in r: if not k.startsWith("$") and k notin rightCols: rightCols.add(k) - var padded = initTable[string, string]() + # If still no right rows, get column names from table definition + if rightCols.len == 0 and rightScanPlan != nil and rightScanPlan.kind == irpkScan: + let rightTable = rightScanPlan.scanTable + if rightTable.len > 0: + let tbl = ctx.getTableDef(rightTable) + for col in tbl.columns: + rightCols.add(col.name) + var padded = initTable[string, Value]() for k, v in l: if not k.startsWith("$"): padded[k] = v for col in rightCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if leftAlias.len > 0: for k, v in l: if not k.startsWith("$"): padded[leftAlias & "." & k] = v if rightAlias.len > 0: for col in rightCols: - padded[rightAlias & "." & col] = "\\N" + padded[rightAlias & "." & col] = Value(kind: vkNull) result.add(padded) return result @@ -3462,17 +3529,17 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var hashTable = initTable[string, seq[Row]]() for row in buildRows: - let key = if buildAlias.len > 0 and buildCol in row: row[buildCol] - elif buildCol in row: row[buildCol] + let key = if buildAlias.len > 0 and buildCol in row: valueToString(row[buildCol]) + elif buildCol in row: valueToString(row[buildCol]) else: "" - if key.len > 0 and key != "\\N": + if key.len > 0 : if key notin hashTable: hashTable[key] = @[] hashTable[key].add(row) var matchedProbe = initTable[int, bool]() for i, prow in probeRows: - let key = if probeAlias.len > 0 and probeCol in prow: prow[probeCol] - elif probeCol in prow: prow[probeCol] + let key = if probeAlias.len > 0 and probeCol in prow: valueToString(prow[probeCol]) + elif probeCol in prow: valueToString(prow[probeCol]) else: "" if key in hashTable: matchedProbe[i] = true @@ -3486,32 +3553,32 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if plan.joinKind == irjkLeft or plan.joinKind == irjkFull or plan.joinKind == irjkRight: for i, prow in probeRows: if not matchedProbe.getOrDefault(i, false): - var padded = initTable[string, string]() + var padded = initTable[string, Value]() for k, v in prow: if not k.startsWith("$"): padded[k] = v if buildIsLeft: # probe is right side for col in leftCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if probeAlias.len > 0: for k, v in prow: if not k.startsWith("$"): padded[probeAlias & "." & k] = v if leftAlias.len > 0: for col in leftCols: - padded[leftAlias & "." & col] = "\\N" + padded[leftAlias & "." & col] = Value(kind: vkNull) else: # probe is left side for col in rightCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if probeAlias.len > 0: for k, v in prow: if not k.startsWith("$"): padded[probeAlias & "." & k] = v if rightAlias.len > 0: for col in rightCols: - padded[rightAlias & "." & col] = "\\N" + padded[rightAlias & "." & col] = Value(kind: vkNull) result.add(padded) # RIGHT / FULL padding for non-matching build rows @@ -3519,40 +3586,40 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for i, brow in buildRows: var found = false for j, prow in probeRows: - let pkey = if probeAlias.len > 0 and probeCol in prow: prow[probeCol] - elif probeCol in prow: prow[probeCol] + let pkey = if probeAlias.len > 0 and probeCol in prow: valueToString(prow[probeCol]) + elif probeCol in prow: valueToString(prow[probeCol]) else: "" - let bkey = if buildAlias.len > 0 and buildCol in brow: brow[buildCol] - elif buildCol in brow: brow[buildCol] + let bkey = if buildAlias.len > 0 and buildCol in brow: valueToString(brow[buildCol]) + elif buildCol in brow: valueToString(brow[buildCol]) else: "" if pkey == bkey and bkey.len > 0: found = true break if not found: - var padded = initTable[string, string]() + var padded = initTable[string, Value]() for k, v in brow: if not k.startsWith("$"): padded[k] = v if buildIsLeft: for col in rightCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if leftAlias.len > 0: for k, v in brow: if not k.startsWith("$"): padded[leftAlias & "." & k] = v if rightAlias.len > 0: for col in rightCols: - padded[rightAlias & "." & col] = "\\N" + padded[rightAlias & "." & col] = Value(kind: vkNull) else: for col in leftCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if rightAlias.len > 0: for k, v in brow: if not k.startsWith("$"): padded[rightAlias & "." & k] = v if leftAlias.len > 0: for col in leftCols: - padded[leftAlias & "." & col] = "\\N" + padded[leftAlias & "." & col] = Value(kind: vkNull) result.add(padded) of irjsIndexNestedLoop: @@ -3588,13 +3655,13 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let btree = ctx.btrees[idxName] var matchedOuter = initTable[int, bool]() for i, orow in outerRows: - let key = if orow.hasKey(outerCol): orow[outerCol] else: "" - if key.len > 0 and key != "\\N": + let key = if orow.hasKey(outerCol): valueToString(orow[outerCol]) else: "" + if key.len > 0 : let entries = btree.get(key) if entries.len > 0: matchedOuter[i] = true for entry in entries: - let parsed = parseRowData(entry.rowValue) + let parsed = parseRowDataToValueRow(entry.rowValue) if outerIsLeft: result.add(mergeRow(orow, parsed, leftAlias, rightAlias, leftTable, rightTable)) else: @@ -3606,30 +3673,30 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if plan.joinKind == irjkLeft or plan.joinKind == irjkRight or plan.joinKind == irjkFull: for i, orow in outerRows: if not matchedOuter.getOrDefault(i, false): - var padded = initTable[string, string]() + var padded = initTable[string, Value]() for k, v in orow: if not k.startsWith("$"): padded[k] = v if outerIsLeft: for col in rightCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if leftAlias.len > 0: for k, v in orow: if not k.startsWith("$"): padded[leftAlias & "." & k] = v if rightAlias.len > 0: for col in rightCols: - padded[rightAlias & "." & col] = "\\N" + padded[rightAlias & "." & col] = Value(kind: vkNull) else: for col in leftCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if rightAlias.len > 0: for k, v in orow: if not k.startsWith("$"): padded[rightAlias & "." & k] = v if leftAlias.len > 0: for col in leftCols: - padded[leftAlias & "." & col] = "\\N" + padded[leftAlias & "." & col] = Value(kind: vkNull) result.add(padded) return result @@ -3638,23 +3705,23 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var matched = false for r in rightRows: let merged = mergeRow(l, r, leftAlias, rightAlias, leftTable, rightTable) - if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": + if plan.joinCond == nil or valueToString(evalExpr(plan.joinCond, merged, ctx)) == "true": result.add(merged) matched = true if not matched and (plan.joinKind == irjkLeft or plan.joinKind == irjkFull): - var padded = initTable[string, string]() + var padded = initTable[string, Value]() for k, v in l: if not k.startsWith("$"): padded[k] = v for col in rightCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if leftAlias.len > 0: for k, v in l: if not k.startsWith("$"): padded[leftAlias & "." & k] = v if rightAlias.len > 0: for col in rightCols: - padded[rightAlias & "." & col] = "\\N" + padded[rightAlias & "." & col] = Value(kind: vkNull) result.add(padded) if plan.joinKind == irjkRight or plan.joinKind == irjkFull: @@ -3662,23 +3729,23 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var found = false for l in leftRows: let merged = mergeRow(l, r, leftAlias, rightAlias, leftTable, rightTable) - if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": + if plan.joinCond == nil or valueToString(evalExpr(plan.joinCond, merged, ctx)) == "true": found = true break if not found: - var padded = initTable[string, string]() + var padded = initTable[string, Value]() for k, v in r: if not k.startsWith("$"): padded[k] = v for col in leftCols: - if col notin padded: padded[col] = "\\N" + if col notin padded: padded[col] = Value(kind: vkNull) if rightAlias.len > 0: for k, v in r: if not k.startsWith("$"): padded[rightAlias & "." & k] = v if leftAlias.len > 0: for col in leftCols: - padded[leftAlias & "." & col] = "\\N" + padded[leftAlias & "." & col] = Value(kind: vkNull) result.add(padded) return result @@ -3704,13 +3771,13 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for row in sourceRows: var groupKey = "" for col in groupCols: - groupKey &= (if col in row: row[col] else: "") & "|" + groupKey &= (if col in row: valueToString(row[col]) else: "") & "|" if groupKey notin groups: groups[groupKey] = @[] groups[groupKey].add(row) # For each group, create a pivoted row for gk, groupRows in groups: - var newRow: Table[string, string] + var newRow: Row for col in groupCols: if col in groupRows[0]: newRow[col] = groupRows[0][col] @@ -3718,7 +3785,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for pivotVal in plan.pivotInValues: var matchingRows: seq[Row] for row in groupRows: - if plan.pivotForCol in row and row[plan.pivotForCol] == pivotVal: + if plan.pivotForCol in row and valueToString(row[plan.pivotForCol]) == pivotVal: matchingRows.add(row) # Compute aggregate var aggResult = "" @@ -3731,34 +3798,34 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var count = 0 for row in matchingRows: let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) - if v.len > 0: count += 1 + if valueToString(v).len > 0: count += 1 aggResult = $count of irSum: var sum = 0.0 for row in matchingRows: let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) - try: sum += parseFloat(v) except: discard + try: sum += parseFloat(valueToString(v)) except: discard aggResult = $sum of irAvg: var sum = 0.0 var count = 0 for row in matchingRows: let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) - try: sum += parseFloat(v); count += 1 except: discard + try: sum += parseFloat(valueToString(v)); count += 1 except: discard aggResult = if count > 0: $(sum / float(count)) else: "0" of irMin: var minVal = "" for row in matchingRows: let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) - if v == "\\N": continue - if minVal == "" or cmpMin(v, minVal): minVal = v + if v.kind == vkNull: continue + if minVal == "" or cmpMin(valueToString(v), minVal): minVal = valueToString(v) aggResult = minVal of irMax: var maxVal = "" for row in matchingRows: let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) - if v == "\\N": continue - if maxVal == "" or cmpMax(v, maxVal): maxVal = v + if v.kind == vkNull: continue + if maxVal == "" or cmpMax(valueToString(v), maxVal): maxVal = valueToString(v) aggResult = maxVal else: discard # Clean pivot value (remove quotes) @@ -3779,12 +3846,12 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = # For each source row, create one row per IN column for row in sourceRows: for inCol in plan.unpivotInCols: - var newRow: Table[string, string] + var newRow: Row for col in identityCols: if col in row: newRow[col] = row[col] newRow[plan.unpivotForCol] = inCol - newRow[plan.unpivotValueCol] = (if inCol in row: row[inCol] else: "\\N") + newRow[plan.unpivotValueCol] = (if inCol in row: row[inCol] else: Value(kind: vkNull)) result.add(newRow) return result @@ -3811,7 +3878,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let maxDepth = if plan.graphMaxDepth >= 0: plan.graphMaxDepth else: -1 let traverseResult = gengine.bfs(g, startId, maxDepth) for nodeId in traverseResult: - var row = initTable[string, string]() + var row = initTable[string, Value]() let nid = uint64(nodeId) row["_node_id"] = $nid if nodeId in g.nodes: @@ -3831,7 +3898,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let maxDepth = if plan.graphMaxDepth >= 0: plan.graphMaxDepth else: -1 let traverseResult = gengine.dfs(g, startId, maxDepth) for nodeId in traverseResult: - var row = initTable[string, string]() + var row = initTable[string, Value]() let nid = uint64(nodeId) row["_node_id"] = $nid if nodeId in g.nodes: @@ -3854,7 +3921,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let vb = prResult.getOrDefault(b, 0.0) if va > vb: return -1 elif va < vb: return 1 else: return 0) for nodeId in sortedNodes: - var row = initTable[string, string]() + var row = initTable[string, Value]() let nid = uint64(nodeId) row["_node_id"] = $nid row["rank"] = $prResult.getOrDefault(nodeId, 0.0) @@ -3873,7 +3940,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let endId = gengine.NodeId(explicitEnd) let path = gengine.shortestPath(g, startId, endId) for nodeId in path: - var row = initTable[string, string]() + var row = initTable[string, Value]() let nid = uint64(nodeId) row["_node_id"] = $nid if nodeId in g.nodes: @@ -3891,7 +3958,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let startId = gengine.NodeId(explicitStart) let dists = gengine.dijkstra(g, startId) for nodeId, dist in dists: - var row = initTable[string, string]() + var row = initTable[string, Value]() row["_node_id"] = $(uint64(nodeId)) row["distance"] = $dist if nodeId in g.nodes: @@ -3903,7 +3970,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = of "community", "community_detect", "louvain": let louvainResult = gcomm.louvain(g) for nodeId, communityId in louvainResult.communities: - var row = initTable[string, string]() + var row = initTable[string, Value]() row["_node_id"] = $(uint64(nodeId)) row["community"] = $communityId if nodeId in g.nodes: @@ -3912,7 +3979,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = else: for nodeId in g.nodes.keys: - var row = initTable[string, string]() + var row = initTable[string, Value]() let nid = uint64(nodeId) row["_node_id"] = $nid row["_node_label"] = g.nodes[nodeId].label @@ -4156,11 +4223,11 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu # UNION: deduplicate against all already-accumulated rows var seen = initTable[string, bool]() for existing in allRows: - let key = if "$value" in existing: existing["$value"] else: $existing + let key = if "$value" in existing: valueToString(existing["$value"]) else: $existing if key.len > 0: seen[key] = true for row in rightRes.rows: - let key = if "$value" in row: row["$value"] else: $row + let key = if "$value" in row: valueToString(row["$value"]) else: $row if not seen.getOrDefault(key, false): if key.len > 0: seen[key] = true @@ -4210,7 +4277,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu let whereIr = lowerExpr(stmt.selWhere.whereExpr) var tmp: seq[Row] = @[] for row in filteredRows: - if evalExpr(whereIr, row, ctx) == "true": + if valueToString(evalExpr(whereIr, row, ctx)) == "true": tmp.add(row) filteredRows = tmp if stmt.selOrderBy.len > 0: @@ -4220,13 +4287,13 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu let va = evalExpr(sortExpr, a, ctx) let vb = evalExpr(sortExpr, b, ctx) try: - let fa = parseFloat(va) - let fb = parseFloat(vb) + let fa = parseFloat(valueToString(va)) + let fb = parseFloat(valueToString(vb)) if fa < fb: return -1 if fa > fb: return 1 return 0 except: - return cmp(va, vb) + return cmp(valueToString(va), valueToString(vb)) filteredRows.sort(sortCmp, if asc: Ascending else: Descending) if stmt.selLimit != nil: let limitVal = if stmt.selLimit.limitExpr.kind == nkIntLit: @@ -4269,7 +4336,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for entry in entries: let (found, val) = ctx.db.get(entry.lsmKey) if found: - rows.add(parseRowData(cast[string](val))) + rows.add(parseRowDataToValueRow(cast[string](val))) let tbl = ctx.getTableDef(stmt.selFrom.fromTable) var cols: seq[string] = @[] for c in tbl.columns: cols.add(c.name) @@ -4309,7 +4376,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for entry in entries: let (found, val) = ctx.db.get(entry.lsmKey) if found: - rows.add(parseRowData(cast[string](val))) + rows.add(parseRowDataToValueRow(cast[string](val))) let tbl = ctx.getTableDef(stmt.selFrom.fromTable) var cols: seq[string] = @[] for c in tbl.columns: cols.add(c.name) @@ -4335,7 +4402,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if isCovered and coveredCols.len > 0: var rows: seq[Row] = @[] for entry in entries: - var row = initTable[string, string]() + var row = initTable[string, Value]() row[colName] = w.binRight.strVal rows.add(row) return okResult(rows, coveredCols) @@ -4359,7 +4426,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for entry in entries: let (found, val) = ctx.db.get(entry.lsmKey) if found: - rows.add(parseRowData(cast[string](val))) + rows.add(parseRowDataToValueRow(cast[string](val))) let tbl = ctx.getTableDef(stmt.selFrom.fromTable) var cols: seq[string] = @[] for c in tbl.columns: cols.add(c.name) @@ -4394,7 +4461,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for entry in entries: let (found, val) = ctx.db.get(entry.lsmKey) if found: - rows.add(parseRowData(cast[string](val))) + rows.add(parseRowDataToValueRow(cast[string](val))) let tbl = ctx.getTableDef(stmt.selFrom.fromTable) var cols: seq[string] = @[] for c in tbl.columns: cols.add(c.name) @@ -4450,28 +4517,28 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu # UNION: deduplicate var seen: Table[string, bool] for row in leftRes.rows: - seen[row["$value"]] = true + seen[valueToString(row["$value"])] = true for row in rightRes.rows: - if not seen.getOrDefault(row["$value"], false): - seen[row["$value"]] = true + if not seen.getOrDefault(valueToString(row["$value"]), false): + seen[valueToString(row["$value"])] = true rows.add(row) of sdkIntersect: var leftSet: Table[string, bool] for row in leftRes.rows: - leftSet[row["$value"]] = true + leftSet[valueToString(row["$value"])] = true for row in rightRes.rows: - if leftSet.getOrDefault(row["$value"], false): + if leftSet.getOrDefault(valueToString(row["$value"]), false): rows.add(row) if not stmt.setOpAll: - leftSet.del(row["$value"]) # remove to prevent duplicates for INTERSECT (not ALL) + leftSet.del(valueToString(row["$value"])) # remove to prevent duplicates for INTERSECT (not ALL) of sdkExcept: var rightSet: Table[string, bool] for row in rightRes.rows: - rightSet[row["$value"]] = true + rightSet[valueToString(row["$value"])] = true for row in leftRes.rows: - if not rightSet.getOrDefault(row["$value"], false): + if not rightSet.getOrDefault(valueToString(row["$value"]), false): rows.add(row) return okResult(rows, cols) @@ -4550,7 +4617,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if not valid: return errResult(errMsg) # Fire BEFORE INSERT triggers - var row = initTable[string, string]() + var row = initTable[string, Value]() if mutableValues.len > 0: for i, f in mutableFields: if i < mutableValues[0].len: @@ -4581,11 +4648,11 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu else: returnCols.add("col" & $returnCols.len) for rowVals in mutableValues: - var rowMap = initTable[string, string]() + var rowMap = initTable[string, Value]() for i, f in mutableFields: if i < rowVals.len: rowMap[f] = rowVals[i] - var returnRow = initTable[string, string]() + var returnRow = initTable[string, Value]() for i, retExpr in stmt.insReturning: let ir = lowerExpr(retExpr) let val = evalExpr(ir, rowMap, ctx) @@ -4621,15 +4688,15 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu elif s.binRight.kind == nkFloatLit: $s.binRight.floatVal elif s.binRight.kind == nkBoolLit: $s.binRight.boolVal elif s.binRight.kind == nkNullLit: "\\N" - else: evalExpr(lowerExpr(s.binRight), row, ctx) + else: valueToString(evalExpr(lowerExpr(s.binRight), row, ctx)) sets[s.binLeft.identName] = val # Check WHERE if stmt.updWhere != nil and stmt.updWhere.whereExpr != nil: let whereExpr = lowerExpr(stmt.updWhere.whereExpr) - if evalExpr(whereExpr, row, ctx) != "true": continue + if valueToString(evalExpr(whereExpr, row, ctx)) != "true": continue # Get key from row if "$key" in row: - let old = row["$key"] + let old = valueToString(row["$key"]) # Build updated row for constraint validation var updFields: seq[string] = @[] var updValues: seq[string] = @[] @@ -4638,7 +4705,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if col.name in sets: updValues.add(sets[col.name]) elif col.name in row: - updValues.add(row[col.name]) + updValues.add(valueToString(row[col.name])) else: updValues.add("\\N") let (valid, errMsg) = validateConstraints(ctx, stmt.updTarget, updFields, @[updValues], skipPkCheck = true) @@ -4651,7 +4718,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu refCols.add(col.fkColumn) for refCol in refCols: if refCol in sets and refCol in row: - let (fkOk, fkErr) = enforceFkOnUpdate(ctx, stmt.updTarget, refCol, row[refCol], sets[refCol]) + let (fkOk, fkErr) = enforceFkOnUpdate(ctx, stmt.updTarget, refCol, valueToString(row[refCol]), sets[refCol]) if not fkOk: return errResult(fkErr) # FK ON UPDATE enforcement (child side — validate new FK values) @@ -4663,10 +4730,10 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu var oldRow = row var newRow = row for col, val in sets: - newRow[col] = val + newRow[col] = Value(kind: vkString, strVal: val) fireTriggers(ctx, stmt.updTarget, "before", "update", oldRow) - count += execUpdateRow(ctx, stmt.updTarget, row["$key"], sets, kvPairs) + count += execUpdateRow(ctx, stmt.updTarget, valueToString(row["$key"]), sets, kvPairs) # Fire AFTER UPDATE triggers fireTriggers(ctx, stmt.updTarget, "after", "update", newRow) @@ -4683,9 +4750,9 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for row in rows: if stmt.delWhere != nil and stmt.delWhere.whereExpr != nil: let whereExpr = lowerExpr(stmt.delWhere.whereExpr) - if evalExpr(whereExpr, row, ctx) != "true": continue + if valueToString(evalExpr(whereExpr, row, ctx)) != "true": continue if "$key" in row: - let old = row["$key"] + let old = valueToString(row["$key"]) # Fire BEFORE DELETE triggers fireTriggers(ctx, stmt.delTarget, "before", "delete", row) @@ -4697,11 +4764,11 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu refCols.add(col.fkColumn) for refCol in refCols: if refCol in row: - let (fkOk, fkErr) = enforceFkOnDelete(ctx, stmt.delTarget, refCol, row[refCol]) + let (fkOk, fkErr) = enforceFkOnDelete(ctx, stmt.delTarget, refCol, valueToString(row[refCol])) if not fkOk: return errResult(fkErr) - count += execDelete(ctx, stmt.delTarget, row["$key"], kvPairs) + count += execDelete(ctx, stmt.delTarget, valueToString(row["$key"]), kvPairs) # Fire AFTER DELETE triggers fireTriggers(ctx, stmt.delTarget, "after", "delete", row) @@ -4735,7 +4802,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for k, v in tgtRow: rowWithTarget[stmt.mergeTargetAlias & "." & k] = v let onExpr = lowerExpr(stmt.mergeOn) - if evalExpr(onExpr, rowWithTarget, ctx) == "true": + if valueToString(evalExpr(onExpr, rowWithTarget, ctx)) == "true": matched = true if stmt.mergeMatchedUpdate.len > 0 and "$key" in tgtRow: var updateSets = initTable[string, string]() @@ -4743,15 +4810,15 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if s.kind == nkBinOp and s.binOp == bkAssign: if s.binLeft.kind == nkIdent: let valExpr = lowerExpr(s.binRight) - updateSets[s.binLeft.identName] = evalExpr(valExpr, rowWithTarget, ctx) + updateSets[s.binLeft.identName] = valueToString(evalExpr(valExpr, rowWithTarget, ctx)) var newRow = tgtRow for col, val in updateSets: - newRow[col] = val + newRow[col] = Value(kind: vkString, strVal: val) fireTriggers(ctx, stmt.mergeTarget, "before", "update", tgtRow) - count += execUpdateRow(ctx, stmt.mergeTarget, tgtRow["$key"], updateSets, kvPairs) + count += execUpdateRow(ctx, stmt.mergeTarget, valueToString(tgtRow["$key"]), updateSets, kvPairs) fireTriggers(ctx, stmt.mergeTarget, "after", "update", newRow) if ctx.onChange != nil: - ctx.onChange(ChangeEvent(table: stmt.mergeTarget, kind: ckUpdate, key: tgtRow["$key"], data: "")) + ctx.onChange(ChangeEvent(table: stmt.mergeTarget, kind: ckUpdate, key: valueToString(tgtRow["$key"]), data: "")) break if not matched and stmt.mergeNotMatchedInsert.len > 0: @@ -4763,13 +4830,13 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if i < stmt.mergeNotMatchedValues.len: let v = stmt.mergeNotMatchedValues[i] let valExpr = lowerExpr(v) - values.add(evalExpr(valExpr, combinedRow, ctx)) + values.add(valueToString(evalExpr(valExpr, combinedRow, ctx))) else: values.add("\\N") if fields.len > 0: - var row = initTable[string, string]() + var row = initTable[string, Value]() for i, f in fields: - if i < values.len: row[f] = values[i] + if i < values.len: row[f] = Value(kind: vkString, strVal: values[i]) fireTriggers(ctx, stmt.mergeTarget, "before", "insert", row) var insKvPairs: seq[(string, seq[byte])] count += execInsert(ctx, stmt.mergeTarget, fields, @[values], insKvPairs) @@ -5094,7 +5161,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for name in listMigrations(ctx): let applied = isMigrationApplied(ctx, name) let rec = getMigrationRecord(ctx, name) - var row = initTable[string, string]() + var row = initTable[string, Value]() row["name"] = name row["status"] = if applied: "applied" else: "pending" row["applied_at"] = if rec.appliedAt > 0: $rec.appliedAt else: "" @@ -5207,13 +5274,13 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu var ftsIdx = fts.newInvertedIndex() let rows = execScan(ctx, stmt.ciTarget) for row in rows: - let lsmKey = if "$key" in row: row["$key"] else: "" + let lsmKey = if "$key" in row: valueToString(row["$key"]) else: "" let docKey = stmt.ciTarget & "." & lsmKey var docId: uint64 = 0 for ch in docKey: docId = docId * 31 + uint64(ord(ch)) for col in stmt.ciColumns: - let text = if col in row: row[col] else: "" + let text = if col in row: valueToString(row[col]) else: "" if text.len > 0: ftsIdx.addDocument(docId, text) ctx.ftsIndexes[colKey] = ftsIdx @@ -5226,7 +5293,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for row in rows: for col in stmt.ciColumns: if col in row: - let vec = parseVectorString(row[col]) + let vec = parseVectorString(valueToString(row[col])) if vec.len > 0: dimensions = vec.len break @@ -5237,15 +5304,15 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu for row in rows: for col in stmt.ciColumns: if col in row: - let vec = parseVectorString(row[col]) + let vec = parseVectorString(valueToString(row[col])) if vec.len > 0: var meta = initTable[string, string]() if "$key" in row: - meta["key"] = row["$key"] + meta["key"] = valueToString(row["$key"]) for col, val in row: if col.len > 0 and col != "$key" and col != "$value": - meta[col] = val - let fullKey = stmt.ciTarget & "." & row["$key"] + meta[col] = valueToString(val) + let fullKey = stmt.ciTarget & "." & valueToString(row["$key"]) var docId: uint64 = 0 for ch in fullKey: docId = docId * 31 + uint64(ord(ch)) @@ -5260,12 +5327,12 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu var colVals: seq[string] = @[] for col in stmt.ciColumns: if col in row: - colVals.add(row[col]) + colVals.add(valueToString(row[col])) else: colVals.add("\\N") let idxVal = colVals.join("|") if idxVal.len > 0 and not isNull(idxVal): - let lsmKey = if "$key" in row: stmt.ciTarget & "." & row["$key"] else: "" + let lsmKey = if "$key" in row: stmt.ciTarget & "." & valueToString(row["$key"]) else: "" ctx.btrees[colKey].insert(idxVal, IndexEntry(lsmKey: lsmKey, rowValue: "")) return okResult(msg="CREATE INDEX " & idxName & " on " & stmt.ciTarget) @@ -5428,7 +5495,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu return errResult("Multi-database support not enabled") var rows: seq[Row] = @[] for dbName in listDatabases(ctx.registry): - var row = initTable[string, string]() + var row = initTable[string, Value]() row["name"] = dbName rows.add(row) return okResult(rows, @["name"]) diff --git a/src/barabadb/query/lexer.nim b/src/barabadb/query/lexer.nim index 699921d..ffaf6e9 100644 --- a/src/barabadb/query/lexer.nim +++ b/src/barabadb/query/lexer.nim @@ -160,6 +160,10 @@ type tkBfs tkDfs tkPath + tkDatabase + tkDatabases + tkUse + tkShow tkAutoIncrement tkSequence @@ -382,6 +386,10 @@ const keywords*: Table[string, TokenKind] = { "bfs": tkBfs, "dfs": tkDfs, "path": tkPath, + "database": tkDatabase, + "databases": tkDatabases, + "use": tkUse, + "show": tkShow, "over": tkOver, "partition": tkPartition, "row": tkRow, diff --git a/src/barabadb/query/parser.nim b/src/barabadb/query/parser.nim index 7f876fb..6cdc5d3 100644 --- a/src/barabadb/query/parser.nim +++ b/src/barabadb/query/parser.nim @@ -1646,6 +1646,42 @@ proc parseDropGraph(p: var Parser): Node = result = Node(kind: nkDropGraph, dgName: name, dgIfExists: ifExists, line: tok.line, col: tok.col) +proc parseCreateDatabase(p: var Parser): Node = + let tok = p.expect(tkCreate) + discard p.expect(tkDatabase) + var ifNotExists = false + if p.peek().kind == tkIf: + discard p.advance() + discard p.expect(tkNot) + discard p.expect(tkExists) + ifNotExists = true + let name = p.expect(tkIdent).value + result = Node(kind: nkCreateDatabase, cdDbName: name, cdIfNotExists: ifNotExists, + line: tok.line, col: tok.col) + +proc parseDropDatabase(p: var Parser): Node = + let tok = p.expect(tkDrop) + discard p.expect(tkDatabase) + var ifExists = false + if p.peek().kind == tkIf: + discard p.advance() + discard p.expect(tkExists) + ifExists = true + let name = p.expect(tkIdent).value + result = Node(kind: nkDropDatabase, ddDbName: name, ddIfExists: ifExists, + line: tok.line, col: tok.col) + +proc parseUseDatabase(p: var Parser): Node = + let tok = p.expect(tkUse) + let name = p.expect(tkIdent).value + result = Node(kind: nkUseDatabase, udDbName: name, + line: tok.line, col: tok.col) + +proc parseShowDatabases(p: var Parser): Node = + let tok = p.expect(tkShow) + discard p.expect(tkDatabases) + result = Node(kind: nkShowDatabases, line: tok.line, col: tok.col) + proc parseStatement*(p: var Parser): Node = case p.peek().kind of tkWith, tkSelect: p.parseSelect() @@ -1674,6 +1710,8 @@ proc parseStatement*(p: var Parser): Node = p.parseCreatePolicy() elif next.kind == tkGraph: p.parseCreateGraph() + elif next.kind == tkDatabase: + p.parseCreateDatabase() else: p.parseCreateType() else: @@ -1695,6 +1733,8 @@ proc parseStatement*(p: var Parser): Node = p.parseDropPolicy() elif next.kind == tkGraph: p.parseDropGraph() + elif next.kind == tkDatabase: + p.parseDropDatabase() else: let tok = p.advance() Node(kind: nkNullLit, line: tok.line, col: tok.col) @@ -1713,6 +1753,14 @@ proc parseStatement*(p: var Parser): Node = p.parseRevoke() of tkSet: p.parseSetVar() + of tkUse: + p.parseUseDatabase() + of tkShow: + if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkDatabases: + p.parseShowDatabases() + else: + let tok = p.advance() + Node(kind: nkNullLit, line: tok.line, col: tok.col) of tkApply: if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkMigration: p.parseApplyMigration() diff --git a/src/baradadb.nim b/src/baradadb.nim index 0ad6870..1d1a129 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -17,9 +17,11 @@ import barabadb/protocol/ssl import barabadb/storage/lsm import barabadb/storage/compaction import barabadb/core/raft +import barabadb/query/executor import barabadb/core/gossip import barabadb/core/replication import barabadb/core/disttxn +import barabadb/core/registry import barabadb/tools/repair import barabadb/tools/migrate @@ -28,6 +30,10 @@ type db*: LSMTree strategy*: compaction.CompactionStrategy + MultiCompactionManager = ref object + registry: DatabaseRegistry + strategies: Table[string, compaction.CompactionStrategy] + proc newCompactionManager*(db: LSMTree): CompactionManager = result = CompactionManager(db: db, strategy: compaction.newCompactionStrategy(db.dir)) for sst in db.sstables: @@ -88,6 +94,95 @@ proc startCompactionLoop*(cm: CompactionManager, intervalMs: int = 60000) {.asyn await sleepAsync(intervalMs) cm.compact() +proc newMultiCompactionManager*(registry: DatabaseRegistry): MultiCompactionManager = + result = MultiCompactionManager(registry: registry, strategies: initTable[string, compaction.CompactionStrategy]()) + + # Initialize strategies for each existing database + for name in listDatabases(registry): + let info = getDatabaseInfo(registry, name) + if info != nil: + result.strategies[name] = compaction.newCompactionStrategy(info.db.dir) + for sst in info.db.sstables: + let meta = compaction.SSTableMeta( + path: sst.path, level: sst.level, minKey: sst.minKey, maxKey: sst.maxKey, + entryCount: sst.entryCount, sizeBytes: sst.entryCount * 64, createdAt: 0) + result.strategies[name].addTable(meta) + +proc compactAll(mcm: MultiCompactionManager) = + for name in listDatabases(mcm.registry): + let info = getDatabaseInfo(mcm.registry, name) + if info == nil: continue + let db = info.db + + # Initialize strategy if not already + if name notin mcm.strategies: + mcm.strategies[name] = compaction.newCompactionStrategy(db.dir) + for sst in db.sstables: + let meta = compaction.SSTableMeta( + path: sst.path, level: sst.level, minKey: sst.minKey, maxKey: sst.maxKey, + entryCount: sst.entryCount, sizeBytes: sst.entryCount * 64, createdAt: 0) + mcm.strategies[name].addTable(meta) + + let strategy = mcm.strategies[name] + acquire(db.lock) + try: + for level in 0 ..< compaction.MaxLevel: + if strategy.needsCompaction(level): + let result = strategy.compact(level) + if result.outputTables.len == 0: continue + + var newSSTables: seq[SSTable] = @[] + var removedPaths = initTable[string, bool]() + for t in result.inputTables: + removedPaths[t.path] = true + for sst in db.sstables: + if sst.path notin removedPaths: + newSSTables.add(sst) + + for meta in result.outputTables: + try: + var sst = loadSSTable(meta.path) + let sstName = splitFile(meta.path).name + sst.id = try: parseInt(sstName) except: db.nextSSTableId + sst.level = meta.level + newSSTables.add(sst) + db.nextSSTableId = max(db.nextSSTableId, sst.id + 1) + except CatchableError as e: + warn("Compaction output SSTable failed to load: " & meta.path & " - " & e.msg) + + newSSTables.sort(proc(a, b: SSTable): int = cmp(a.id, b.id)) + db.sstables = newSSTables + inc db.manifestSequence + try: + writeManifest(db) + except CatchableError as e: + warn("Failed to write MANIFEST after compaction: " & e.msg) + finally: + release(db.lock) + +proc startMultiCompactionLoop*(mcm: MultiCompactionManager, intervalMs: int = 60000) {.async.} = + while true: + await sleepAsync(intervalMs) + mcm.compactAll() + +proc migrateLegacyData(registry: DatabaseRegistry, config: BaraConfig) = + let legacyDir = config.dataDir / "server" + let defaultDir = registry.dataRoot / "default" + if dirExists(legacyDir / "sstables") and not dirExists(defaultDir): + info("Migrating legacy data from " & legacyDir & " to " & defaultDir) + createDir(defaultDir / "sstables") + createDir(defaultDir / "wal") + for kind, path in walkDir(legacyDir / "sstables"): + if kind == pcFile: + copyFile(path, defaultDir / "sstables" / splitPath(path).tail) + for kind, path in walkDir(legacyDir / "wal"): + if kind == pcFile: + copyFile(path, defaultDir / "wal" / splitPath(path).tail) + if fileExists(legacyDir / "MANIFEST"): + copyFile(legacyDir / "MANIFEST", defaultDir / "MANIFEST") + moveDir(legacyDir, legacyDir & ".migrated") + info("Legacy data migration complete. Original renamed to " & legacyDir & ".migrated") + proc runTcpServer(config: BaraConfig) {.async.} = info("BaraDB TCP listening on " & config.address & ":" & $config.port) var server = newServer(config) @@ -230,34 +325,45 @@ proc main() = warn("Failed to generate self-signed certificate. TLS disabled.") config.tlsEnabled = false - # Shared LSMTree instance for both TCP and HTTP servers - let dataDir = config.dataDir / "server" - let sharedDb = newLSMTree(dataDir) + # Create database registry with multi-database support + let registry = newDatabaseRegistry (config) + registry.setContextFactory(proc(db: LSMTree, reg: DatabaseRegistry): ContextRef {.closure.} = + cast[ContextRef](cast[pointer](newExecutionContext(db, reg)))) + + # Migrate legacy data from old single-DB layout + migrateLegacyData(registry, config) + + # Load existing databases and ensure default + registry.loadExistingDatabases() + registry.ensureDefaultDatabase() + + info("Databases loaded: " & $listDatabases(registry).join(", ")) # Start HTTP server (blocking, multi-threaded via hunos) in background thread - var httpServer = newHttpServerWithDb(config, sharedDb) + var httpServer = newHttpServerWithRegistry(config, registry) spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440 - # Start background compaction loop - let cm = newCompactionManager(sharedDb) - asyncCheck cm.startCompactionLoop() + # Start background compaction loop for all databases + let mcm = newMultiCompactionManager(registry) + asyncCheck mcm.startMultiCompactionLoop() # Create TCP server (initialization is synchronous, run is async) let localId {.used.} = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port - var tcpServer = newServerWithDb(config, sharedDb) + var tcpServer = newServerWithRegistry(config, registry) # Start Raft cluster if enabled if config.raftEnabled: info("Starting Raft node " & config.raftNodeId & " on port " & $config.raftPort) var raftNode = newRaftNode(config.raftNodeId, config.raftPeers, config.raftPort) - # Wire state machine to apply committed entries to the database + # Wire state machine to apply committed entries to the default database + let defaultDbInfo = getDatabaseInfo(registry, "default") raftNode.applyCommand = proc(cmd: string, data: seq[byte]) {.gcsafe.} = if cmd == "put": let parts = cast[string](data).split("\x00") if parts.len >= 2: - sharedDb.put(parts[0], cast[seq[byte]](parts[1])) + defaultDbInfo.db.put(parts[0], cast[seq[byte]](parts[1])) elif cmd == "delete": - sharedDb.delete(cast[string](data)) + defaultDbInfo.db.delete(cast[string](data)) # Wire RAFT ↔ DistTxn wireRaftDistTxn(raftNode, tcpServer) @@ -291,6 +397,7 @@ proc main() = # Shutdown httpServer.stop() + tcpServer.stop() if tcpServer.gossipProtocol != nil: tcpServer.gossipProtocol.stop() diff --git a/tests/bugfix_test b/tests/bugfix_test new file mode 100755 index 0000000..5dccfbf Binary files /dev/null and b/tests/bugfix_test differ diff --git a/tests/bugfix_test.nim b/tests/bugfix_test.nim index ddde124..c27a8a8 100644 --- a/tests/bugfix_test.nim +++ b/tests/bugfix_test.nim @@ -3,6 +3,7 @@ import std/strutils import std/os import std/tables import ../src/barabadb/query/[parser, executor, lexer, ast] +import ../src/barabadb/core/types import ../src/barabadb/storage/lsm const testDir = "/tmp/baradb_bugfix_test" @@ -73,7 +74,7 @@ suite "Bug fixes — IN list, nkPath exprToSql, multi-table joins": check r.success check r.rows.len == 1 # Column is "t.id" (qualified by alias) - check r.rows[0]["t.id"] == "42" + check valueToString(r.rows[0]["t.id"]) == "42" test "IN (list) executes with actual data": var ctx = setupCtx() @@ -123,9 +124,9 @@ suite "Bug fixes — IN list, nkPath exprToSql, multi-table joins": check r.success check r.rows.len == 1 # Verify qualified column references resolve correctly - check r.rows[0]["post.id"] == "100" - check r.rows[0]["users.name"] == "Alice" - check r.rows[0]["post.thread"] == "5" + check valueToString(r.rows[0]["post.id"]) == "100" + check valueToString(r.rows[0]["users.name"]) == "Alice" + check valueToString(r.rows[0]["post.thread"]) == "5" # Check column names don't contain "nkPath" for col in r.columns: check not col.contains("nkPath") @@ -178,10 +179,10 @@ suite "Bug fixes — IN list, nkPath exprToSql, multi-table joins": check r.rows.len == 4 # After sorting by name ASC, then id DESC within same name: # alice(3), alice(1), bob(2), charlie(4) - check r.rows[0]["id"] == "3" - check r.rows[1]["id"] == "1" - check r.rows[2]["id"] == "2" - check r.rows[3]["id"] == "4" + check valueToString(r.rows[0]["id"]) == "3" + check valueToString(r.rows[1]["id"]) == "1" + check valueToString(r.rows[2]["id"]) == "2" + check valueToString(r.rows[3]["id"]) == "4" test "Numeric != comparison is consistent with =": var ctx = setupCtx() @@ -203,7 +204,7 @@ suite "Bug fixes — IN list, nkPath exprToSql, multi-table joins": # Column names should match the full path "posts.id", "posts.title" check r.columns == @["posts.id", "posts.title"] # Data should be accessible via both qualified and unqualified names - check r.rows[0]["posts.id"] == "1" + check valueToString(r.rows[0]["posts.id"]) == "1" test "DELETE inside transaction actually removes row": var ctx = setupCtx() @@ -244,8 +245,8 @@ suite "Bug fixes — IN list, nkPath exprToSql, multi-table joins": let rmin = executeQuery(ctx, parse("SELECT MIN(val) AS m FROM scores")) check rmin.success check rmin.rows.len == 1 - check rmin.rows[0]["m"] == "10" + check valueToString(rmin.rows[0]["m"]) == "10" let rmax = executeQuery(ctx, parse("SELECT MAX(val) AS m FROM scores")) check rmax.success check rmax.rows.len == 1 - check rmax.rows[0]["m"] == "30" + check valueToString(rmax.rows[0]["m"]) == "30" diff --git a/tests/join_tests.nim b/tests/join_tests.nim index f0c7972..8152c87 100644 --- a/tests/join_tests.nim +++ b/tests/join_tests.nim @@ -3,6 +3,8 @@ import std/os import std/strutils import std/monotimes import std/tables +import std/sequtils +import barabadb/core/types import barabadb/query/executor as qexec import barabadb/query/parser import barabadb/storage/lsm @@ -97,7 +99,7 @@ suite "JOIN execution": var foundBob = false for row in r.rows: if row["u.name"] == "Bob": - check row["x.total"] == "\\N" + check row["x.total"].kind == vkNull foundBob = true check foundBob diff --git a/tests/test_all.nim b/tests/test_all.nim index 904b42b..249e954 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -2831,7 +2831,7 @@ suite "Window Functions": # Sales: Diana(75000)=1, Charlie(70000)=2 var found = initTable[string, string]() for row in r.rows: - found[row["name"]] = row["rn"] + found[$row["name"]] = $row["rn"] check found["Eve"] == "1" check found["Alice"] == "2" check found["Bob"] == "3" @@ -3474,7 +3474,7 @@ suite "Auto-Increment & ID Generators": let r = qexec.executeQuery(ctx, parse("SELECT gen_random_uuid() AS uid")) check r.success check r.rows.len == 1 - let uid = r.rows[0]["uid"] + let uid = $r.rows[0]["uid"] check uid.len == 36 check uid[8] == '-' check uid[13] == '-' @@ -3486,7 +3486,7 @@ suite "Auto-Increment & ID Generators": let r = qexec.executeQuery(ctx, parse("SELECT uuid() AS uid")) check r.success check r.rows.len == 1 - check r.rows[0]["uid"].len == 36 + check ($r.rows[0]["uid"]).len == 36 test "Two UUIDs are different": let r = qexec.executeQuery(ctx, parse("SELECT uuid() AS a, uuid() AS b")) @@ -3564,7 +3564,7 @@ suite "Auto-Increment & ID Generators": let r = qexec.executeQuery(ctx, parse("SELECT snowflake_id(1) AS sid")) check r.success check r.rows.len == 1 - let sid = r.rows[0]["sid"] + let sid = $r.rows[0]["sid"] check sid.len > 0 var num: int64 = 0 try: @@ -3720,17 +3720,17 @@ suite "Type Safety — evalExprValue": check r.success check r.rows[0]["r"] == "17.5" - test "evalExprValue returns correct Value kind for literals": + test "evalExpr returns correct Value kind for literals": let lit = IRExpr(kind: irekLiteral, valueKind: vkInt64) lit.literal = IRLiteral(kind: vkInt64, int64Val: 42) - let v = evalExprValue(lit, initTable[string, string](), nil) + let v = evalExpr(lit, initTable[string, Value](), nil) check v.kind == vkInt64 check v.int64Val == 42 - test "evalExprValue returns correct Value kind for float literal": + test "evalExpr returns correct Value kind for float literal": let lit = IRExpr(kind: irekLiteral, valueKind: vkFloat64) lit.literal = IRLiteral(kind: vkFloat64, float64Val: 3.14) - let v = evalExprValue(lit, initTable[string, string](), nil) + let v = evalExpr(lit, initTable[string, Value](), nil) check v.kind == vkFloat64 check v.float64Val == 3.14 @@ -3998,7 +3998,7 @@ suite "Hybrid RAG Search": let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search('docs', 'embedding', 'content', 'quick brown', '[1.0, 0.0, 0.0]', 10) AS res")) check r.success check r.rows.len == 1 - let jsonStr = r.rows[0]["res"] + let jsonStr = $r.rows[0]["res"] check jsonStr.len > 2 # not "[]" let arr = parseJson(jsonStr) check arr.kind == JArray @@ -4013,7 +4013,7 @@ suite "Hybrid RAG Search": let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_ids('docs2', 'embedding', 'content', 'machine learning', '[0.0, 1.0, 0.0]', 10) AS ids")) check r.success check r.rows.len == 1 - let idsStr = r.rows[0]["ids"] + let idsStr = $r.rows[0]["ids"] check idsStr.len > 0 check idsStr.contains("20") @@ -4029,7 +4029,7 @@ suite "Hybrid RAG Search": discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts3 ON docs3(content) USING FTS")) let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search('docs3', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10) AS res")) check r.success - let arr = parseJson(r.rows[0]["res"]) + let arr = parseJson($r.rows[0]["res"]) check arr.len == 3 # Doc 3 should be first (matches both vector and FTS), doc 1 second (vector only), doc 2 third (no match) check arr[0]["id"].getStr() == "3" @@ -4038,7 +4038,7 @@ suite "Hybrid RAG Search": let r = qexec.executeQuery(ctx, parse("SELECT rerank('quick brown', '[{\"id\":\"1\",\"score\":\"0.5\"},{\"id\":\"2\",\"score\":\"0.5\"}]') AS res")) check r.success # Both have same score, rerank should preserve order (no content to boost) - let arr = parseJson(r.rows[0]["res"]) + let arr = parseJson($r.rows[0]["res"]) check arr.kind == JArray check arr.len == 2 @@ -4060,7 +4060,7 @@ suite "Hybrid RAG Search": discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts5 ON docs5(content) USING FTS")) let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_filtered('docs5', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10, 'tenant_id', 'tenant-a') AS res")) check r.success - let arr = parseJson(r.rows[0]["res"]) + let arr = parseJson($r.rows[0]["res"]) # Doc 1 (tenant-a) should be first and highest scored; Doc 2 (tenant-b) must be excluded check arr[0]["id"].getStr() == "1" for elem in arr: @@ -4074,6 +4074,6 @@ suite "Hybrid RAG Search": discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts6 ON docs6(content) USING FTS")) let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_filtered('docs6', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10, '', '') AS res")) check r.success - let arr = parseJson(r.rows[0]["res"]) + let arr = parseJson($r.rows[0]["res"]) check arr.len == 2