diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index bb41739..97374a4 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -190,12 +190,97 @@ proc openApiHandler(): RequestHandler = } }) -# ---------------------------------------------------------------------- -# Server lifecycle -# ---------------------------------------------------------------------- +proc adminHandler(server: HttpServer): RequestHandler = + return proc(request: Request) {.gcsafe.} = + let html = """ + + +BaraDB Admin + +

BaraDB Admin

+
+ SQL Playground + Schema + Metrics +
+
+
+ + + +
+
+
+ + + +
BaraDB v0.1.0 — Production-ready multimodal database
+""" + request.respond(200, @[("Content-Type", "text/html")], html) proc run*(server: HttpServer, port: int = 8080) = var router = newRouter() + router.get("/admin", server.adminHandler()) + router.get("/", server.adminHandler()) router.post("/query", server.queryHandler()) router.get("/health", healthHandler()) router.get("/metrics", server.metricsHandler()) diff --git a/src/barabadb/core/logging.nim b/src/barabadb/core/logging.nim new file mode 100644 index 0000000..5e47c47 --- /dev/null +++ b/src/barabadb/core/logging.nim @@ -0,0 +1,49 @@ +## BaraDB Structured JSON Logger +import std/json +import std/times +import std/os + +type + LogLevel* = enum + llDebug = 0 + llInfo = 1 + llWarn = 2 + llError = 3 + + Logger* = ref object + level*: LogLevel + output*: File + +var defaultLogger* = Logger(level: llInfo, output: stdout) + +proc newLogger*(level: LogLevel = llInfo, filepath: string = ""): Logger = + var f = stdout + if filepath.len > 0: + f = open(filepath, fmAppend) + Logger(level: level, output: f) + +proc log*(logger: Logger, level: LogLevel, msg: string, extra: JsonNode = newJNull()) = + if level < logger.level: return + let entry = %*{ + "ts": $now(), + "level": $level, + "msg": msg, + "extra": extra + } + logger.output.writeLine($entry) + logger.output.flushFile() + +proc log*(msg: string, level: LogLevel = llInfo) = + defaultLogger.log(level, msg) + +proc debug*(msg: string) = defaultLogger.log(llDebug, msg) +proc info*(msg: string) = defaultLogger.log(llInfo, msg) +proc warn*(msg: string) = defaultLogger.log(llWarn, msg) +proc errorMsg*(msg: string) = defaultLogger.log(llError, msg) + +proc setLevel*(logger: Logger, level: LogLevel) = + logger.level = level + +proc close*(logger: Logger) = + if logger.output != stdout: + logger.output.close() diff --git a/src/barabadb/core/mvcc.nim b/src/barabadb/core/mvcc.nim index feab5aa..ad36625 100644 --- a/src/barabadb/core/mvcc.nim +++ b/src/barabadb/core/mvcc.nim @@ -42,14 +42,15 @@ type nextLsn: uint64 activeTxns: Table[TxnId, Transaction] committedTxns: seq[TxnId] - globalVersions*: Table[string, seq[VersionedRecord]] # key -> versions + globalVersions*: Table[string, seq[VersionedRecord]] + txnTimeoutMs: int64 proc `==`*(a, b: TxnId): bool {.borrow.} proc `==`*(a, b: Lsn): bool {.borrow.} proc `$`*(id: TxnId): string = $uint64(id) proc `$`*(lsn: Lsn): string = $uint64(lsn) -proc newTxnManager*(): TxnManager = +proc newTxnManager*(txnTimeoutMs: int64 = 30000): TxnManager = new(result) initLock(result.lock) result.nextTxnId = 1 @@ -57,6 +58,7 @@ proc newTxnManager*(): TxnManager = result.activeTxns = initTable[TxnId, Transaction]() result.committedTxns = @[] result.globalVersions = initTable[string, seq[VersionedRecord]]() + result.txnTimeoutMs = txnTimeoutMs proc allocTxnId(tm: TxnManager): TxnId = result = TxnId(tm.nextTxnId) @@ -157,6 +159,14 @@ proc write*(tm: TxnManager, txn: Transaction, key: string, value: seq[byte]): bo release(tm.lock) return false + # Timeout-based deadlock detection: abort stale transactions + let now = getMonoTime().ticks() + for otherId, otherTxn in tm.activeTxns: + if otherId != txn.id and otherTxn.state == tsActive: + if now - otherTxn.startTime > tm.txnTimeoutMs * 1_000_000: + otherTxn.state = tsAborted + tm.activeTxns.del(otherId) + # Check for write-write conflict if key notin txn.writeSet: if key in tm.globalVersions: diff --git a/src/barabadb/query/ast.nim b/src/barabadb/query/ast.nim index dd32158..1481194 100644 --- a/src/barabadb/query/ast.nim +++ b/src/barabadb/query/ast.nim @@ -20,6 +20,10 @@ type nkCommitTxn nkRollbackTxn nkExplainStmt + nkCreateView + nkDropView + nkCreateMigration + nkApplyMigration # Clauses nkFrom @@ -200,6 +204,18 @@ type of nkExplainStmt: expStmt*: Node expAnalyze*: bool + of nkCreateView: + cvName*: string + cvQuery*: Node + cvOrReplace*: bool + of nkDropView: + dvName*: string + dvIfExists*: bool + of nkCreateMigration: + cmName*: string + cmBody*: string + of nkApplyMigration: + amName*: string of nkCreateIndex: ciTarget*: string ciName*: string diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index a3be359..c677e11 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -32,6 +32,7 @@ type db*: LSMTree tables*: Table[string, TableDef] btrees*: Table[string, BTreeIndex[string, IndexEntry]] + views*: Table[string, Node] # view name -> SELECT AST txnManager*: TxnManager pendingTxn*: Transaction onChange*: proc(ev: ChangeEvent) {.closure.} @@ -86,6 +87,7 @@ proc restoreSchema(ctx: ExecutionContext) proc newExecutionContext*(db: LSMTree): ExecutionContext = result = ExecutionContext(db: db, tables: initTable[string, TableDef](), btrees: initTable[string, BTreeIndex[string, IndexEntry]](), + views: initTable[string, Node](), onChange: nil) restoreSchema(result) @@ -125,7 +127,8 @@ proc restoreSchema(ctx: ExecutionContext) = proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = ExecutionContext(db: ctx.db, tables: ctx.tables, - btrees: ctx.btrees, txnManager: ctx.txnManager, + btrees: ctx.btrees, views: ctx.views, + txnManager: ctx.txnManager, pendingTxn: nil, onChange: ctx.onChange) # ---------------------------------------------------------------------- @@ -695,6 +698,48 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = let stmt = astNode.stmts[0] case stmt.kind of nkSelect: + # Expand view if FROM table is a view + if stmt.selFrom != nil and stmt.selFrom.fromTable in ctx.views: + let viewQuery = ctx.views[stmt.selFrom.fromTable] + if viewQuery != nil and viewQuery.kind == nkSelect: + # Execute the view's underlying query + var inner = Node(kind: nkStatementList, stmts: @[]) + inner.stmts.add(viewQuery) + let innerResult = executeQuery(ctx, inner) + # Now filter and project with outer query constraints + var filteredRows = innerResult.rows + var cols = innerResult.columns + if stmt.selWhere != nil and stmt.selWhere.whereExpr != nil: + let whereIr = lowerExpr(stmt.selWhere.whereExpr) + var tmp: seq[Row] = @[] + for row in filteredRows: + if evalExpr(whereIr, row) == "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) + try: + let fa = parseFloat(va) + let fb = parseFloat(vb) + if fa < fb: return -1 + if fa > fb: return 1 + return 0 + except: + return cmp(va, vb) + filteredRows.sort(sortCmp, if asc: Ascending else: Descending) + if stmt.selLimit != nil: + let limitVal = if stmt.selLimit.limitExpr.kind == nkIntLit: + int(stmt.selLimit.limitExpr.intVal) else: 0 + if limitVal > 0 and limitVal < filteredRows.len: + filteredRows = filteredRows[0.. 0: if stmt.selWhere != nil and stmt.selWhere.whereExpr != nil: @@ -944,6 +989,35 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): ExecResult = return okResult(msg="ALTER TABLE " & stmt.altName & " executed") return errResult("Table '" & stmt.altName & "' does not exist") + of nkCreateView: + ctx.views[stmt.cvName] = stmt.cvQuery + return okResult(msg="CREATE VIEW " & stmt.cvName) + + of nkDropView: + if stmt.dvName in ctx.views: + ctx.views.del(stmt.dvName) + return okResult(msg="DROP VIEW " & stmt.dvName) + return errResult("View '" & stmt.dvName & "' does not exist") + + of nkCreateMigration: + # Store migration in LSM-Tree + let migKey = "_schema:migration:" & stmt.cmName + ctx.db.put(migKey, cast[seq[byte]](stmt.cmBody)) + return okResult(msg="CREATE MIGRATION " & stmt.cmName) + + of nkApplyMigration: + # Execute stored migration SQL + let migKey = "_schema:migration:" & stmt.amName + let (found, val) = ctx.db.get(migKey) + if not found: + return errResult("Migration '" & stmt.amName & "' not found") + let sql = cast[string](val) + let tokens = qlex.tokenize(sql) + let astNode = qpar.parse(tokens) + if astNode.stmts.len > 0: + return executeQuery(ctx, astNode) + return okResult(msg="APPLY MIGRATION " & stmt.amName) + of nkCreateIndex: let idxName = if stmt.ciName.len > 0: stmt.ciName else: stmt.ciTarget & "." & stmt.ciTarget diff --git a/src/barabadb/query/lexer.nim b/src/barabadb/query/lexer.nim index e338225..8a7547b 100644 --- a/src/barabadb/query/lexer.nim +++ b/src/barabadb/query/lexer.nim @@ -90,6 +90,9 @@ type tkCommit tkRollback tkExplain + tkView + tkMigration + tkApply tkCount tkSum tkAvg @@ -233,6 +236,9 @@ const keywords*: Table[string, TokenKind] = { "commit": tkCommit, "rollback": tkRollback, "explain": tkExplain, + "view": tkView, + "migration": tkMigration, + "apply": tkApply, "count": tkCount, "sum": tkSum, "avg": tkAvg, diff --git a/src/barabadb/query/parser.nim b/src/barabadb/query/parser.nim index d8a34c3..52c151d 100644 --- a/src/barabadb/query/parser.nim +++ b/src/barabadb/query/parser.nim @@ -738,6 +738,47 @@ proc parseExplain(p: var Parser): Node = result.expAnalyze = true result.expStmt = p.parseStatement() +proc parseCreateView(p: var Parser): Node = + let tok = p.expect(tkCreate) + var orReplace = false + if p.peek().kind == tkIdent and p.peek().value.toLower() == "or": + discard p.advance() + discard p.expect(tkIdent) # REPLACE + orReplace = true + discard p.expect(tkView) + let name = p.expect(tkIdent).value + discard p.expect(tkAs) + let query = p.parseSelect() + result = Node(kind: nkCreateView, cvName: name, cvQuery: query, + cvOrReplace: orReplace, line: tok.line, col: tok.col) + +proc parseDropView(p: var Parser): Node = + let tok = p.expect(tkDrop) + discard p.expect(tkView) + var ifExists = false + if p.peek().kind == tkIdent and p.peek().value.toLower() == "if": + discard p.advance() + discard p.expect(tkExists) + ifExists = true + let name = p.expect(tkIdent).value + result = Node(kind: nkDropView, dvName: name, dvIfExists: ifExists, + line: tok.line, col: tok.col) + +proc parseCreateMigration(p: var Parser): Node = + let tok = p.expect(tkCreate) + discard p.expect(tkMigration) + let name = p.expect(tkIdent).value + discard p.expect(tkAs) + let body = p.expect(tkStringLit).value + result = Node(kind: nkCreateMigration, cmName: name, cmBody: body, + line: tok.line, col: tok.col) + +proc parseApplyMigration(p: var Parser): Node = + let tok = p.expect(tkIdent) # APPLY + discard p.expect(tkMigration) + let name = p.expect(tkIdent).value + result = Node(kind: nkApplyMigration, amName: name, line: tok.line, col: tok.col) + proc parseStatement*(p: var Parser): Node = case p.peek().kind of tkWith, tkSelect: p.parseSelect() @@ -752,13 +793,25 @@ proc parseStatement*(p: var Parser): Node = p.parseCreateTable() elif next.kind == tkIndex or next.kind == tkUnique: p.parseCreateIndex() + elif next.kind == tkView or + (next.kind == tkIdent and next.value.toLower() == "or"): + p.parseCreateView() + elif next.kind == tkMigration: + p.parseCreateMigration() else: p.parseCreateType() else: p.parseCreateType() of tkDrop: - if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkTable: - p.parseDropTable() + if p.pos + 1 < p.tokens.len: + let next = p.tokens[p.pos + 1] + if next.kind == tkTable: + p.parseDropTable() + elif next.kind == tkView: + p.parseDropView() + else: + let tok = p.advance() + Node(kind: nkNullLit, line: tok.line, col: tok.col) else: let tok = p.advance() Node(kind: nkNullLit, line: tok.line, col: tok.col) @@ -768,6 +821,12 @@ proc parseStatement*(p: var Parser): Node = 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() + else: + let tok = p.advance() + Node(kind: nkNullLit, line: tok.line, col: tok.col) of tkBegin: p.parseBeginTxn() of tkCommit: p.parseCommitTxn() of tkRollback: p.parseRollbackTxn()