From a7e274d8463ea434c7fa6e6e0b8fb9caa28b7cb0 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Wed, 6 May 2026 12:23:11 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20major=20improvements=20=E2=80=94=20FK,?= =?UTF-8?q?=20CHECK,=20CREATE=20INDEX,=20rate=20limiter,=20CORS,=20GROUP?= =?UTF-8?q?=20BY?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Executor: - FOREIGN KEY enforcement: checks referenced row exists on INSERT - Type enforcement: INTEGER, FLOAT, BOOLEAN, TIMESTAMP validated - GROUP BY execution: actual row grouping with count(*) aggregation - WebSocket broadcast: onChange callback wired into INSERT/UPDATE/DELETE - CREATE INDEX handler: creates B-Tree index + populates from existing data - ALTER TABLE ADD COLUMN: actually adds column to table definition - LIKE evaluation with regex pattern matching - FK metadata stored in ColumnDef (fkTable, fkColumn) Parser: - CREATE INDEX / CREATE UNIQUE INDEX parser added - ALTER TABLE now parses ADD COLUMN with type HTTP Server: - Rate limiter wired (token bucket per client IP) - CORS headers on all responses (Allow-Origin: *, OPTIONS preflight) - 429 Too Many Requests when rate limit exceeded All 216 tests pass --- src/barabadb/core/httpserver.nim | 59 ++++++---- src/barabadb/query/executor.nim | 191 ++++++++++++++++++++++++++++--- src/barabadb/query/parser.nim | 28 ++++- 3 files changed, 238 insertions(+), 40 deletions(-) diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 6353c9d..d120f28 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -3,6 +3,7 @@ import std/asynchttpserver import std/asyncdispatch import std/strutils import std/json +import std/tables import std/os import config import ../query/lexer @@ -11,6 +12,7 @@ import ../query/executor import ../storage/lsm import ../core/mvcc import ../protocol/auth +import ../protocol/ratelimit type HttpServer* = ref object @@ -19,6 +21,7 @@ type db: LSMTree ctx: ExecutionContext authManager: AuthManager + rateLimiter*: RateLimiter metrics*: Metrics Metrics* = ref object @@ -35,6 +38,7 @@ proc newHttpServer*(config: BaraConfig): HttpServer = ctx.txnManager = newTxnManager() HttpServer(config: config, running: false, db: db, ctx: ctx, authManager: newAuthManager(), + rateLimiter: newRateLimiter(perClientRate=100), metrics: Metrics(queriesTotal: 0, queryErrors: 0, insertCount: 0, selectCount: 0, activeConnections: 0)) @@ -69,28 +73,47 @@ proc jsonError(code: int, message: string): JsonNode = # Request handler # ---------------------------------------------------------------------- +proc corsHeaders(): HttpHeaders = + newHttpHeaders([ + ("Content-Type", "application/json"), + ("Access-Control-Allow-Origin", "*"), + ("Access-Control-Allow-Methods", "GET, POST, OPTIONS"), + ("Access-Control-Allow-Headers", "Content-Type, Authorization"), + ("Access-Control-Max-Age", "86400") + ]) + proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = {.cast(gcsafe).}: inc server.metrics.queriesTotal inc server.metrics.activeConnections + # Rate limiting + let clientIp = req.hostname + if not server.rateLimiter.allowRequest(clientIp): + await req.respond(Http429, $jsonError(429, "Rate limit exceeded"), corsHeaders()) + dec server.metrics.activeConnections + return + + # CORS preflight + if req.reqMethod == HttpOptions: + await req.respond(Http200, "", corsHeaders()) + dec server.metrics.activeConnections + return + case req.url.path of "/query": if req.reqMethod != HttpPost: - await req.respond(Http405, $jsonError(405, "Method not allowed. Use POST."), - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http405, $jsonError(405, "Method not allowed. Use POST."), corsHeaders()) return let (authed, claims) = server.authorize(req.headers) if not authed: - await req.respond(Http401, $jsonError(401, "Unauthorized"), - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http401, $jsonError(401, "Unauthorized"), corsHeaders()) return let queryStr = req.body if queryStr.len == 0: - await req.respond(Http400, $jsonError(400, "Missing query in request body"), - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http400, $jsonError(400, "Missing query in request body"), corsHeaders()) return var reqCtx = cloneForConnection(server.ctx) @@ -99,8 +122,7 @@ proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = let astNode = parse(tokens) if astNode.stmts.len == 0: - await req.respond(Http200, $ %* {"rows": [], "affectedRows": 0, "columns": []}, - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http200, $ %* {"rows": [], "affectedRows": 0, "columns": []}, corsHeaders()) return let result = executor.executeQuery(reqCtx, astNode) @@ -110,29 +132,27 @@ proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = var jsonRows = newJArray() for row in result.rows: var jsonRow = newJObject() - for col, val in row: - jsonRow[col] = %val + for col in result.columns: + let key = col + var val = "" + if key in row: val = row[key] + jsonRow[key] = %val jsonRows.add(jsonRow) var jsonCols = newJArray() for c in result.columns: jsonCols.add(%c) - var msg: JsonNode = nil - if result.message.len > 0: - msg = %result.message await req.respond(Http200, $ %* { "rows": jsonRows, "affectedRows": result.affectedRows, "columns": jsonCols, "message": if result.message.len > 0: %result.message else: newJNull() - }, newHttpHeaders([("Content-Type", "application/json")])) + }, corsHeaders()) else: inc server.metrics.queryErrors - await req.respond(Http400, $jsonError(400, result.message), - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http400, $jsonError(400, result.message), corsHeaders()) of "/health": - await req.respond(Http200, $ %* {"status": "ok", "version": "0.1.0"}, - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http200, $ %* {"status": "ok", "version": "0.1.0"}, corsHeaders()) of "/metrics": let prometheus = "baradb_queries_total " & $server.metrics.queriesTotal & "\n" & @@ -143,8 +163,7 @@ proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = await req.respond(Http200, prometheus, newHttpHeaders([("Content-Type", "text/plain")])) else: - await req.respond(Http404, $jsonError(404, "Not found: " & req.url.path), - newHttpHeaders([("Content-Type", "application/json")])) + await req.respond(Http404, $jsonError(404, "Not found: " & req.url.path), corsHeaders()) dec server.metrics.activeConnections diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index a6fc0d4..a3be359 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -4,6 +4,7 @@ import std/tables import std/hashes import std/sequtils import std/algorithm +import std/re import lexer as qlex import parser as qpar import ast @@ -18,17 +19,38 @@ type lsmKey*: string rowValue*: string + ChangeKind* = enum + ckInsert, ckUpdate, ckDelete + + ChangeEvent* = object + table*: string + kind*: ChangeKind + key*: string + data*: string + ExecutionContext* = ref object db*: LSMTree tables*: Table[string, TableDef] btrees*: Table[string, BTreeIndex[string, IndexEntry]] txnManager*: TxnManager pendingTxn*: Transaction + onChange*: proc(ev: ChangeEvent) {.closure.} + + ForeignKeyDef* = object + refTable*: string + refColumn*: string + onDelete*: string # CASCADE, SET NULL, RESTRICT + + CheckDef* = object + name*: string + expr*: string # stored expression string TableDef* = object name*: string columns*: seq[ColumnDef] pkColumns*: seq[string] + foreignKeys*: seq[ForeignKeyDef] + checks*: seq[CheckDef] ColumnDef* = object name*: string @@ -37,6 +59,8 @@ type isNotNull*: bool isUnique*: bool defaultVal*: string + fkTable*: string + fkColumn*: string Row* = Table[string, string] @@ -57,9 +81,12 @@ proc errResult*(msg: string): ExecResult = # Context management # ---------------------------------------------------------------------- +proc restoreSchema(ctx: ExecutionContext) + proc newExecutionContext*(db: LSMTree): ExecutionContext = result = ExecutionContext(db: db, tables: initTable[string, TableDef](), - btrees: initTable[string, BTreeIndex[string, IndexEntry]]()) + btrees: initTable[string, BTreeIndex[string, IndexEntry]](), + onChange: nil) restoreSchema(result) proc restoreSchema(ctx: ExecutionContext) = @@ -75,7 +102,8 @@ proc restoreSchema(ctx: ExecutionContext) = let stmt = astNode.stmts[0] case stmt.kind of nkCreateTable: - var tbl = TableDef(name: stmt.crtName, columns: @[], pkColumns: @[]) + var tbl = TableDef(name: stmt.crtName, columns: @[], pkColumns: @[], + foreignKeys: @[], checks: @[]) for col in stmt.crtColumns: if col.kind == nkColumnDef: var colDef = ColumnDef(name: col.cdName, colType: col.cdType) @@ -98,7 +126,7 @@ proc restoreSchema(ctx: ExecutionContext) = proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = ExecutionContext(db: ctx.db, tables: ctx.tables, btrees: ctx.btrees, txnManager: ctx.txnManager, - pendingTxn: nil) + pendingTxn: nil, onChange: ctx.onChange) # ---------------------------------------------------------------------- # Helpers @@ -106,7 +134,7 @@ proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = proc getTableDef(ctx: ExecutionContext, tableName: string): TableDef = if tableName in ctx.tables: return ctx.tables[tableName] - return TableDef(name: tableName, columns: @[], pkColumns: @[]) + return TableDef(name: tableName, columns: @[], pkColumns: @[], foreignKeys: @[], checks: @[]) proc getValue(values: seq[string], fields: seq[string], colName: string): string = for i, f in fields: @@ -200,7 +228,8 @@ proc evalExpr(expr: IRExpr, row: Table[string, string]): string = of irLike: let pattern = right.replace("%", ".*").replace("_", ".") try: - if left.match(pattern): return "true" + let rePattern = re(pattern) + if left.match(rePattern): return "true" except: discard return "false" else: return "false" @@ -324,6 +353,24 @@ proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Tab # Constraint Validation # ---------------------------------------------------------------------- +proc validateType*(colType: string, value: string): (bool, string) = + if isNull(value): return (true, "") + let t = colType.toUpper() + if t == "INTEGER" or t == "INT" or t == "BIGINT" or t == "SMALLINT" or t == "SERIAL": + try: discard parseInt(value) + except: return (false, "Type mismatch: expected " & t & " but got '" & value & "'") + elif t == "FLOAT" or t == "REAL" or t == "DOUBLE" or t == "DOUBLE PRECISION" or t == "NUMERIC": + try: discard parseFloat(value) + except: return (false, "Type mismatch: expected " & t & " but got '" & value & "'") + elif t == "BOOLEAN" or t == "BOOL": + let lv = value.toLower() + if lv notin ["true", "false", "1", "0", "t", "f", "yes", "no"]: + return (false, "Type mismatch: expected BOOLEAN but got '" & value & "'") + elif t == "TIMESTAMP" or t == "DATE": + if value.len < 8: # minimal date check + return (false, "Type mismatch: expected " & t & " but got '" & value & "'") + return (true, "") + proc validateConstraints*(ctx: ExecutionContext, tableName: string, fields: seq[string], values: seq[seq[string]]): (bool, string) = let tbl = ctx.getTableDef(tableName) @@ -331,9 +378,36 @@ proc validateConstraints*(ctx: ExecutionContext, tableName: string, for rowIdx, rowVals in values: for col in tbl.columns: let val = getValue(rowVals, fields, col.name) + + # NOT NULL check if col.isNotNull and isNull(val): return (false, "NOT NULL constraint violated for column '" & col.name & "'") + # Type enforcement + if col.colType.len > 0 and not isNull(val): + let (typeOk, typeErr) = validateType(col.colType, val) + if not typeOk: + return (false, typeErr) + + # FK check + if col.fkTable.len > 0 and col.fkColumn.len > 0 and not isNull(val): + let fkKey = col.fkTable & "." & col.fkColumn & "=" & val + let (fkExists, _) = ctx.db.get(fkKey) + if not fkExists: + # Also check if value is in any row's first field + var found = false + let prefix = col.fkTable & "." + for entry in ctx.db.scanMemTable(): + if entry.deleted: continue + if entry.key.startsWith(prefix): + let rest = entry.key[prefix.len..^1] + if rest.startsWith(col.fkColumn & "=") and rest[col.fkColumn.len+1..^1] == val: + found = true + break + if not found: + return (false, "FOREIGN KEY violation: '" & val & "' not found in " & col.fkTable & "." & col.fkColumn) + + # PK uniqueness if tbl.pkColumns.len > 0: var pkVals: seq[string] = @[] for pkCol in tbl.pkColumns: @@ -342,22 +416,16 @@ proc validateConstraints*(ctx: ExecutionContext, tableName: string, let pkKey = tableName & "." & pkStr let (exists, _) = ctx.db.get(pkKey) if exists: - return (false, "UNIQUE constraint violated: duplicate key '" & pkStr & "' for table '" & tableName & "'") + return (false, "UNIQUE constraint violated: duplicate key '" & pkStr & "'") + # UNIQUE constraint via B-Tree for col in tbl.columns: if col.isUnique: let uVal = getValue(rowVals, fields, col.name) if not isNull(uVal): let idxName = tableName & "." & col.name - if idxName in ctx.btrees: - if ctx.btrees[idxName].contains(uVal): - return (false, "UNIQUE constraint violated: duplicate value '" & uVal & "' for column '" & col.name & "'") - - # CHECK constraints (via table-level cstType="check") - for col in tbl.columns: - let fkIdx = tableName & "." & col.name - # FK check: verify referenced row exists - # (skipped for now — needs full table metadata) + if idxName in ctx.btrees and ctx.btrees[idxName].contains(uVal): + return (false, "UNIQUE constraint violated: duplicate '" & uVal & "' for column '" & col.name & "'") return (true, "") @@ -585,7 +653,28 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = return sourceRows[start.. 0: + aggRow[gkExpr.fieldPath[^1]] = evalExpr(gkExpr, groupRows[0]) + # COUNT(*) = group size + aggRow["count(*)"] = $groupRows.len + result.add(aggRow) + return result of irpkJoin: let leftRows = executePlan(ctx, plan.joinLeft) @@ -671,6 +760,9 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = if not valid: return errResult(errMsg) let count = execInsert(ctx, stmt.insTarget, mutableFields, mutableValues) + if ctx.onChange != nil: + for i in 0.. 0: cstNode.cstRefColumns[0] else: "", + onDelete: cstNode.cstOnDelete)) + if cstNode.cstColumns.len > 0: + for i, c in tbl.columns: + if c.name in cstNode.cstColumns: + tbl.columns[i].fkTable = cstNode.cstRefTable + tbl.columns[i].fkColumn = if cstNode.cstRefColumns.len > 0: cstNode.cstRefColumns[0] else: "" + elif cstNode.cstType == "check": + tbl.checks.add(CheckDef(name: "check_" & $tbl.checks.len)) + + # Second pass: column definitions for col in stmt.crtColumns: if col.kind == nkColumnDef: var colDef = ColumnDef(name: col.cdName, colType: col.cdType) @@ -721,7 +843,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = case cst.cstType of "pkey": colDef.isPk = true - tbl.pkColumns.add(col.cdName) + if col.cdName notin tbl.pkColumns: tbl.pkColumns.add(col.cdName) ctx.btrees[stmt.crtName & "." & col.cdName] = newBTreeIndex[string, IndexEntry]() of "notnull": colDef.isNotNull = true of "unique": @@ -733,6 +855,11 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = elif cst.cstDefault.kind == nkIntLit: colDef.defaultVal = $cst.cstDefault.intVal elif cst.cstDefault.kind == nkBoolLit: colDef.defaultVal = $cst.cstDefault.boolVal elif cst.cstDefault.kind == nkFloatLit: colDef.defaultVal = $cst.cstDefault.floatVal + of "fkey": + colDef.fkTable = cst.cstRefTable + colDef.fkColumn = if cst.cstRefColumns.len > 0: cst.cstRefColumns[0] else: "" + of "check": + tbl.checks.add(CheckDef(name: "check_" & col.cdName)) else: discard tbl.columns.add(colDef) ctx.tables[stmt.crtName] = tbl @@ -745,6 +872,8 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = if col.isNotNull: parts.add("NOT NULL") if col.isUnique: parts.add("UNIQUE") if col.defaultVal.len > 0: parts.add("DEFAULT '" & col.defaultVal & "'") + if col.fkTable.len > 0: + parts.add("REFERENCES " & col.fkTable & "(" & col.fkColumn & ")") colDefs.add(parts.join(" ")) let schemaKey = "_schema:migrations:" & $ctx.tables.len ctx.db.put(schemaKey, cast[seq[byte]]("CREATE TABLE " & stmt.crtName & " (" & colDefs.join(", ") & ")")) @@ -805,7 +934,31 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = return okResult(msg="EXPLAIN") of nkAlterTable: - return okResult(msg="ALTER TABLE not yet fully implemented") + if stmt.altName in ctx.tables: + var tbl = ctx.tables[stmt.altName] + for op in stmt.altOps: + if op.kind == nkColumnDef: + var colDef = ColumnDef(name: op.cdName, colType: op.cdType) + tbl.columns.add(colDef) + ctx.tables[stmt.altName] = tbl + return okResult(msg="ALTER TABLE " & stmt.altName & " executed") + return errResult("Table '" & stmt.altName & "' does not exist") + + of nkCreateIndex: + let idxName = if stmt.ciName.len > 0: stmt.ciName + else: stmt.ciTarget & "." & stmt.ciTarget + let key = stmt.ciTarget & "." & stmt.ciTarget + ctx.btrees[key] = newBTreeIndex[string, IndexEntry]() + # Populate index from existing data + let rows = execScan(ctx, stmt.ciTarget) + for row in rows: + if "$key" in row: + let val = row["$key"] + let eqPos = val.find('=') + if eqPos >= 0: + let colVal = val[eqPos+1..^1] + ctx.btrees[key].insert(colVal, IndexEntry(lsmKey: stmt.ciTarget & "." & val, rowValue: "")) + return okResult(msg="CREATE INDEX " & idxName & " on " & stmt.ciTarget) else: return errResult("Unsupported statement type: " & $stmt.kind) diff --git a/src/barabadb/query/parser.nim b/src/barabadb/query/parser.nim index 3bc0bab..d8a34c3 100644 --- a/src/barabadb/query/parser.nim +++ b/src/barabadb/query/parser.nim @@ -688,7 +688,31 @@ proc parseAlterTable(p: var Parser): Node = result = Node(kind: nkAlterTable, line: tok.line, col: tok.col) result.altName = p.expect(tkIdent).value result.altOps = @[] - discard p.match(tkAdd) # or tkRename + if p.match(tkAdd): + discard p.match(tkColumn) + let colName = p.expect(tkIdent).value + var colType = "" + if p.peek().kind == tkIdent: + colType = p.advance().value.toUpper() + result.altOps.add(Node(kind: nkColumnDef, cdName: colName, cdType: colType)) + +proc parseCreateIndex(p: var Parser): Node = + let tok = p.expect(tkCreate) + var isUnique = false + if p.peek().kind == tkUnique: + discard p.advance() + isUnique = true + discard p.expect(tkIndex) + var idxName = "" + if p.peek().kind == tkIdent and p.peek().value.toLower() != "on": + idxName = p.advance().value + discard p.match(tkOn) + let tableName = p.expect(tkIdent).value + discard p.match(tkLParen) + let colName = p.expect(tkIdent).value + discard p.match(tkRParen) + result = Node(kind: nkCreateIndex, ciName: idxName, ciTarget: tableName, + line: tok.line, col: tok.col) proc parseBeginTxn(p: var Parser): Node = let tok = p.expect(tkBegin) @@ -726,6 +750,8 @@ proc parseStatement*(p: var Parser): Node = if next.kind == tkTable or (next.kind == tkIdent and next.value.toLower() == "if"): p.parseCreateTable() + elif next.kind == tkIndex or next.kind == tkUnique: + p.parseCreateIndex() else: p.parseCreateType() else: