feat: Phase 2 — MVCC transactions wired into server pipeline

- Added TxnManager to server and ExecutionContext
- Per-connection ExecutionContext with isolated transaction state
- BEGIN creates new transaction via MVCC
- INSERT/DELETE uses transaction write buffer when active
- COMMIT flushes write set to LSM-Tree
- ROLLBACK aborts and discards pending writes
- cloneForConnection() shares tables/btrees/data, isolates transactions
- All 216 tests pass
This commit is contained in:
2026-05-06 11:19:18 +03:00
parent 61f1e24913
commit 1c037ff4ac
2 changed files with 52 additions and 8 deletions
+10 -2
View File
@@ -4,6 +4,7 @@ import std/asyncnet
import std/strutils
import std/os
import std/endians
import std/tables
import config
import ../protocol/wire
import ../query/lexer
@@ -11,6 +12,7 @@ import ../query/parser
import ../query/ast
import ../query/executor
import ../storage/lsm
import ../core/mvcc
type
Server* = ref object
@@ -18,15 +20,20 @@ type
running: bool
db: LSMTree
ctx: ExecutionContext
txnManager: TxnManager
ClientConnection = ref object
socket: AsyncSocket
id: int
currentTxn: Transaction
proc newServer*(config: BaraConfig): Server =
let dataDir = config.dataDir / "server"
let db = newLSMTree(dataDir)
Server(config: config, running: false, db: db, ctx: newExecutionContext(db))
let ctx = newExecutionContext(db)
ctx.txnManager = newTxnManager()
Server(config: config, running: false, db: db, ctx: ctx,
txnManager: ctx.txnManager)
# ----------------------------------------------------------------------
# Wire Protocol Helpers
@@ -141,6 +148,7 @@ proc recvExact(client: AsyncSocket, size: int): Future[string] {.async.} =
proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} =
echo "Client ", clientId, " connected"
var connCtx = cloneForConnection(server.ctx)
try:
while true:
# Read 12-byte header
@@ -166,7 +174,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
let queryStr = readString(cast[seq[byte]](payload), pos)
echo "[", clientId, "] Query: ", queryStr
let (success, result, errorMsg) = executeQuery(server.db, server.ctx, queryStr)
let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr)
if success:
if result.rows.len > 0:
let dataMsg = serializeResult(result, header.requestId)
+38 -2
View File
@@ -7,6 +7,7 @@ import ir
import ../core/types
import ../storage/lsm
import ../storage/btree
import ../core/mvcc
type
IndexEntry = ref object
@@ -17,6 +18,8 @@ type
db*: LSMTree
tables*: Table[string, TableDef] # table name -> definition
btrees*: Table[string, BTreeIndex[string, IndexEntry]] # index name -> btree
txnManager*: TxnManager
pendingTxn*: Transaction # active transaction for this context
TableDef* = object
name*: string
@@ -37,6 +40,12 @@ proc newExecutionContext*(db: LSMTree): ExecutionContext =
ExecutionContext(db: db, tables: initTable[string, TableDef](),
btrees: initTable[string, BTreeIndex[string, IndexEntry]]())
proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext =
## Create a per-connection context that shares the same data but has its own transaction.
ExecutionContext(db: ctx.db, tables: ctx.tables,
btrees: ctx.btrees, txnManager: ctx.txnManager,
pendingTxn: nil)
proc execScan(ctx: ExecutionContext, table: string): seq[Row] =
## Full table scan via LSM-Tree memtable scan.
## Rows are stored as: "{table}.{key}" -> value
@@ -66,7 +75,6 @@ proc execPointRead(ctx: ExecutionContext, table: string, key: string): seq[Row]
return @[]
proc execInsert(ctx: ExecutionContext, table: string, fields: seq[string], values: seq[seq[string]]): int =
## Insert rows into LSM-Tree with constraint validation and B-Tree index population.
var count = 0
for rowVals in values:
var key = ""
@@ -83,9 +91,14 @@ proc execInsert(ctx: ExecutionContext, table: string, fields: seq[string], value
valParts.add(f & "=")
let valStr = valParts.join(",")
let fullKey = table & "." & key
# Use MVCC transaction if active, otherwise write directly
if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive:
discard ctx.txnManager.write(ctx.pendingTxn, fullKey, cast[seq[byte]](valStr))
else:
ctx.db.put(fullKey, cast[seq[byte]](valStr))
# Populate B-Tree indexes
# Populate B-Tree indexes (always direct, not transactional for now)
for colName, btIdx in ctx.btrees:
if colName.startsWith(table & "."):
let colOnly = colName[table.len + 1..^1]
@@ -228,6 +241,9 @@ proc execDelete(ctx: ExecutionContext, table: string, key: string): int =
let fullKey = table & "." & key
let (found, _) = ctx.db.get(fullKey)
if found:
if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive:
discard ctx.txnManager.delete(ctx.pendingTxn, fullKey)
else:
ctx.db.delete(fullKey)
return 1
return 0
@@ -580,13 +596,33 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node): (bool, string, int) =
return (true, "", 0)
of nkBeginTxn:
if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive:
# Auto-commit previous pending transaction
discard ctx.txnManager.commit(ctx.pendingTxn)
ctx.pendingTxn = ctx.txnManager.beginTxn(ilReadCommitted)
return (true, "", 0)
of nkCommitTxn:
if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive:
# Flush write set to LSM-Tree
for key, version in ctx.pendingTxn.writeSet:
if version.value == @[]:
ctx.db.delete(key)
else:
ctx.db.put(key, version.value)
discard ctx.txnManager.commit(ctx.pendingTxn)
ctx.pendingTxn = nil
return (true, "", 0)
else:
return (false, "No active transaction to commit", 0)
of nkRollbackTxn:
if ctx.pendingTxn != nil:
discard ctx.txnManager.abortTxn(ctx.pendingTxn)
ctx.pendingTxn = nil
return (true, "", 0)
else:
return (false, "No active transaction to rollback", 0)
of nkCreateType:
return (true, "", 0)