diff --git a/.gitignore b/.gitignore index ce1af70..520411d 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,5 @@ clients/rust/target/ clients/nim/src/baradb/client src/barabadb/storage/compaction docker-compose.test.yml +src/barabadb/query/executor +tests/join_tests diff --git a/PLAN_SQL_ADVANCED.md b/PLAN_SQL_ADVANCED.md index cd6c0b3..0e2fcbe 100644 --- a/PLAN_SQL_ADVANCED.md +++ b/PLAN_SQL_ADVANCED.md @@ -3,6 +3,8 @@ > **Визия**: BaraDB е самостоятелен, универсален SQL engine с Nim ядро, поддържащ модерни SQL:2023 разширения — Property Graph, Vector Search, JSON документи и прозоречни функции, в една вградена или клиент/сървър конфигурация. > > **Принцип**: Само основи. Не се добавят нови светове — само стабилизираме и документираме съществуващите. +> +> **Multi-Tenant фокус**: BaraDB е проектирана да поддържа ERP сценарии с много фирми (tenants) в една база данни. Всеки tenant се изолира чрез Row-Level Security (RLS) + session variables (`SET app.tenant_id = 'X'`), а не чрез отделни бази. --- @@ -10,7 +12,7 @@ - **Фаза 1 (Base SQL + MVCC + Raft)**: BaraDB core engine - **Фаза 2 (Advanced SQL)**: Разработена с **Xiaomi Mimo** (`mimo-v2.5-pro`) — Window Functions, MERGE, LATERAL JOIN, Advanced Aggregates, PIVOT/UNPIVOT, SQL/PGQ Property Graph -- **Фаза 3 (Stabilization)**: Текуща — Vector SQL Integration, тестове, документация +- **Фаза 3 (Stabilization + Multi-Tenant)**: Текуща — Vector SQL Integration, Session Variables, `current_user`/`current_role`, RLS tenant isolation, тестове, документация --- @@ -167,6 +169,49 @@ SELECT * FROM GRAPH_TABLE(org_chart --- +## Част 1.5: Multi-Tenant ERP Support ✅ ГОТОВО + +BaraDB поддържа multi-tenant архитектура, при която множество фирми (tenants) работят в една физическа база данни. Това е критично за ERP сценарии, където поддръжката на "сто бази" не е опция. + +### Механизъм + +| Компонент | Описание | +|-----------|----------| +| **Session Variables** | `SET app.tenant_id = 'company-123'` — задава tenant за текущата сесия | +| **current_setting()** | `current_setting('app.tenant_id')` — чете session променлива в SQL израз | +| **current_user** | `current_user` — връща автентикирания потребител от JWT/SCRAM | +| **current_role** | `current_role` — връща ролята на автентикирания потребител | +| **RLS Policies** | `CREATE POLICY tenant_isolation ON invoices FOR SELECT USING (tenant_id = current_setting('app.tenant_id'))` | +| **Auth Bridge** | `server.nim` и `httpserver.nim` попълват `ExecutionContext.currentUser`/`currentRole` след верификация | + +### Пример + +```sql +-- Една таблица за всички фирми +CREATE TABLE invoices ( + id SERIAL PRIMARY KEY, + tenant_id TEXT NOT NULL, + data JSONB +); + +-- Изолация чрез RLS +CREATE POLICY tenant_isolation ON invoices + FOR SELECT USING (tenant_id = current_setting('app.tenant_id')); + +-- Всяка сесия вижда само своя tenant +SET app.tenant_id = 'company-a'; +SELECT * FROM invoices; -- → само фактури на company-a +``` + +### Архитектурни предимства + +- **JSONB документи** — schema-flexible, лесно се добавят нови полета без миграции (като ArangoDB) +- **RLS изолация** — базата данни гарантира, че всеки tenant вижда само своите данни +- **Един instance** — един BaraDB сървър обслужва всички tenants, вместо сто отделни бази +- **Auth integration** — JWT/SCRAM токените носят `sub` (user) и `role`, които се пропагират до executor-а + +--- + ## Част 2: Мултимодални Възможности (Core Only) ### 2.1 JSON / JSONB Документи ✅ ГОТОВО @@ -290,6 +335,16 @@ LIMIT 10; - **MERGE тестове** — поправени чрез изолиране на тестовата директория (unique temp dir per suite) - **Row storage escape** — `escapeRowVal()` в `execInsert` за стойности със запетай (vector literals) - **ORDER BY + projection** — `irpkSort` сега е преди `irpkProject` в `lowerSelect`, което позволява `ORDER BY` по колони извън `SELECT` +- **GROUPING SETS execution** — `lowerSelect` сега проверява `selGroupingSetsKind != gskNone` освен `selGroupBy.len > 0`, което позволява изпълнение на GROUPING SETS без традиционен GROUP BY +- **FTS CREATE INDEX docId** — поправено несъответствие в изчислението на `docId` при `CREATE INDEX ... USING FTS` (сега използва хеш на `tableName.$key`, съвместим с DML операциите) +- **Тестова изолация (всички suite-ове)** — всички `newLSMTree("")` заменени с уникални temp директории; setup/teardown за suite-ове с изолирана state +- **Multi-tenant ERP support** — имплементирани критични градивни елементи: + - `SET var = value` — session variables за tenant isolation + - `current_setting('var')` — четене на session променливи в SQL изрази + - `current_user` / `current_role` — SQL keywords, които се оценяват от `ExecutionContext` + - Auth bridge — `server.nim` и `httpserver.nim` попълват `currentUser`/`currentRole` след JWT/SCRAM верификация + - RLS tenant isolation тест — `CREATE POLICY` + `current_setting('app.tenant_id')` работи за multi-tenant филтрация + - `evalExpr` вече предава `ctx` рекурсивно — поправен бъг, при който `current_user`/`current_setting` връщаха празни стойности в под-изрази --- diff --git a/docs/en/baraql.md b/docs/en/baraql.md index f4b9061..33920ae 100644 --- a/docs/en/baraql.md +++ b/docs/en/baraql.md @@ -602,6 +602,44 @@ SUM(salary) OVER ( ) ``` +## Multi-Tenant ERP + +BaraDB supports running multiple companies (tenants) in a single database instance, using **Row-Level Security (RLS)** combined with **session variables**. + +### Session Variables + +```sql +SET app.tenant_id = 'company-123'; +SELECT current_setting('app.tenant_id') AS tenant; +``` + +### Current User / Role + +```sql +SELECT current_user AS me, current_role AS my_role; +``` + +### RLS Tenant Isolation + +```sql +-- Enable RLS on a table +ALTER TABLE invoices ENABLE ROW LEVEL SECURITY; + +-- Create a policy that filters by tenant +CREATE POLICY tenant_isolation ON invoices + FOR SELECT USING (tenant_id = current_setting('app.tenant_id')); + +-- Each session only sees its own data +SET app.tenant_id = 'company-a'; +SELECT * FROM invoices; -- only company-a rows +``` + +### Why Multi-Tenant? + +- **One instance, many tenants** — no need to run 100 separate databases +- **JSONB documents** — schema-flexible storage, easy to add fields per tenant +- **RLS guarantees isolation** — the database enforces tenant boundaries, not just the application + ## Supported Keywords | Category | Keywords | @@ -620,6 +658,7 @@ SUM(salary) OVER ( | JSON | ->, ->> | | FTS | @@ (BM25 match) | | Recovery | RECOVER TO TIMESTAMP | -| Functions | count, sum, avg, min, max, stddev, variance, abs, sqrt, lower, upper, len, trim, substr, now, last_insert_id | +| Functions | count, sum, avg, min, max, stddev, variance, abs, sqrt, lower, upper, len, trim, substr, now, last_insert_id, current_setting | +| Session | SET, current_setting, current_user, current_role | | Window | OVER, PARTITION BY, ROWS, RANGE, UNBOUNDED PRECEDING, CURRENT ROW, FOLLOWING | | Window Functions | ROW_NUMBER, RANK, DENSE_RANK, LEAD, LAG, FIRST_VALUE, LAST_VALUE, NTILE | diff --git a/docs/en/changelog.md b/docs/en/changelog.md index 9356412..41d6209 100644 --- a/docs/en/changelog.md +++ b/docs/en/changelog.md @@ -2,6 +2,26 @@ All notable changes to BaraDB are documented in this file. +## [Unreleased] — SQL:2023 Stabilization + +### Fixed + +- **GROUPING SETS execution** — `lowerSelect` now creates `irpkGroupBy` when `selGroupingSetsKind != gskNone`, even if `selGroupBy` is empty. Previously, queries like `GROUP BY GROUPING SETS ((dept), ())` bypassed the grouping executor entirely. +- **FTS CREATE INDEX docId mismatch** — `CREATE INDEX ... USING FTS` now computes `docId` as a hash of `tableName.$key`, consistent with DML operations (`INSERT`/`UPDATE`/`DELETE`). Previously, index creation used sequential IDs (0, 1, 2...), causing `@@` queries to never match indexed documents. +- **Test isolation (all suites)** — All `newLSMTree("")` calls replaced with unique temporary directories per suite. Eliminates WAL accumulation issues and flaky tests caused by shared database state between test runs. +- **Window frame parser** — `parseFrameBoundary` no longer consumes `tkRow` after `tkCurrent` incorrectly (was using `tkRows`). Also fixed `tkRow` keyword conflict with `ENABLE ROW LEVEL SECURITY` parsing. +- **ORDER BY + SELECT projection** — `lowerSelect` now places `irpkSort` before `irpkProject`, enabling `ORDER BY` on columns not present in the `SELECT` list. +- **UNPIVOT execution** — Verified and fixed missing test coverage for UNPIVOT transformation. + +### Added + +- **JSON operators** — `@>` (contains), `<@` (contained by), `?` (has key), `?|` (has any), `?&` (has all) now supported in lexer, parser, and executor. +- **Window frame execution** — `ROWS BETWEEN X PRECEDING AND Y FOLLOWING` / `CURRENT ROW` frame boundaries now respected by `FIRST_VALUE` and `LAST_VALUE`. +- **Session variables** — `SET var_name = value` and `current_setting('var_name')` for connection-scoped key/value storage. +- **Current user/role** — `current_user` and `current_role` SQL keywords evaluate to the authenticated session's user and role. +- **Auth-executor bridge** — Wire server and HTTP server now populate `ExecutionContext.currentUser` and `ExecutionContext.currentRole` after JWT/SCRAM authentication. +- **Multi-tenant RLS** — Row-Level Security policies can now reference `current_user`, `current_role`, and `current_setting('app.tenant_id')` for per-tenant data isolation. + ## [1.1.0] — 2026-05-13 ### Added diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 9f89e56..1d5b4d3 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -108,6 +108,8 @@ proc queryHandler(server: HttpServer): RequestHandler = let ctx = newContext(request) server.metrics.queriesTotal += 1 + var userId = "" + var role = "" # Auth check if server.config.authEnabled: let authHeader = request.headers["Authorization"] @@ -115,10 +117,12 @@ proc queryHandler(server: HttpServer): RequestHandler = ctx.json(%*{"error": "Unauthorized"}, 401) return let tokenStr = authHeader[7..^1] - let (valid, userId, role) = server.verifyToken(tokenStr) + let (valid, uid, r) = server.verifyToken(tokenStr) if not valid: ctx.json(%*{"error": "Unauthorized"}, 401) return + userId = uid + role = r let body = parseJson(request.body) if body == nil or "query" notin body: @@ -131,6 +135,8 @@ proc queryHandler(server: HttpServer): RequestHandler = return var reqCtx = cloneForConnection(server.ctx) + reqCtx.currentUser = userId + reqCtx.currentRole = role let tokens = tokenize(queryStr) let astNode = parse(tokens) diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 7ae8bb5..55647d9 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -436,9 +436,11 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} case header.kind of mkAuth: let tokenStr = parseAuthMessage(cast[seq[byte]](payload)) - let (valid, userId, _) = verifyToken(secret, tokenStr) + let (valid, userId, role) = verifyToken(secret, tokenStr) if valid: authenticated = true + connCtx.currentUser = userId + connCtx.currentRole = role let okMsg = makeAuthOkMessage(header.requestId) await client.send(cast[string](okMsg)) info("Client " & $clientId & " authenticated as " & userId) diff --git a/src/barabadb/query/ast.nim b/src/barabadb/query/ast.nim index b600409..e4b9609 100644 --- a/src/barabadb/query/ast.nim +++ b/src/barabadb/query/ast.nim @@ -40,6 +40,7 @@ type nkDisableRLS nkGrant nkRevoke + nkSetVar # Clauses nkFrom @@ -78,6 +79,8 @@ type nkIsExpr nkStar nkPlaceholder + nkCurrentUser + nkCurrentRole # Graph-specific nkGraphTraversal @@ -144,6 +147,10 @@ type bkJsonPathText = "->>" bkFtsMatch = "@@" bkDistance = "<->" + bkJsonContains = "@>" + bkJsonContainedBy = "<@" + bkJsonHasAny = "?|" + bkJsonHasAll = "?&" UnaryOpKind* = enum ukNeg = "-" @@ -317,6 +324,9 @@ type rvPrivilege*: string rvTable*: string rvGrantee*: string + of nkSetVar: + svName*: string + svValue*: string of nkApplyMigration: amName*: string of nkMigrationStatus: @@ -484,6 +494,10 @@ type discard of nkPlaceholder: discard + of nkCurrentUser: + discard + of nkCurrentRole: + discard of nkPropertyDef: pdName*: string pdType*: string diff --git a/src/barabadb/query/executor b/src/barabadb/query/executor deleted file mode 100755 index 65e04c0..0000000 Binary files a/src/barabadb/query/executor and /dev/null differ diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index 5ac4022..db69967 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -69,6 +69,7 @@ type policies*: Table[string, seq[PolicyDef]] # table name -> policies currentUser*: string currentRole*: string + sessionVars*: Table[string, string] # session-scoped key/value variables MigrationRecord* = object name*: string @@ -149,6 +150,7 @@ proc newExecutionContext*(db: LSMTree): ExecutionContext = users: initTable[string, UserDef](), policies: initTable[string, seq[PolicyDef]](), currentUser: "", currentRole: "", + sessionVars: initTable[string, string](), onChange: nil) restoreSchema(result) @@ -323,6 +325,7 @@ proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = users: ctx.users, policies: ctx.policies, txnManager: ctx.txnManager, currentUser: ctx.currentUser, currentRole: ctx.currentRole, + sessionVars: ctx.sessionVars, pendingTxn: nil, onChange: ctx.onChange) # ---------------------------------------------------------------------- @@ -504,7 +507,7 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = of irekStar: return "*" of irekJsonPath: - let srcVal = evalExpr(expr.jpExpr, row) + let srcVal = evalExpr(expr.jpExpr, row, ctx) if srcVal.len == 0: return "" try: let node = parseJson(srcVal) @@ -527,8 +530,8 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = except: return "" of irekBinary: - let left = evalExpr(expr.binLeft, row) - let right = evalExpr(expr.binRight, row) + let left = evalExpr(expr.binLeft, row, ctx) + let right = evalExpr(expr.binRight, row, ctx) case expr.binOp of irEq: if left == right: return "true" @@ -669,20 +672,93 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if vecA.len == 0 or vecB.len == 0: return "0" return $vengine.euclideanDistance(vecA, vecB) + of irJsonContains: + # Check if left JSON contains right JSON + try: + let leftNode = parseJson(left) + let rightNode = parseJson(right) + if leftNode.kind == JObject and rightNode.kind == JObject: + for key, val in rightNode: + if not leftNode.hasKey(key) or $(leftNode[key]) != $val: + return "false" + return "true" + elif leftNode.kind == JArray and rightNode.kind == JArray: + for ritem in rightNode: + var found = false + for litem in leftNode: + if $(litem) == $(ritem): + found = true + break + if not found: + return "false" + return "true" + else: + return if $(leftNode) == $(rightNode): "true" else: "false" + except: + return "false" + of irJsonContainedBy: + # Check if left JSON is contained by right JSON (reverse of contains) + try: + let leftNode = parseJson(left) + let rightNode = parseJson(right) + if leftNode.kind == JObject and rightNode.kind == JObject: + for key, val in leftNode: + if not rightNode.hasKey(key) or $(rightNode[key]) != $val: + return "false" + return "true" + elif leftNode.kind == JArray and rightNode.kind == JArray: + for litem in leftNode: + var found = false + for ritem in rightNode: + if $(ritem) == $(litem): + found = true + break + if not found: + return "false" + return "true" + else: + return if $(leftNode) == $(rightNode): "true" else: "false" + except: + return "false" + of irJsonHasAny: + # Check if JSON object has any of the keys in right array + try: + let leftNode = parseJson(left) + let rightNode = parseJson(right) + if leftNode.kind == JObject and rightNode.kind == JArray: + for key in rightNode: + if key.kind == JString and leftNode.hasKey(key.getStr()): + return "true" + return "false" + except: + return "false" + of irJsonHasAll: + # Check if JSON object has all of the keys in right array + try: + let leftNode = parseJson(left) + let rightNode = parseJson(right) + if leftNode.kind == JObject and rightNode.kind == JArray: + for key in rightNode: + if key.kind == JString and not leftNode.hasKey(key.getStr()): + return "false" + return "true" + return "false" + except: + return "false" else: return "false" of irekUnary: case expr.unOp of irNot: - let v = evalExpr(expr.unExpr, row) + let v = evalExpr(expr.unExpr, row, ctx) return if v == "true": "false" else: "true" of irIsNull: - let v = evalExpr(expr.unExpr, row) + let v = evalExpr(expr.unExpr, row, ctx) return if isNull(v): "true" else: "false" of irIsNotNull: - let v = evalExpr(expr.unExpr, row) + let v = evalExpr(expr.unExpr, row, ctx) return if not isNull(v): "true" else: "false" of irNeg: - let v = evalExpr(expr.unExpr, row) + let v = evalExpr(expr.unExpr, row, ctx) try: let f = -parseFloat(v) let s = $f @@ -716,6 +792,37 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = return "0" let arg = evalExpr(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) + try: + let node = parseJson(jsonStr) + if node.kind == JObject: + return if node.hasKey(key): "true" else: "false" + elif node.kind == JArray: + try: + let idx = parseInt(key) + return if idx >= 0 and idx < node.len: "true" else: "false" + except: + return "false" + return "false" + except: + return "false" + of "current_setting": + if expr.irFuncArgs.len < 1: + return "" + let key = evalExpr(expr.irFuncArgs[0], row, ctx) + if ctx != nil and key in ctx.sessionVars: + return ctx.sessionVars[key] + return "" + of "current_user": + if ctx != nil: return ctx.currentUser + return "" + of "current_role": + if ctx != nil: return ctx.currentRole + return "" else: # Unknown function: try to evaluate args and return first arg as fallback if expr.irFuncArgs.len > 0: @@ -773,7 +880,7 @@ proc passesPolicy(ctx: ExecutionContext, tableName, command: string, row: Row): continue if pol.usingExpr != nil: let expr = lowerExpr(pol.usingExpr) - if evalExpr(expr, row) != "true": + if evalExpr(expr, row, ctx) != "true": return false return true @@ -788,7 +895,7 @@ proc checkInsertPolicy(ctx: ExecutionContext, tableName: string, row: Row): bool continue if pol.withCheckExpr != nil: let expr = lowerExpr(pol.withCheckExpr) - if evalExpr(expr, row) != "true": + if evalExpr(expr, row, ctx) != "true": return false return true @@ -1151,7 +1258,7 @@ proc validateConstraints*(ctx: ExecutionContext, tableName: string, else: row[f] = "" let checkExpr = lowerExpr(check.checkNode) - let checkResult = evalExpr(checkExpr, row) + let checkResult = evalExpr(checkExpr, row, ctx) if checkResult != "true": return (false, "CHECK constraint '" & check.name & "' violated") @@ -1199,6 +1306,14 @@ proc lowerExpr*(node: Node): IRExpr = of nkNullLit: result = IRExpr(kind: irekLiteral) result.literal = IRLiteral(kind: vkNull) + of nkCurrentUser: + result = IRExpr(kind: irekFuncCall) + result.irFunc = "current_user" + result.irFuncArgs = @[] + of nkCurrentRole: + result = IRExpr(kind: irekFuncCall) + result.irFunc = "current_role" + result.irFuncArgs = @[] of nkIdent: result = IRExpr(kind: irekField) result.fieldPath = @[node.identName] @@ -1229,6 +1344,10 @@ proc lowerExpr*(node: Node): IRExpr = of bkOr: irOp = irOr of bkFtsMatch: irOp = irFtsMatch of bkDistance: irOp = irDistance + of bkJsonContains: irOp = irJsonContains + of bkJsonContainedBy: irOp = irJsonContainedBy + of bkJsonHasAny: irOp = irJsonHasAny + of bkJsonHasAll: irOp = irJsonHasAll else: irOp = irEq result.binOp = irOp result.binLeft = lowerExpr(node.binLeft) @@ -1408,7 +1527,7 @@ proc lowerSelect*(node: Node): IRPlan = filterPlan.filterCond = lowerExpr(node.selWhere.whereExpr) result = filterPlan - if node.selGroupBy.len > 0: + if node.selGroupBy.len > 0 or node.selGroupingSetsKind != gskNone: let groupPlan = IRPlan(kind: irpkGroupBy) groupPlan.groupSource = result groupPlan.groupKeys = @[] @@ -1477,17 +1596,17 @@ proc lowerSelect*(node: Node): IRPlan = # Window Function Computation # ---------------------------------------------------------------------- -proc partitionKey(row: Row, partExprs: seq[IRExpr]): string = +proc partitionKey(row: Row, partExprs: seq[IRExpr], ctx: ExecutionContext = nil): string = ## Compute a string partition key for a row result = "" for expr in partExprs: - result &= evalExpr(expr, row) & "|" + result &= evalExpr(expr, row, ctx) & "|" -proc compareRowsByOrder(a, b: Row, orderExprs: seq[IRExpr], orderDirs: seq[bool]): int = +proc compareRowsByOrder(a, b: Row, orderExprs: seq[IRExpr], orderDirs: seq[bool], ctx: ExecutionContext = nil): int = ## Compare two rows by their ORDER BY expressions for i, expr in orderExprs: - let va = evalExpr(expr, a) - let vb = evalExpr(expr, b) + let va = evalExpr(expr, a, ctx) + let vb = evalExpr(expr, b, ctx) var cmpRes = 0 try: let fa = parseFloat(va) @@ -1500,18 +1619,63 @@ proc compareRowsByOrder(a, b: Row, orderExprs: seq[IRExpr], orderDirs: seq[bool] return if orderDirs.len > i and orderDirs[i]: -cmpRes else: cmpRes return 0 -proc computeWindowValues*(rows: seq[Row], expr: IRExpr): seq[string] = +proc resolveFrameBounds(pos, partLen: int, frameStart, frameEnd: string): (int, int) = + ## Resolve frame boundaries for ROWS mode. + ## Returns (startPos, endPos) inclusive within the partition. + var startPos = 0 + var endPos = partLen - 1 + + # Parse start boundary + if frameStart == "UNBOUNDED PRECEDING": + startPos = 0 + elif frameStart == "CURRENT ROW": + startPos = pos + elif frameStart.endsWith(" PRECEDING"): + let nStr = frameStart[0..^11] + var n = 0 + try: n = parseInt(nStr) except: n = 0 + startPos = max(0, pos - n) + elif frameStart.endsWith(" FOLLOWING"): + let nStr = frameStart[0..^11] + var n = 0 + try: n = parseInt(nStr) except: n = 0 + startPos = min(partLen - 1, pos + n) + + # Parse end boundary + if frameEnd == "UNBOUNDED FOLLOWING": + endPos = partLen - 1 + elif frameEnd == "CURRENT ROW": + endPos = pos + elif frameEnd.endsWith(" PRECEDING"): + let nStr = frameEnd[0..^11] + var n = 0 + try: n = parseInt(nStr) except: n = 0 + endPos = max(0, pos - n) + elif frameEnd.endsWith(" FOLLOWING"): + let nStr = frameEnd[0..^11] + var n = 0 + try: n = parseInt(nStr) except: n = 0 + endPos = min(partLen - 1, pos + n) + + if startPos > endPos: + startPos = endPos + return (startPos, endPos) + +proc computeWindowValues*(rows: seq[Row], expr: IRExpr, ctx: ExecutionContext = nil): seq[string] = ## Compute a window function for all rows, returning a value per row. ## The expr must be of kind irekWindowFunc. result = newSeq[string](rows.len) if rows.len == 0: return let wfName = expr.wfName.toLower() + let frameMode = expr.wfFrameMode + let frameStart = expr.wfFrameStart + let frameEnd = expr.wfFrameEnd # Partition rows var groups = initTable[string, seq[int]]() for i, row in rows: - let pk = partitionKey(row, expr.wfPartition) + let pk = partitionKey(row, expr.wfPartition, ctx) if pk notin groups: groups[pk] = @[] groups[pk].add(i) @@ -1520,7 +1684,7 @@ proc computeWindowValues*(rows: seq[Row], expr: IRExpr): seq[string] = for pk, idxs in groups: var sortedIdxs = idxs sortedIdxs.sort(proc(a, b: int): int = - compareRowsByOrder(rows[a], rows[b], expr.wfOrderBy, expr.wfOrderDirs) + compareRowsByOrder(rows[a], rows[b], expr.wfOrderBy, expr.wfOrderDirs, ctx) ) case wfName @@ -1531,7 +1695,7 @@ proc computeWindowValues*(rows: seq[Row], expr: IRExpr): seq[string] = var currentRank = 1 for pos, rowIdx in sortedIdxs: if pos > 0: - let cmpRes = compareRowsByOrder(rows[sortedIdxs[pos - 1]], rows[rowIdx], expr.wfOrderBy, expr.wfOrderDirs) + let cmpRes = compareRowsByOrder(rows[sortedIdxs[pos - 1]], rows[rowIdx], expr.wfOrderBy, expr.wfOrderDirs, ctx) if cmpRes != 0: currentRank = pos + 1 result[rowIdx] = $currentRank @@ -1539,14 +1703,14 @@ proc computeWindowValues*(rows: seq[Row], expr: IRExpr): seq[string] = var currentRank = 1 for pos, rowIdx in sortedIdxs: if pos > 0: - let cmpRes = compareRowsByOrder(rows[sortedIdxs[pos - 1]], rows[rowIdx], expr.wfOrderBy, expr.wfOrderDirs) + let cmpRes = compareRowsByOrder(rows[sortedIdxs[pos - 1]], rows[rowIdx], expr.wfOrderBy, expr.wfOrderDirs, ctx) if cmpRes != 0: currentRank += 1 result[rowIdx] = $currentRank of "ntile": var n = 1 if expr.wfArgs.len > 0: - try: n = parseInt(evalExpr(expr.wfArgs[0], rows[sortedIdxs[0]])) except: n = 1 + try: n = parseInt(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 @@ -1565,36 +1729,36 @@ proc computeWindowValues*(rows: seq[Row], expr: IRExpr): seq[string] = var offset = 1 var defaultVal = "" if expr.wfArgs.len > 1: - try: offset = parseInt(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]])) except: offset = 1 + try: offset = parseInt(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]], ctx)) except: offset = 1 if expr.wfArgs.len > 2: - defaultVal = evalExpr(expr.wfArgs[2], rows[sortedIdxs[0]]) + defaultVal = 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]]) + result[rowIdx] = 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]])) except: offset = 1 + try: offset = parseInt(evalExpr(expr.wfArgs[1], rows[sortedIdxs[0]], ctx)) except: offset = 1 if expr.wfArgs.len > 2: - defaultVal = evalExpr(expr.wfArgs[2], rows[sortedIdxs[0]]) + defaultVal = 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]]) + result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[targetPos]], ctx) else: result[rowIdx] = defaultVal of "first_value": - let firstVal = evalExpr(expr.wfArgs[0], rows[sortedIdxs[0]]) - for rowIdx in sortedIdxs: - result[rowIdx] = firstVal + for pos, rowIdx in sortedIdxs: + let (fStart, fEnd) = resolveFrameBounds(pos, sortedIdxs.len, frameStart, frameEnd) + result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[fStart]], ctx) of "last_value": - let lastVal = evalExpr(expr.wfArgs[0], rows[sortedIdxs[^1]]) - for rowIdx in sortedIdxs: - result[rowIdx] = lastVal + for pos, rowIdx in sortedIdxs: + let (fStart, fEnd) = resolveFrameBounds(pos, sortedIdxs.len, frameStart, frameEnd) + result[rowIdx] = evalExpr(expr.wfArgs[0], rows[sortedIdxs[fEnd]], ctx) else: # Unknown window function — fill with empty for rowIdx in sortedIdxs: @@ -1621,8 +1785,12 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = result.add(row) of irpkProject: - let sourceRows = executePlan(ctx, plan.projectSource) + var sourceRows = executePlan(ctx, plan.projectSource) if plan.projectAliases.len == 0: return sourceRows + # 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]()] # Check if this projection contains aggregates without GROUP BY var hasAggs = false @@ -1660,7 +1828,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if not k.startsWith("$") and not k.contains("."): newRow[k] = v else: - let val = evalExpr(expr, row) + let val = evalExpr(expr, row, ctx) if alias.len > 0: newRow[alias] = val else: newRow["col" & $i] = val if newRow.len > 0: @@ -1686,7 +1854,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if expr.aggFilter != nil: filteredRows = @[] for row in sourceRows: - if evalExpr(expr.aggFilter, row) == "true": + if evalExpr(expr.aggFilter, row, ctx) == "true": filteredRows.add(row) case expr.aggOp of irCount: @@ -1695,49 +1863,49 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = else: var count = 0 for row in filteredRows: - let v = evalExpr(expr.aggArgs[0], row) + let v = evalExpr(expr.aggArgs[0], row, ctx) if 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) + let v = evalExpr(expr.aggArgs[0], row, ctx) try: sum += parseFloat(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) + let v = evalExpr(expr.aggArgs[0], row, ctx) try: sum += parseFloat(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) + let v = evalExpr(expr.aggArgs[0], row, ctx) if minVal == "" or v < minVal: minVal = v newRow[alias] = minVal of irMax: var maxVal = "" for row in filteredRows: - let v = evalExpr(expr.aggArgs[0], row) + let v = evalExpr(expr.aggArgs[0], row, ctx) if maxVal == "" or v > maxVal: maxVal = 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)) + arr.add(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]()) else: "," + let delim = if expr.aggArgs.len > 1: evalExpr(expr.aggArgs[1], initTable[string, string](), ctx) else: "," for row in filteredRows: if expr.aggArgs.len > 0: - parts.add(evalExpr(expr.aggArgs[0], row)) + parts.add(evalExpr(expr.aggArgs[0], row, ctx)) newRow[alias] = parts.join(delim) else: - let val = evalExpr(expr, if sourceRows.len > 0: sourceRows[0] else: initTable[string, string]()) + let val = evalExpr(expr, if sourceRows.len > 0: sourceRows[0] else: initTable[string, string](), ctx) if alias.len > 0: newRow[alias] = val else: newRow["col" & $i] = val result = @[newRow] @@ -1768,7 +1936,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if alias.len > 0: newRow[alias] = "0" else: newRow["col" & $i] = "0" else: - let val = evalExpr(expr, row) + let val = evalExpr(expr, row, ctx) if alias.len > 0: newRow[alias] = val else: newRow["col" & $i] = val if newRow.len > 0: @@ -1782,8 +1950,8 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let sortExpr = plan.sortExprs[0] let ascending = if plan.sortDirs.len > 0: plan.sortDirs[0] else: true proc sortCmp(a, b: Row): int = - let va = evalExpr(sortExpr, a) - let vb = evalExpr(sortExpr, b) + let va = evalExpr(sortExpr, a, ctx) + let vb = evalExpr(sortExpr, b, ctx) try: let fa = parseFloat(va) let fb = parseFloat(vb) @@ -1839,7 +2007,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for row in sourceRows: var groupKey = "" for gk in gkeys: - groupKey &= evalExpr(gk, row) & "|" + groupKey &= evalExpr(gk, row, ctx) & "|" if groupKey notin groups: groups[groupKey] = @[] groups[groupKey].add(row) @@ -1847,7 +2015,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var aggRow: Table[string, string] for gkExpr in gkeys: if gkExpr.kind == irekField and gkExpr.fieldPath.len > 0: - aggRow[gkExpr.fieldPath[^1]] = evalExpr(gkExpr, groupRows[0]) + aggRow[gkExpr.fieldPath[^1]] = evalExpr(gkExpr, groupRows[0], ctx) # Compute each aggregate expression for aggExpr in plan.groupAggs: let aggKey = "$agg_" & $aggExpr.aggOp & "_" & $plan.groupAggs.find(aggExpr) @@ -1855,7 +2023,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if aggExpr.aggFilter != nil: filteredRows = @[] for row in groupRows: - if evalExpr(aggExpr.aggFilter, row) == "true": + if evalExpr(aggExpr.aggFilter, row, ctx) == "true": filteredRows.add(row) case aggExpr.aggOp of irCount: @@ -1864,50 +2032,50 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = else: var count = 0 for row in filteredRows: - let v = evalExpr(aggExpr.aggArgs[0], row) + let v = evalExpr(aggExpr.aggArgs[0], row, ctx) if 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) + let v = evalExpr(aggExpr.aggArgs[0], row, ctx) try: sum += parseFloat(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) + let v = evalExpr(aggExpr.aggArgs[0], row, ctx) try: sum += parseFloat(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) + let v = evalExpr(aggExpr.aggArgs[0], row, ctx) if minVal == "" or v < minVal: minVal = v aggRow[aggKey] = minVal of irMax: var maxVal = "" for row in filteredRows: - let v = evalExpr(aggExpr.aggArgs[0], row) + let v = evalExpr(aggExpr.aggArgs[0], row, ctx) if maxVal == "" or v > maxVal: maxVal = 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)) + arr.add(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]()) else: "," + let delim = if aggExpr.aggArgs.len > 1: evalExpr(aggExpr.aggArgs[1], initTable[string, string](), ctx) else: "," for row in filteredRows: if aggExpr.aggArgs.len > 0: - parts.add(evalExpr(aggExpr.aggArgs[0], row)) + parts.add(evalExpr(aggExpr.aggArgs[0], row, ctx)) aggRow[aggKey] = parts.join(delim) # Apply HAVING filter if plan.groupHaving != nil: - if evalExpr(plan.groupHaving, aggRow) != "true": + if evalExpr(plan.groupHaving, aggRow, ctx) != "true": continue result.add(aggRow) return result @@ -1979,9 +2147,9 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var mergedRows: seq[Row] for r in rawRightRows: let merged = mergeRow(l, r, leftAlias, rightAlias) - if rightFilter != nil and evalExpr(rightFilter, merged) != "true": + if rightFilter != nil and evalExpr(rightFilter, merged, ctx) != "true": continue - if plan.joinCond != nil and evalExpr(plan.joinCond, merged) != "true": + if plan.joinCond != nil and evalExpr(plan.joinCond, merged, ctx) != "true": continue mergedRows.add(merged) @@ -1989,8 +2157,8 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = if rightSortExprs.len > 0 and mergedRows.len > 1: mergedRows.sort(proc(a, b: Row): int = for i, sExpr in rightSortExprs: - let aVal = evalExpr(sExpr, a) - let bVal = evalExpr(sExpr, b) + let aVal = evalExpr(sExpr, a, ctx) + let bVal = evalExpr(sExpr, b, ctx) let asc = if i < rightSortDirs.len: rightSortDirs[i] else: true var cmp = 0 let aNum = parseFloat(aVal) @@ -2060,7 +2228,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var matched = false for r in rightRows: let merged = mergeRow(l, r, leftAlias, rightAlias) - if plan.joinCond == nil or evalExpr(plan.joinCond, merged) == "true": + if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": result.add(merged) matched = true if not matched and (plan.joinKind == irjkLeft or plan.joinKind == irjkFull): @@ -2084,7 +2252,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var found = false for l in leftRows: let merged = mergeRow(l, r, leftAlias, rightAlias) - if plan.joinCond == nil or evalExpr(plan.joinCond, merged) == "true": + if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": found = true break if not found: @@ -2152,32 +2320,32 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = else: var count = 0 for row in matchingRows: - let v = evalExpr(plan.pivotAgg.aggArgs[0], row) + let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) if 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) + let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) try: sum += parseFloat(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) + let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) try: sum += parseFloat(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) + let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) if minVal == "" or v < minVal: minVal = v aggResult = minVal of irMax: var maxVal = "" for row in matchingRows: - let v = evalExpr(plan.pivotAgg.aggArgs[0], row) + let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) if maxVal == "" or v > maxVal: maxVal = v aggResult = maxVal else: discard @@ -2469,15 +2637,15 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] let whereIr = lowerExpr(stmt.selWhere.whereExpr) var tmp: seq[Row] = @[] for row in filteredRows: - if evalExpr(whereIr, row) == "true": + if evalExpr(whereIr, row, ctx) == "true": tmp.add(row) filteredRows = tmp if stmt.selOrderBy.len > 0: let sortExpr = lowerExpr(stmt.selOrderBy[0].orderByExpr) let asc = stmt.selOrderBy[0].orderByDir == sdAsc proc sortCmp(a, b: Row): int = - let va = evalExpr(sortExpr, a) - let vb = evalExpr(sortExpr, b) + let va = evalExpr(sortExpr, a, ctx) + let vb = evalExpr(sortExpr, b, ctx) try: let fa = parseFloat(va) let fb = parseFloat(vb) @@ -2798,7 +2966,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] # Check WHERE if stmt.updWhere != nil and stmt.updWhere.whereExpr != nil: let whereExpr = lowerExpr(stmt.updWhere.whereExpr) - if evalExpr(whereExpr, row) != "true": continue + if evalExpr(whereExpr, row, ctx) != "true": continue # Get key from row if "$key" in row: let old = row["$key"] @@ -2839,7 +3007,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] for row in rows: if stmt.delWhere != nil and stmt.delWhere.whereExpr != nil: let whereExpr = lowerExpr(stmt.delWhere.whereExpr) - if evalExpr(whereExpr, row) != "true": continue + if evalExpr(whereExpr, row, ctx) != "true": continue if "$key" in row: let old = row["$key"] # Fire BEFORE DELETE triggers @@ -2879,7 +3047,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] for k, v in tgtRow: rowWithTarget[stmt.mergeTargetAlias & "." & k] = v let onExpr = lowerExpr(stmt.mergeOn) - if evalExpr(onExpr, rowWithTarget) == "true": + if evalExpr(onExpr, rowWithTarget, ctx) == "true": matched = true if stmt.mergeMatchedUpdate.len > 0 and "$key" in tgtRow: var updateSets = initTable[string, string]() @@ -2887,7 +3055,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] 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) + updateSets[s.binLeft.identName] = evalExpr(valExpr, rowWithTarget, ctx) var newRow = tgtRow for col, val in updateSets: newRow[col] = val @@ -2907,7 +3075,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] if i < stmt.mergeNotMatchedValues.len: let v = stmt.mergeNotMatchedValues[i] let valExpr = lowerExpr(v) - values.add(evalExpr(valExpr, combinedRow)) + values.add(evalExpr(valExpr, combinedRow, ctx)) else: values.add("") if fields.len > 0: @@ -3284,14 +3452,16 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] # Full-text search index var ftsIdx = fts.newInvertedIndex() let rows = execScan(ctx, stmt.ciTarget) - var docId: uint64 = 0 for row in rows: + let lsmKey = if "$key" in row: 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: "" if text.len > 0: ftsIdx.addDocument(docId, text) - let lsmKey = if "$key" in row: row["$key"] else: "" - docId += 1 ctx.ftsIndexes[colKey] = ftsIdx return okResult(msg="CREATE INDEX " & idxName & " on " & stmt.ciTarget & " USING FTS") @@ -3426,6 +3596,10 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] ctx.db.delete(grantKey) return okResult(msg="REVOKE " & stmt.rvPrivilege & " ON " & stmt.rvTable & " FROM " & stmt.rvGrantee) + of nkSetVar: + ctx.sessionVars[stmt.svName] = stmt.svValue + return okResult(msg="SET " & stmt.svName & " = " & stmt.svValue) + else: return errResult("Unsupported statement type: " & $stmt.kind) diff --git a/src/barabadb/query/ir.nim b/src/barabadb/query/ir.nim index 5af20c4..141d492 100644 --- a/src/barabadb/query/ir.nim +++ b/src/barabadb/query/ir.nim @@ -29,6 +29,10 @@ type irIsNull, irIsNotNull irFtsMatch irDistance + irJsonContains + irJsonContainedBy + irJsonHasAny + irJsonHasAll IRAggregate* = enum irCount, irSum, irAvg, irMin, irMax diff --git a/src/barabadb/query/lexer.nim b/src/barabadb/query/lexer.nim index f466fc5..2a04207 100644 --- a/src/barabadb/query/lexer.nim +++ b/src/barabadb/query/lexer.nim @@ -147,6 +147,10 @@ type tkArrowR # -> tkArrowRR # ->> tkFtsMatch # @@ + tkJsonContains # @> + tkJsonContainedBy # <@ + tkJsonHasAny # ?| + tkJsonHasAll # ?& tkSimilar tkNearest tkTo @@ -157,12 +161,15 @@ type # Window functions tkOver tkPartition + tkRow tkRows tkRange tkUnbounded tkPreceding tkFollowing tkCurrent + tkCurrentUser + tkCurrentRole tkRowNumber tkRank tkDenseRank @@ -365,12 +372,15 @@ const keywords*: Table[string, TokenKind] = { "path": tkPath, "over": tkOver, "partition": tkPartition, + "row": tkRow, "rows": tkRows, "range": tkRange, "unbounded": tkUnbounded, "preceding": tkPreceding, "following": tkFollowing, "current": tkCurrent, + "current_user": tkCurrentUser, + "current_role": tkCurrentRole, "row_number": tkRowNumber, "rank": tkRank, "dense_rank": tkDenseRank, @@ -586,6 +596,10 @@ proc nextToken*(l: var Lexer): Token = discard l.advance() discard l.advance() return Token(kind: tkNotEq, value: "<>", line: startLine, col: startCol) + if l.pos + 1 < l.input.len and l.input[l.pos + 1] == '@': + discard l.advance() + discard l.advance() + return Token(kind: tkJsonContainedBy, value: "<@", line: startLine, col: startCol) discard l.advance() return Token(kind: tkLt, value: "<", line: startLine, col: startCol) of '>': @@ -596,6 +610,16 @@ proc nextToken*(l: var Lexer): Token = discard l.advance() return Token(kind: tkGt, value: ">", line: startLine, col: startCol) of '?': + if l.pos + 2 < l.input.len and l.input[l.pos + 1] == '|': + discard l.advance() + discard l.advance() + discard l.advance() + return Token(kind: tkJsonHasAny, value: "?|", line: startLine, col: startCol) + if l.pos + 2 < l.input.len and l.input[l.pos + 1] == '&': + discard l.advance() + discard l.advance() + discard l.advance() + return Token(kind: tkJsonHasAll, value: "?&", line: startLine, col: startCol) if l.pos + 1 < l.input.len and l.input[l.pos + 1] == '?': discard l.advance() discard l.advance() @@ -607,6 +631,10 @@ proc nextToken*(l: var Lexer): Token = discard l.advance() discard l.advance() return Token(kind: tkFtsMatch, value: "@@", line: startLine, col: startCol) + if l.pos + 1 < l.input.len and l.input[l.pos + 1] == '>': + discard l.advance() + discard l.advance() + return Token(kind: tkJsonContains, value: "@>", line: startLine, col: startCol) discard l.advance() return Token(kind: tkInvalid, value: "@", line: startLine, col: startCol) of '.': diff --git a/src/barabadb/query/parser.nim b/src/barabadb/query/parser.nim index 97babec..85c84bb 100644 --- a/src/barabadb/query/parser.nim +++ b/src/barabadb/query/parser.nim @@ -41,6 +41,7 @@ proc parseCreateType(p: var Parser): Node proc parseOverClause(p: var Parser): Node proc parseFrameSpec(p: var Parser): Node proc parseFrameBoundary(p: var Parser): string +proc parseSetVar(p: var Parser): Node proc parsePrimary(p: var Parser): Node = let tok = p.peek() @@ -60,6 +61,12 @@ proc parsePrimary(p: var Parser): Node = of tkNull: discard p.advance() Node(kind: nkNullLit, line: tok.line, col: tok.col) + of tkCurrentUser: + discard p.advance() + Node(kind: nkCurrentUser, line: tok.line, col: tok.col) + of tkCurrentRole: + discard p.advance() + Node(kind: nkCurrentRole, line: tok.line, col: tok.col) of tkIdent, tkRowNumber, tkRank, tkDenseRank, tkLead, tkLag, tkFirstValue, tkLastValue, tkNtile: discard p.advance() let funcName = tok.value @@ -231,7 +238,7 @@ proc parseFrameBoundary(p: var Parser): string = else: raise newException(ValueError, "Expected PRECEDING or FOLLOWING after UNBOUNDED") elif p.match(tkCurrent): - discard p.expect(tkRows) + discard p.expect(tkRow) return "CURRENT ROW" else: # Expect a simple integer literal for offset boundaries @@ -318,7 +325,7 @@ proc parseComparison(p: var Parser): Node = discard p.advance() # consume NULL token (assumed) return Node(kind: nkIsExpr, isExpr: result, isNegated: negated, line: tok.line, col: tok.col) - while p.peek().kind in {tkEq, tkNotEq, tkLt, tkLtEq, tkGt, tkGtEq, tkFtsMatch, tkDistanceOp}: + while p.peek().kind in {tkEq, tkNotEq, tkLt, tkLtEq, tkGt, tkGtEq, tkFtsMatch, tkDistanceOp, tkJsonContains, tkJsonContainedBy, tkJsonHasAny, tkJsonHasAll}: let op = case p.peek().kind of tkEq: bkEq of tkNotEq: bkNotEq @@ -328,6 +335,10 @@ proc parseComparison(p: var Parser): Node = of tkGtEq: bkGtEq of tkFtsMatch: bkFtsMatch of tkDistanceOp: bkDistance + of tkJsonContains: bkJsonContains + of tkJsonContainedBy: bkJsonContainedBy + of tkJsonHasAny: bkJsonHasAny + of tkJsonHasAll: bkJsonHasAll else: bkEq let tok = p.advance() let right = p.parseAddSub() @@ -1056,13 +1067,13 @@ proc parseAlterTable(p: var Parser): Node = # Check for ENABLE/DISABLE ROW LEVEL SECURITY if p.peek().kind == tkEnable: discard p.advance() - discard p.expect(tkIdent) # ROW + discard p.expect(tkRow) # ROW discard p.expect(tkIdent) # LEVEL discard p.expect(tkIdent) # SECURITY return Node(kind: nkEnableRLS, erlsTable: tableName, line: tok.line, col: tok.col) elif p.peek().kind == tkDisable: discard p.advance() - discard p.expect(tkIdent) # ROW + discard p.expect(tkRow) # ROW discard p.expect(tkIdent) # LEVEL discard p.expect(tkIdent) # SECURITY return Node(kind: nkDisableRLS, drlsTable: tableName, line: tok.line, col: tok.col) @@ -1422,6 +1433,35 @@ proc parseRevoke(p: var Parser): Node = result = Node(kind: nkRevoke, rvPrivilege: priv, rvTable: tableName, rvGrantee: grantee, line: tok.line, col: tok.col) +proc parseSetVar(p: var Parser): Node = + let tok = p.expect(tkSet) + var varName = p.expect(tkIdent).value + while p.peek().kind == tkDot: + discard p.advance() + varName.add(".") + varName.add(p.expect(tkIdent).value) + if p.match(tkEq) or p.match(tkTo): + discard + let valTok = p.peek() + var valStr = "" + case valTok.kind + of tkStringLit: + valStr = valTok.value + discard p.advance() + of tkIntLit: + valStr = valTok.value + discard p.advance() + of tkFloatLit: + valStr = valTok.value + discard p.advance() + of tkIdent: + valStr = valTok.value + discard p.advance() + else: + raise newException(ValueError, "Expected value after SET " & varName) + result = Node(kind: nkSetVar, svName: varName, svValue: valStr, + line: tok.line, col: tok.col) + proc parseStatement*(p: var Parser): Node = case p.peek().kind of tkWith, tkSelect: p.parseSelect() @@ -1483,6 +1523,8 @@ proc parseStatement*(p: var Parser): Node = p.parseGrant() of tkRevoke: p.parseRevoke() + of tkSet: + p.parseSetVar() of tkApply: if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkMigration: p.parseApplyMigration() diff --git a/tests/test_all.nim b/tests/test_all.nim index 59ec246..6199b57 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -2287,6 +2287,69 @@ suite "Row-Level Security": check res.success check res.rows.len == 1 # superuser sees all (only 1 row exists) +suite "Session Variables and Multi-Tenant": + test "SET statement parse": + let ast = parse("SET app.tenant_id = 'company-123'") + check ast.stmts.len == 1 + check ast.stmts[0].kind == nkSetVar + check ast.stmts[0].svName == "app.tenant_id" + check ast.stmts[0].svValue == "company-123" + + test "SET and current_setting": + let tmpDir = getTempDir() / "baradb_session_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) + var ctx = qexec.newExecutionContext(db) + discard qexec.executeQuery(ctx, parse("SET app.tenant_id = 'company-123'")) + let r = qexec.executeQuery(ctx, parse("SELECT current_setting('app.tenant_id') AS tenant")) + check r.success + check r.rows.len == 1 + check r.rows[0]["tenant"] == "company-123" + + test "current_user in SELECT": + let tmpDir = getTempDir() / "baradb_user_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) + var ctx = qexec.newExecutionContext(db) + ctx.currentUser = "alice" + let r = qexec.executeQuery(ctx, parse("SELECT current_user AS me")) + check r.success + check r.rows.len == 1 + check r.rows[0]["me"] == "alice" + + test "current_role in SELECT": + let tmpDir = getTempDir() / "baradb_role_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) + var ctx = qexec.newExecutionContext(db) + ctx.currentRole = "admin" + let r = qexec.executeQuery(ctx, parse("SELECT current_role AS role")) + check r.success + check r.rows.len == 1 + check r.rows[0]["role"] == "admin" + + test "Multi-tenant RLS with current_setting": + let tmpDir = getTempDir() / "baradb_multitenant_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) + var ctx = qexec.newExecutionContext(db) + discard qexec.executeQuery(ctx, parse("CREATE TABLE invoices (id INTEGER, tenant_id TEXT, amount INT)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO invoices (id, tenant_id, amount) VALUES (1, 'company-a', 100)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO invoices (id, tenant_id, amount) VALUES (2, 'company-b', 200)")) + # Set session variable for tenant + discard qexec.executeQuery(ctx, parse("SET app.tenant_id = 'company-a'")) + # Create policy that uses current_setting + ctx.users["app"] = qexec.UserDef(name: "app", passwordHash: "", isSuperuser: false, roles: @[]) + ctx.currentUser = "app" + ctx.policies["invoices"] = @[ + qexec.PolicyDef(name: "tenant_isolation", tableName: "invoices", command: "SELECT", + usingExpr: Node(kind: nkBinOp, binOp: bkEq, + binLeft: Node(kind: nkIdent, identName: "tenant_id"), + binRight: Node(kind: nkFuncCall, funcName: "current_setting", + funcArgs: @[Node(kind: nkStringLit, strVal: "app.tenant_id")])), + withCheckExpr: nil) + ] + let r = qexec.executeQuery(ctx, parse("SELECT id, tenant_id FROM invoices")) + check r.success + check r.rows.len == 1 + check r.rows[0]["tenant_id"] == "company-a" + suite "UTF-8 Support": test "Tokenize UTF-8 identifiers": let tokens = lex.tokenize("SELECT имя FROM потребители") @@ -2303,7 +2366,8 @@ suite "UTF-8 Support": check ast.stmts[0].selWhere.whereExpr.binRight.strVal == "София" test "Execute query with UTF-8 data": - var db = newLSMTree("") + let tmpDir = getTempDir() / "baradb_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) var ctx = qexec.newExecutionContext(db) discard qexec.executeQuery(ctx, parse("CREATE TABLE потребители (имя TEXT, град TEXT)")) discard qexec.executeQuery(ctx, parse("INSERT INTO потребители (имя, град) VALUES ('Иван', 'София'), ('Мария', 'Пловдив')")) @@ -2315,7 +2379,8 @@ suite "UTF-8 Support": suite "B-Tree Range Scan": test "BETWEEN uses index range scan": - var db = newLSMTree("") + let tmpDir = getTempDir() / "baradb_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) var ctx = qexec.newExecutionContext(db) discard qexec.executeQuery(ctx, parse("CREATE TABLE products (id INTEGER, name TEXT)")) discard qexec.executeQuery(ctx, parse("INSERT INTO products (id, name) VALUES (1, 'apple'), (2, 'banana'), (3, 'cherry'), (4, 'date'), (5, 'elderberry')")) @@ -2325,7 +2390,8 @@ suite "B-Tree Range Scan": check res.rows.len == 3 test "Greater than uses index range scan": - var db = newLSMTree("") + let tmpDir = getTempDir() / "baradb_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) var ctx = qexec.newExecutionContext(db) discard qexec.executeQuery(ctx, parse("CREATE TABLE nums (id INTEGER, val TEXT)")) discard qexec.executeQuery(ctx, parse("INSERT INTO nums (id, val) VALUES (1, '10'), (2, '20'), (3, '30'), (4, '40'), (5, '50')")) @@ -2335,7 +2401,8 @@ suite "B-Tree Range Scan": check res.rows.len == 3 test "Less than or equal uses index range scan": - var db = newLSMTree("") + let tmpDir = getTempDir() / "baradb_test_" & $getMonoTime().ticks + var db = newLSMTree(tmpDir) var ctx = qexec.newExecutionContext(db) discard qexec.executeQuery(ctx, parse("CREATE TABLE nums2 (id INTEGER, val TEXT)")) discard qexec.executeQuery(ctx, parse("INSERT INTO nums2 (id, val) VALUES (1, '10'), (2, '20'), (3, '30'), (4, '40'), (5, '50')")) @@ -2458,13 +2525,18 @@ suite "Enhanced Migrations": suite "Parameterized queries": var db: LSMTree var ctx: qexec.ExecutionContext + var tmpDir: string + var paramInitialized = false setup: - db = newLSMTree("") - ctx = qexec.newExecutionContext(db) - discard qexec.executeQuery(ctx, parse("CREATE TABLE users (id INT, name TEXT, age INT)")) - discard qexec.executeQuery(ctx, parse("INSERT INTO users (id, name, age) VALUES (1, 'Alice', 30)")) - discard qexec.executeQuery(ctx, parse("INSERT INTO users (id, name, age) VALUES (2, 'Bob', 25)")) + if not paramInitialized: + tmpDir = getTempDir() / "baradb_param_test_" & $getMonoTime().ticks + db = newLSMTree(tmpDir) + ctx = qexec.newExecutionContext(db) + discard qexec.executeQuery(ctx, parse("CREATE TABLE users (id INT, name TEXT, age INT, active BOOLEAN)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO users (id, name, age, active) VALUES (1, 'Alice', 30, true)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO users (id, name, age, active) VALUES (2, 'Bob', 25, false)")) + paramInitialized = true test "SELECT with placeholder params": let sql = "SELECT * FROM users WHERE id = ?" @@ -2583,6 +2655,34 @@ suite "Parameterized queries": check r.success check r.rows.len >= 1 + test "JSON contains @>": + discard qexec.executeQuery(ctx, parse("CREATE TABLE IF NOT EXISTS jsontest2 (id INT PRIMARY KEY, data JSON)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO jsontest2 (id, data) VALUES (1, '{\"name\": \"Alice\", \"age\": 30}')")) + let r = qexec.executeQuery(ctx, parse("SELECT id FROM jsontest2 WHERE data @> '{\"name\": \"Alice\"}'")) + check r.success + check r.rows.len == 1 + check r.rows[0]["id"] == "1" + + test "JSON contained by <@": + let r = qexec.executeQuery(ctx, parse("SELECT id FROM jsontest2 WHERE '{\"name\": \"Alice\"}' <@ data")) + check r.success + check r.rows.len == 1 + + test "JSON has key json_has_key": + let r = qexec.executeQuery(ctx, parse("SELECT id FROM jsontest2 WHERE json_has_key(data, 'name') = 'true'")) + check r.success + check r.rows.len == 1 + + test "JSON has any key ?|": + let r = qexec.executeQuery(ctx, parse("SELECT id FROM jsontest2 WHERE data ?| '[\"name\", \"missing\"]'")) + check r.success + check r.rows.len == 1 + + test "JSON has all keys ?&": + let r = qexec.executeQuery(ctx, parse("SELECT id FROM jsontest2 WHERE data ?& '[\"name\", \"age\"]'")) + check r.success + check r.rows.len == 1 + test "FTS match operator @@ parse": let ast = parse("SELECT * FROM docs WHERE content @@ 'hello'") check ast.stmts[0].kind == nkSelect @@ -2630,9 +2730,11 @@ suite "Parameterized queries": suite "Window Functions": var db: LSMTree var ctx: qexec.ExecutionContext + var tmpDir: string setup: - db = newLSMTree("") + tmpDir = getTempDir() / "baradb_window_test_" & $getMonoTime().ticks + db = newLSMTree(tmpDir) ctx = qexec.newExecutionContext(db) discard qexec.executeQuery(ctx, parse("CREATE TABLE employees (id INT, name TEXT, department TEXT, salary INT)")) discard qexec.executeQuery(ctx, parse("INSERT INTO employees (id, name, department, salary) VALUES (1, 'Alice', 'Engineering', 90000)")) @@ -2641,6 +2743,9 @@ suite "Window Functions": discard qexec.executeQuery(ctx, parse("INSERT INTO employees (id, name, department, salary) VALUES (4, 'Diana', 'Sales', 75000)")) discard qexec.executeQuery(ctx, parse("INSERT INTO employees (id, name, department, salary) VALUES (5, 'Eve', 'Engineering', 95000)")) + teardown: + removeDir(tmpDir) + test "ROW_NUMBER with PARTITION BY and ORDER BY": let r = qexec.executeQuery(ctx, parse("SELECT name, department, ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) AS rn FROM employees")) check r.success @@ -2692,12 +2797,38 @@ suite "Window Functions": for row in r.rows: check row["bucket"] in @["1", "2"] + test "FIRST_VALUE with frame": + let r = qexec.executeQuery(ctx, parse("SELECT name, salary, FIRST_VALUE(salary) OVER (ORDER BY salary ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) AS first_sal FROM employees")) + check r.success + check r.rows.len == 5 + for row in r.rows: + if row["name"] == "Charlie": + check row["first_sal"] == "70000" + if row["name"] == "Diana": + check row["first_sal"] == "70000" + if row["name"] == "Bob": + check row["first_sal"] == "75000" + + test "LAST_VALUE with frame": + let r = qexec.executeQuery(ctx, parse("SELECT name, salary, LAST_VALUE(salary) OVER (ORDER BY salary ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING) AS last_sal FROM employees")) + check r.success + check r.rows.len == 5 + for row in r.rows: + if row["name"] == "Charlie": + check row["last_sal"] == "75000" + if row["name"] == "Diana": + check row["last_sal"] == "80000" + if row["name"] == "Bob": + check row["last_sal"] == "90000" + suite "GROUP BY Aggregates": var db: LSMTree var ctx: qexec.ExecutionContext + var tmpDir: string setup: - db = newLSMTree("") + tmpDir = getTempDir() / "baradb_group_test_" & $getMonoTime().ticks + db = newLSMTree(tmpDir) ctx = qexec.newExecutionContext(db) discard qexec.executeQuery(ctx, parse("CREATE TABLE sales (id INT, dept TEXT, amount INT)")) discard qexec.executeQuery(ctx, parse("INSERT INTO sales (id, dept, amount) VALUES (1, 'A', 100)")) @@ -2706,6 +2837,9 @@ suite "GROUP BY Aggregates": discard qexec.executeQuery(ctx, parse("INSERT INTO sales (id, dept, amount) VALUES (4, 'B', 50)")) discard qexec.executeQuery(ctx, parse("INSERT INTO sales (id, dept, amount) VALUES (5, 'C', 300)")) + teardown: + removeDir(tmpDir) + test "GROUP BY with COUNT(*)": let r = qexec.executeQuery(ctx, parse("SELECT dept, COUNT(*) AS cnt FROM sales GROUP BY dept")) check r.success @@ -2808,6 +2942,26 @@ suite "GROUP BY Aggregates": check r.success check r.rows.len == 4 # one row per employee + test "CUBE execution": + let r = qexec.executeQuery(ctx, parse("SELECT dept, SUM(amount) AS total FROM sales GROUP BY CUBE (dept)")) + check r.success + # 3 dept groups + 1 grand total = 4 rows + check r.rows.len == 4 + + test "GROUPING SETS execution": + let r = qexec.executeQuery(ctx, parse("SELECT dept, SUM(amount) AS total FROM sales GROUP BY GROUPING SETS ((dept), ())")) + check r.success + # 3 dept groups + 1 grand total = 4 rows + check r.rows.len == 4 + + test "UNPIVOT execution": + discard qexec.executeQuery(ctx, parse("CREATE TABLE pivoted (name TEXT, eng_salary INT, sales_salary INT)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO pivoted (name, eng_salary, sales_salary) VALUES ('Alice', 90000, 0)")) + discard qexec.executeQuery(ctx, parse("INSERT INTO pivoted (name, eng_salary, sales_salary) VALUES ('Bob', 0, 70000)")) + let r = qexec.executeQuery(ctx, parse("SELECT * FROM pivoted UNPIVOT (salary FOR dept IN (eng_salary, sales_salary))")) + check r.success + check r.rows.len == 4 + # JOIN tests include "join_tests"