From 14596c9584f0537c1143a2090f9da6ed3202efbd Mon Sep 17 00:00:00 2001 From: dimgigov Date: Tue, 19 May 2026 20:43:00 +0300 Subject: [PATCH] fix(adapter,executor,tests): harden nimforum adapter and fix transaction/aggregate bugs Adapter (baradb_sqlite.nim): - getRow now returns @[] for missing rows (db_sqlite compat) - insertID/nextId handle NULL (\N) values correctly - preprocessQuery replaces SQLite-specific syntax: TIMESTAMP->DATETIME, DATETIME('now')->current timestamp, INET->STRING - add dbBegin/dbCommit/dbRollback for transaction support Executor: - fix COMMIT treating empty values as DELETE (rows with only PK were lost) - fix MIN/MAX aggregates including NULL (\N) which skewed results - add numeric comparison for irNeq Tests: - stress_test: shared DB, 16 workers, correctness scan after run - nimforum_smoke_test: 10 tests including schema, transactions, iterators --- .gitignore | 1 + adaptors/nim/baradb_sqlite.nim | 51 +++++++-- src/barabadb/query/ast.nim | 3 + src/barabadb/query/executor.nim | 178 ++++++++++++++++++++---------- src/barabadb/query/parser.nim | 32 +++++- tests/nimforum_smoke_test.nim | 188 +++++++++++++++++++++++++++++++- tests/stress_test.nim | 102 ++++++++++++----- 7 files changed, 453 insertions(+), 102 deletions(-) diff --git a/.gitignore b/.gitignore index 0dad0f7..c1cf022 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,4 @@ src/barabadb/storage/compaction src/barabadb/query/executor tests/join_tests *.tar.gz +tests/nimforum_smoke_test diff --git a/adaptors/nim/baradb_sqlite.nim b/adaptors/nim/baradb_sqlite.nim index 83e1685..c0997ed 100644 --- a/adaptors/nim/baradb_sqlite.nim +++ b/adaptors/nim/baradb_sqlite.nim @@ -1,6 +1,6 @@ ## BaraDB adapter mimicking db_sqlite API for NimForum -import std/strutils, std/tables, std/sequtils, std/parseutils +import std/strutils, std/parseutils, std/times import ../../clients/nim/src/baradb/client as baradb_client type @@ -24,6 +24,17 @@ proc dbQuote*(s: string): string = else: add(result, c) add(result, '\'') +proc preprocessQuery(q: string): string = + ## Replace SQLite-specific expressions with BaraDB-compatible equivalents. + result = q + let nowStr = format(now(), "yyyy-MM-dd HH:mm:ss") + result = result.replace("TIMESTAMP", "DATETIME") + result = result.replace("timestamp", "DATETIME") + result = result.replace("DATETIME('now')", "'" & nowStr & "'") + result = result.replace("datetime('now')", "'" & nowStr & "'") + result = result.replace("INET", "STRING") + result = result.replace("inet", "STRING") + proc dbFormat*(formatstr: SqlQuery, args: varargs[string]): string = var res = "" var a = 0 @@ -35,7 +46,7 @@ proc dbFormat*(formatstr: SqlQuery, args: varargs[string]): string = inc(a) else: add(res, c) - res + result = preprocessQuery(res) proc toClient(db: DbConn): SyncClient = cast[SyncClient](db) @@ -82,7 +93,7 @@ proc getRow*(db: DbConn, query: SqlQuery, args: varargs[string, `$`]): seq[strin if qr.rows.len > 0: return qr.rows[0] else: - return newSeq[string](qr.columns.len) + return @[] proc getAllRows*(db: DbConn, query: SqlQuery, args: varargs[string, `$`]): seq[seq[string]] = let client = toClient(db) @@ -119,10 +130,12 @@ proc insertID*(db: DbConn, query: SqlQuery, args: varargs[string, `$`]): int64 = if tableName.len > 0: let idQr = client.query("SELECT max(id) FROM " & tableName) if idQr.rows.len > 0 and idQr.rows[0].len > 0: - try: - return parseInt(idQr.rows[0][0]).int64 - except: - discard + let val = idQr.rows[0][0] + if val.len > 0 and val != "\\N": + try: + return parseInt(val).int64 + except: + discard return -1 proc tryInsertID*(db: DbConn, query: SqlQuery, args: varargs[string, `$`]): int64 = @@ -135,11 +148,13 @@ proc nextId*(db: DbConn, tableName: string): int64 = let client = toClient(db) let qr = client.query("SELECT max(id) FROM " & tableName) if qr.rows.len > 0 and qr.rows[0].len > 0: - try: - let current = parseInt(qr.rows[0][0]) - return current.int64 + 1 - except: - return 1 + let val = qr.rows[0][0] + if val.len > 0 and val != "\\N": + try: + let current = parseInt(val) + return current.int64 + 1 + except: + return 1 return 1 proc execAffectedRows*(db: DbConn, query: SqlQuery, args: varargs[string, `$`]): int64 = @@ -147,3 +162,15 @@ proc execAffectedRows*(db: DbConn, query: SqlQuery, args: varargs[string, `$`]): let q = dbFormat(query, args) let qr = client.query(q) return qr.affectedRows.int64 + +proc dbBegin*(db: DbConn) = + let client = toClient(db) + discard client.query("BEGIN") + +proc dbCommit*(db: DbConn) = + let client = toClient(db) + discard client.query("COMMIT") + +proc dbRollback*(db: DbConn) = + let client = toClient(db) + discard client.query("ROLLBACK") diff --git a/src/barabadb/query/ast.nim b/src/barabadb/query/ast.nim index 3cf4f61..e5931cf 100644 --- a/src/barabadb/query/ast.nim +++ b/src/barabadb/query/ast.nim @@ -428,14 +428,17 @@ type of nkInExpr: inLeft*: Node inRight*: Node + inNegated*: bool of nkBetweenExpr: betweenExpr*: Node betweenLow*: Node betweenHigh*: Node + betweenNegated*: bool of nkLikeExpr: likeExpr*: Node likePattern*: Node likeCaseInsensitive*: bool + likeNegated*: bool of nkIsExpr: isExpr*: Node isType*: string diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index f6b9eaa..a2d533f 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -252,6 +252,8 @@ proc exprToSql(node: Node): string = return node.funcName & "()" of nkUnaryOp: return $node.unOp & " " & exprToSql(node.unOperand) + of nkPath: + return node.pathParts.join(".") else: return $node.kind @@ -956,7 +958,12 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if parseFloat(left) == parseFloat(right): return "true" except: discard return "false" - of irNeq: return if left != right: "true" else: "false" + of irNeq: + if left != right: return "true" + # Try numeric comparison + try: + return if parseFloat(left) != parseFloat(right): "true" else: "false" + except: return "false" of irLt: try: return if parseFloat(left) < parseFloat(right): "true" else: "false" @@ -2374,41 +2381,73 @@ proc lowerExpr*(node: Node): IRExpr = result.binOp = if node.likeCaseInsensitive: irILike else: irLike result.binLeft = lowerExpr(node.likeExpr) result.binRight = lowerExpr(node.likePattern) + if node.likeNegated: + let wrapped = result + result = IRExpr(kind: irekUnary, valueKind: vkBool, unOp: irNot, unExpr: wrapped) of nkBetweenExpr: - result = IRExpr(kind: irekBinary, valueKind: vkBool) - result.binOp = irAnd - let leftCmp = IRExpr(kind: irekBinary, valueKind: vkBool) - leftCmp.binOp = irGte - leftCmp.binLeft = lowerExpr(node.betweenExpr) - leftCmp.binRight = lowerExpr(node.betweenLow) - let rightCmp = IRExpr(kind: irekBinary, valueKind: vkBool) - rightCmp.binOp = irLte - rightCmp.binLeft = lowerExpr(node.betweenExpr) - rightCmp.binRight = lowerExpr(node.betweenHigh) - result.binLeft = leftCmp - result.binRight = rightCmp + if node.betweenNegated: + # NOT BETWEEN => NOT (expr >= low AND expr <= high) + result = IRExpr(kind: irekBinary, valueKind: vkBool) + result.binOp = irOr + let leftCmp = IRExpr(kind: irekBinary, valueKind: vkBool) + leftCmp.binOp = irLt + leftCmp.binLeft = lowerExpr(node.betweenExpr) + leftCmp.binRight = lowerExpr(node.betweenLow) + let rightCmp = IRExpr(kind: irekBinary, valueKind: vkBool) + rightCmp.binOp = irGt + rightCmp.binLeft = lowerExpr(node.betweenExpr) + rightCmp.binRight = lowerExpr(node.betweenHigh) + result.binLeft = leftCmp + result.binRight = rightCmp + else: + result = IRExpr(kind: irekBinary, valueKind: vkBool) + result.binOp = irAnd + let leftCmp = IRExpr(kind: irekBinary, valueKind: vkBool) + leftCmp.binOp = irGte + leftCmp.binLeft = lowerExpr(node.betweenExpr) + leftCmp.binRight = lowerExpr(node.betweenLow) + let rightCmp = IRExpr(kind: irekBinary, valueKind: vkBool) + rightCmp.binOp = irLte + rightCmp.binLeft = lowerExpr(node.betweenExpr) + rightCmp.binRight = lowerExpr(node.betweenHigh) + result.binLeft = leftCmp + result.binRight = rightCmp of nkInExpr: if node.inRight.kind == nkArrayLit: - result = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkBool, boolVal: false)) - for elem in node.inRight.arrayElems: - let eqCmp = IRExpr(kind: irekBinary) - eqCmp.binOp = irEq - eqCmp.binLeft = lowerExpr(node.inLeft) - eqCmp.binRight = lowerExpr(elem) - let orNode = IRExpr(kind: irekBinary) - orNode.binOp = irOr - orNode.binLeft = result - orNode.binRight = eqCmp - result = orNode + if node.inNegated: + # NOT IN (list) => AND of != comparisons + result = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkBool, boolVal: true)) + for elem in node.inRight.arrayElems: + let neqCmp = IRExpr(kind: irekBinary) + neqCmp.binOp = irNeq + neqCmp.binLeft = lowerExpr(node.inLeft) + neqCmp.binRight = lowerExpr(elem) + let andNode = IRExpr(kind: irekBinary) + andNode.binOp = irAnd + andNode.binLeft = result + andNode.binRight = neqCmp + result = andNode + else: + result = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkBool, boolVal: false)) + for elem in node.inRight.arrayElems: + let eqCmp = IRExpr(kind: irekBinary) + eqCmp.binOp = irEq + eqCmp.binLeft = lowerExpr(node.inLeft) + eqCmp.binRight = lowerExpr(elem) + let orNode = IRExpr(kind: irekBinary) + orNode.binOp = irOr + orNode.binLeft = result + orNode.binRight = eqCmp + result = orNode elif node.inRight.kind == nkSubquery: result = IRExpr(kind: irekBinary) - result.binOp = irIn + result.binOp = if node.inNegated: irNotIn else: irIn result.binLeft = lowerExpr(node.inLeft) result.binRight = IRExpr(kind: irekSubquery) result.binRight.subqueryPlan = lowerSelect(node.inRight.subQuery) else: result = IRExpr(kind: irekBinary) - result.binOp = irEq + result.binOp = if node.inNegated: irNeq else: irEq result.binLeft = lowerExpr(node.inLeft) result.binRight = lowerExpr(node.inRight) of nkExists: @@ -2900,12 +2939,14 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = 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 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 newRow[alias] = maxVal of irArrayAgg: @@ -2964,20 +3005,23 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = of irpkSort: var sourceRows = executePlan(ctx, plan.sortSource) if plan.sortExprs.len == 0: return sourceRows - 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, ctx) - let vb = evalExpr(sortExpr, b, ctx) - 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) - sourceRows.sort(sortCmp, if ascending: Ascending else: Descending) + for i, sortExpr in plan.sortExprs: + let ascending = if i < plan.sortDirs.len: plan.sortDirs[i] else: true + let va = evalExpr(sortExpr, a, ctx) + let vb = evalExpr(sortExpr, b, ctx) + var cmpRes = 0 + try: + let fa = parseFloat(va) + let fb = parseFloat(vb) + if fa < fb: cmpRes = -1 + elif fa > fb: cmpRes = 1 + except: + cmpRes = cmp(va, vb) + if not ascending: cmpRes = -cmpRes + if cmpRes != 0: return cmpRes + return 0 + sourceRows.sort(sortCmp, Ascending) return sourceRows of irpkLimit: @@ -3075,12 +3119,14 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = 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 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 aggRow[aggKey] = maxVal of irArrayAgg: @@ -3107,7 +3153,21 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let leftRows = executePlan(ctx, plan.joinLeft) result = @[] - proc mergeRow(left, right: Row, leftAlias, rightAlias: string): Row = + proc getScanTable(p: IRPlan): string = + if p == nil: return "" + case p.kind + of irpkScan: p.scanTable + of irpkJoin: getScanTable(p.joinRight) + else: "" + + proc getLeftmostScanTable(p: IRPlan): string = + if p == nil: return "" + case p.kind + of irpkScan: p.scanTable + of irpkJoin: getLeftmostScanTable(p.joinLeft) + else: "" + + proc mergeRow(left, right: Row, leftAlias, rightAlias, leftTable, rightTable: string): Row = result = initTable[string, string]() for k, v in left: if not k.startsWith("$"): @@ -3115,17 +3175,20 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for k, v in right: if not k.startsWith("$") and k notin result: result[k] = v - if leftAlias.len > 0: + let lq = if leftAlias.len > 0: leftAlias elif leftTable.len > 0: leftTable else: "" + if lq.len > 0: for k, v in left: if not k.startsWith("$"): - result[leftAlias & "." & k] = v - if rightAlias.len > 0: + result[lq & "." & k] = v + let rq = if rightAlias.len > 0: rightAlias elif rightTable.len > 0: rightTable else: "" + if rq.len > 0: for k, v in right: if not k.startsWith("$"): - result[rightAlias & "." & k] = v + result[rq & "." & k] = v let leftAlias = if plan.joinLeft != nil and plan.joinLeft.kind == irpkScan: plan.joinLeft.scanAlias else: "" + let leftTable = getLeftmostScanTable(plan.joinLeft) # LATERAL JOIN: for each left row, scan right, merge, then filter/sort/limit if plan.joinLateral: @@ -3169,7 +3232,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = # Merge, filter var mergedRows: seq[Row] for r in rawRightRows: - let merged = mergeRow(l, r, leftAlias, rightAlias) + let merged = mergeRow(l, r, leftAlias, rightAlias, leftTable, getScanTable(plan.joinRight)) if rightFilter != nil and evalExpr(rightFilter, merged, ctx) != "true": continue if plan.joinCond != nil and evalExpr(plan.joinCond, merged, ctx) != "true": @@ -3245,11 +3308,12 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let rightAlias = if plan.joinRight != nil and plan.joinRight.kind == irpkScan: plan.joinRight.scanAlias else: "" + let rightTable = getScanTable(plan.joinRight) if plan.joinKind == irjkCross: for l in leftRows: for r in rightRows: - result.add(mergeRow(l, r, leftAlias, rightAlias)) + result.add(mergeRow(l, r, leftAlias, rightAlias, leftTable, rightTable)) return result case plan.joinStrategy @@ -3296,9 +3360,9 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = matchedProbe[i] = true for brow in hashTable[key]: if buildIsLeft: - result.add(mergeRow(brow, prow, leftAlias, rightAlias)) + result.add(mergeRow(brow, prow, leftAlias, rightAlias, leftTable, rightTable)) else: - result.add(mergeRow(prow, brow, leftAlias, rightAlias)) + result.add(mergeRow(prow, brow, leftAlias, rightAlias, leftTable, rightTable)) # LEFT / RIGHT / FULL padding for non-matching probe rows if plan.joinKind == irjkLeft or plan.joinKind == irjkFull or plan.joinKind == irjkRight: @@ -3414,9 +3478,9 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for entry in entries: let parsed = parseRowData(entry.rowValue) if outerIsLeft: - result.add(mergeRow(orow, parsed, leftAlias, rightAlias)) + result.add(mergeRow(orow, parsed, leftAlias, rightAlias, leftTable, rightTable)) else: - result.add(mergeRow(parsed, orow, leftAlias, rightAlias)) + result.add(mergeRow(parsed, orow, leftAlias, rightAlias, leftTable, rightTable)) elif plan.joinKind == irjkLeft or plan.joinKind == irjkRight or plan.joinKind == irjkFull: matchedOuter[i] = false @@ -3455,7 +3519,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for l in leftRows: var matched = false for r in rightRows: - let merged = mergeRow(l, r, leftAlias, rightAlias) + let merged = mergeRow(l, r, leftAlias, rightAlias, leftTable, rightTable) if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": result.add(merged) matched = true @@ -3479,7 +3543,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = for r in rightRows: var found = false for l in leftRows: - let merged = mergeRow(l, r, leftAlias, rightAlias) + let merged = mergeRow(l, r, leftAlias, rightAlias, leftTable, rightTable) if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": found = true break @@ -3568,12 +3632,14 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = 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 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 aggResult = maxVal else: discard @@ -3878,7 +3944,7 @@ proc getSelectColumns(stmt: Node): seq[string] = elif e.kind == nkIdent: result.add(e.identName) elif e.kind == nkPath and e.pathParts.len > 0: - result.add(e.pathParts[^1]) + result.add(e.pathParts.join(".")) elif e.kind == nkFuncCall: var aliasArgs: seq[string] = @[] for arg in e.funcArgs: @@ -4740,8 +4806,10 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive: var kvPairs: seq[(string, seq[byte])] for key, version in ctx.pendingTxn.writeSet: - if version.value == @[]: ctx.db.delete(key) - else: ctx.db.put(key, version.value) + if version.value == @[] and version.xmax != TxnId(0): + ctx.db.delete(key) + else: + ctx.db.put(key, version.value) kvPairs.add((key, version.value)) discard ctx.txnManager.commit(ctx.pendingTxn) ctx.pendingTxn = nil diff --git a/src/barabadb/query/parser.nim b/src/barabadb/query/parser.nim index 2ce964e..7f876fb 100644 --- a/src/barabadb/query/parser.nim +++ b/src/barabadb/query/parser.nim @@ -296,6 +296,11 @@ proc parseAddSub(p: var Parser): Node = proc parseComparison(p: var Parser): Node = result = p.parseAddSub() + # Check for NOT prefix on IN/LIKE/BETWEEN + var negated = false + if p.peek().kind == tkNot: + discard p.advance() + negated = true # Handle BETWEEN ... AND ... if p.peek().kind == tkBetween: let tok = p.advance() @@ -303,12 +308,27 @@ proc parseComparison(p: var Parser): Node = discard p.expect(tkAnd) let high = p.parseAddSub() return Node(kind: nkBetweenExpr, betweenExpr: result, - betweenLow: low, betweenHigh: high, line: tok.line, col: tok.col) + betweenLow: low, betweenHigh: high, + betweenNegated: negated, + line: tok.line, col: tok.col) # Handle IN (subquery | list) if p.peek().kind == tkIn: let tok = p.advance() - let right = p.parseAddSub() + discard p.expect(tkLParen) + let right: Node = + if p.peek().kind == tkSelect: + let sub = p.parseSelect() + discard p.expect(tkRParen) + Node(kind: nkSubquery, subQuery: sub, line: tok.line, col: tok.col) + else: + var elems: seq[Node] = @[] + elems.add(p.parseExpr()) + while p.match(tkComma): + elems.add(p.parseExpr()) + discard p.expect(tkRParen) + Node(kind: nkArrayLit, arrayElems: elems, line: tok.line, col: tok.col) return Node(kind: nkInExpr, inLeft: result, inRight: right, + inNegated: negated, line: tok.line, col: tok.col) # Handle LIKE / ILIKE if p.peek().kind in {tkLike, tkILike}: @@ -316,7 +336,13 @@ proc parseComparison(p: var Parser): Node = let tok = p.advance() let pattern = p.parseAddSub() return Node(kind: nkLikeExpr, likeExpr: result, likePattern: pattern, - likeCaseInsensitive: isILike, line: tok.line, col: tok.col) + likeCaseInsensitive: isILike, + likeNegated: negated, + line: tok.line, col: tok.col) + # If we consumed NOT but didn't find IN/LIKE/BETWEEN, put it back is not possible. + # Instead, wrap result in a unary NOT node. + if negated: + result = Node(kind: nkUnaryOp, unOp: ukNot, unOperand: result, line: 0, col: 0) # Handle IS NULL / IS NOT NULL if p.peek().kind == tkIs: let tok = p.advance() diff --git a/tests/nimforum_smoke_test.nim b/tests/nimforum_smoke_test.nim index c886ca4..398ba76 100644 --- a/tests/nimforum_smoke_test.nim +++ b/tests/nimforum_smoke_test.nim @@ -1,8 +1,7 @@ -## NimForum Adapter Smoke Test +## NimForum Adapter Smoke Test — comprehensive validation import std/unittest import std/osproc import std/os -import std/strutils import std/strtabs import std/times @@ -24,7 +23,23 @@ suite "NimForum Adapter Smoke Test": env["BARADB_DATA_DIR"] = dataDir env["BARADB_LOG_LEVEL"] = "error" serverProcess = startProcess("./build/baradadb", env=env, options={poStdErrToStdOut, poDaemon}) - sleep(800) + # Wait for server to be ready (with timeout) + var pingDb: DbConn + var connected = false + for i in 0 ..< 50: + sleep(100) + try: + pingDb = open("127.0.0.1:" & $port, "", "", "default") + connected = true + break + except: + discard + if not connected: + serverProcess.terminate() + discard serverProcess.waitForExit() + removeDir(dataDir) + raise newException(IOError, "Server failed to start") + pingDb.close() teardown: if serverProcess != nil: @@ -67,3 +82,170 @@ suite "NimForum Adapter Smoke Test": check row[row.len - 1] == "ten" db.close() + + test "getRow returns @[] for missing rows": + var db = open("127.0.0.1:" & $port, "", "", "default") + db.exec(sql"CREATE TABLE nf_missing (id INT PRIMARY KEY)") + let row = db.getRow(sql"SELECT * FROM nf_missing WHERE id = 999") + check row.len == 0 + db.close() + + test "insertID and nextId": + var db = open("127.0.0.1:" & $port, "", "", "default") + db.exec(sql"CREATE TABLE nf_ids (id INT PRIMARY KEY, name STRING)") + + let id1 = db.insertID(sql"INSERT INTO nf_ids (id, name) VALUES (5, 'five')") + check id1 == 5 + + let next = db.nextId("nf_ids") + check next == 6 + + db.close() + + test "Transactions (BEGIN/COMMIT/ROLLBACK)": + var db = open("127.0.0.1:" & $port, "", "", "default") + db.exec(sql"CREATE TABLE nf_txn (id INT PRIMARY KEY)") + + db.dbBegin() + db.exec(sql"INSERT INTO nf_txn (id) VALUES (1)") + db.dbCommit() + check db.getValue(sql"SELECT count(*) FROM nf_txn") == "1" + + db.dbBegin() + db.exec(sql"INSERT INTO nf_txn (id) VALUES (2)") + db.dbRollback() + check db.getValue(sql"SELECT count(*) FROM nf_txn") == "1" + + db.close() + + test "fastRows iterator": + var db = open("127.0.0.1:" & $port, "", "", "default") + db.exec(sql"CREATE TABLE nf_iter (id INT PRIMARY KEY)") + db.exec(sql"INSERT INTO nf_iter (id) VALUES (1)") + db.exec(sql"INSERT INTO nf_iter (id) VALUES (2)") + db.exec(sql"INSERT INTO nf_iter (id) VALUES (3)") + + var count = 0 + for r in db.fastRows(sql"SELECT id FROM nf_iter ORDER BY id"): + count.inc + check count == 3 + + db.close() + + test "execAffectedRows": + var db = open("127.0.0.1:" & $port, "", "", "default") + db.exec(sql"CREATE TABLE nf_aff (id INT PRIMARY KEY, name STRING)") + db.exec(sql"INSERT INTO nf_aff (id, name) VALUES (1, 'a')") + db.exec(sql"INSERT INTO nf_aff (id, name) VALUES (2, 'b')") + + let affected = db.execAffectedRows(sql"DELETE FROM nf_aff WHERE id = 1") + check affected == 1 + + db.close() + + test "tryExec success and failure": + var db = open("127.0.0.1:" & $port, "", "", "default") + db.exec(sql"CREATE TABLE nf_try (id INT PRIMARY KEY)") + + check db.tryExec(sql"INSERT INTO nf_try (id) VALUES (1)") == true + check db.tryExec(sql"INSERT INTO nf_try (id) VALUES (1)") == false + + db.close() + + test "NimForum schema creation": + var db = open("127.0.0.1:" & $port, "", "", "default") + + # Create nimforum-like schema (adapter preprocesses DATETIME/TIMESTAMP/INET) + db.exec(sql""" + CREATE TABLE thread ( + id INT PRIMARY KEY, + name STRING NOT NULL, + views INT NOT NULL, + modified DATETIME NOT NULL DEFAULT DATETIME('now') + ) + """) + db.exec(sql"CREATE UNIQUE INDEX ThreadNameIx ON thread (name)") + + db.exec(sql""" + CREATE TABLE person ( + id INT PRIMARY KEY, + name STRING NOT NULL, + password STRING NOT NULL, + email STRING NOT NULL, + creation DATETIME NOT NULL DEFAULT DATETIME('now'), + salt STRING NOT NULL, + user_status STRING NOT NULL, + lastOnline DATETIME NOT NULL DEFAULT DATETIME('now'), + ban STRING NOT NULL DEFAULT '' + ) + """) + db.exec(sql"CREATE UNIQUE INDEX UserNameIx ON person (name)") + + db.exec(sql""" + CREATE TABLE post ( + id INT PRIMARY KEY, + author INT NOT NULL, + ip STRING NOT NULL, + header STRING NOT NULL, + content STRING NOT NULL, + thread INT NOT NULL, + creation DATETIME NOT NULL DEFAULT DATETIME('now') + ) + """) + + db.exec(sql""" + CREATE TABLE session ( + id INT PRIMARY KEY, + ip STRING NOT NULL, + password STRING NOT NULL, + userid INT NOT NULL, + lastModified DATETIME NOT NULL DEFAULT DATETIME('now') + ) + """) + + db.exec(sql""" + CREATE TABLE antibot ( + id INT PRIMARY KEY, + ip STRING NOT NULL, + answer STRING NOT NULL, + created DATETIME NOT NULL DEFAULT DATETIME('now') + ) + """) + + db.exec(sql"CREATE INDEX PersonStatusIdx ON person(user_status)") + db.exec(sql"CREATE INDEX PostByAuthorIdx ON post(thread, author)") + + # Insert and query forum data + db.exec(sql"INSERT INTO thread (id, name, views) VALUES (1, 'First Thread', 0)") + db.exec(sql"INSERT INTO person (id, name, password, email, salt, user_status, ban) VALUES (1, 'admin', 'pw', 'a@b.c', 'salt', 'active', '')") + db.exec(sql"INSERT INTO post (id, author, ip, header, content, thread) VALUES (1, 1, '127.0.0.1', 'Hello', 'World', 1)") + + let threads = db.getAllRows(sql"SELECT * FROM thread") + check threads.len == 1 + + let posts = db.getAllRows(sql"SELECT * FROM post WHERE thread = 1") + check posts.len == 1 + + db.close() + + test "NimForum-like operations": + var db = open("127.0.0.1:" & $port, "", "", "default") + + db.exec(sql"CREATE TABLE nf_ops (id INT PRIMARY KEY, views INT NOT NULL)") + db.exec(sql"INSERT INTO nf_ops (id, views) VALUES (1, 0)") + + # Update views (like nimforum does) + let affected = db.execAffectedRows(sql"UPDATE nf_ops SET views = views + 1 WHERE id = 1") + check affected == 1 + + let views = db.getValue(sql"SELECT views FROM nf_ops WHERE id = 1") + check views == "1" + + # Try update with DATETIME('now') (preprocessed by adapter) + db.exec(sql"CREATE TABLE nf_dt (id INT PRIMARY KEY, modified DATETIME)") + db.exec(sql"INSERT INTO nf_dt (id, modified) VALUES (1, DATETIME('now'))") + let dtRow = db.getRow(sql"SELECT modified FROM nf_dt WHERE id = 1") + check dtRow.len == 1 + check dtRow[0].len > 0 + + db.close() diff --git a/tests/stress_test.nim b/tests/stress_test.nim index df38685..3492fdc 100644 --- a/tests/stress_test.nim +++ b/tests/stress_test.nim @@ -1,73 +1,104 @@ -## Stress Test — parallel workloads against LSM-Tree -## Each worker gets its own LSMTree instance (thread-safe via internal locks). +## Stress Test — parallel workloads against shared LSM-Tree +## Validates correctness under concurrent access. import std/os import std/random +import std/strutils import std/times import std/monotimes +import std/locks import barabadb/storage/lsm const - NumWorkers = 10 - OpsPerWorker = 1000 - KeySpace = 500 + NumWorkers = 16 + OpsPerWorker = 2000 + KeySpace = 1000 type WorkerArgs = tuple workerId: int - dataDir: string + db: ptr LSMTree errors: ptr int + lock: ptr Lock randomize() proc runWorker(args: WorkerArgs) {.thread.} = - var db: LSMTree var localErrors = 0 - {.cast(gcsafe).}: - db = newLSMTree(args.dataDir) + var rng = initRand(args.workerId) for i in 0 ..< OpsPerWorker: - let op = rand(0 .. 2) - let key = "key_" & $args.workerId & "_" & $(rand(0 ..< KeySpace)) + let op = rng.rand(0 .. 3) + let key = "key_" & $(rng.rand(0 ..< KeySpace)) let value = cast[seq[byte]]("val_" & $i & "_" & $args.workerId) case op of 0: - db.put(key, value) + # Put + acquire(args.lock[]) + try: + args.db[].put(key, value) + except: + localErrors.inc + finally: + release(args.lock[]) of 1: - let (found, val) = db.get(key) + # Get and verify (if found, value should be from some worker) + acquire(args.lock[]) + var found = false + var val: seq[byte] + try: + (found, val) = args.db[].get(key) + except: + localErrors.inc + finally: + release(args.lock[]) if found: - if val != value and val.len > 0: - discard + let valStr = cast[string](val) + if not valStr.startsWith("val_"): + localErrors.inc of 2: - db.delete(key) + # Delete + acquire(args.lock[]) + try: + args.db[].delete(key) + except: + localErrors.inc + finally: + release(args.lock[]) + of 3: + # Overwrite + acquire(args.lock[]) + try: + args.db[].put(key, value) + except: + localErrors.inc + finally: + release(args.lock[]) else: discard - for k in 0 ..< KeySpace: - let key = "key_" & $args.workerId & "_" & $k - let (found, _) = db.get(key) - discard found - - db.close() args.errors[] = localErrors proc main() = - let baseDir = "/tmp/baradb_stress_test" + let baseDir = "/tmp/baradb_stress_test_shared" removeDir(baseDir) createDir(baseDir) let start = getMonoTime() - var workerDirs: seq[string] = @[] - for i in 0 ..< NumWorkers: - workerDirs.add(baseDir / "worker_" & $i) + # Single shared database + var db = newLSMTree(baseDir) + var dbPtr = addr db + var dbLock: Lock + initLock(dbLock) + var lockPtr = addr dbLock # Run workers in parallel using std/threads var threadArr: array[NumWorkers, Thread[WorkerArgs]] var errorCounts: array[NumWorkers, int] for i in 0 ..< NumWorkers: - let args: WorkerArgs = (workerId: i, dataDir: workerDirs[i], errors: addr errorCounts[i]) + let args: WorkerArgs = (workerId: i, db: dbPtr, errors: addr errorCounts[i], lock: lockPtr) createThread(threadArr[i], runWorker, args) var totalErrors = 0 @@ -75,15 +106,28 @@ proc main() = joinThread(threadArr[i]) totalErrors += errorCounts[i] + # Verify: scan all keys and ensure no corruption + var scanErrors = 0 + for k in 0 ..< KeySpace: + let key = "key_" & $k + let (found, val) = db.get(key) + if found: + let valStr = cast[string](val) + if not valStr.startsWith("val_"): + scanErrors.inc + + db.close() + let elapsed = (getMonoTime() - start).inMilliseconds let totalOps = NumWorkers * OpsPerWorker echo "Stress test completed: ", totalOps, " ops across ", NumWorkers, " workers" echo "Time: ", elapsed, " ms" echo "Throughput: ", float(totalOps) / (float(elapsed) / 1000.0), " ops/sec" - echo "Errors: ", totalErrors + echo "Runtime errors: ", totalErrors + echo "Scan errors: ", scanErrors - if totalErrors > 0: + if totalErrors > 0 or scanErrors > 0: quit(1) removeDir(baseDir)