feat: JSON/JSONB validation, multi-column indexes, CTE execution, wire protocol column types
- Add fkJson to wire protocol with serialization/deserialization - Validate JSON/JSONB on INSERT/UPDATE via std/json - Send real column type metadata in wire protocol responses - Update all 4 clients (Nim, Python, JS, Rust) for JSON and columnTypes - Implement multi-column index parser (CREATE INDEX idx ON t(a, b)) - Add ciColumns to AST nkCreateIndex - Build composite index keys as table.col1.col2 with val1|val2 - Support multi-column exact match in SELECT for AND chains - Implement non-recursive CTE execution via ctx.cteTables materialization - Add tkRecursive token and parse WITH RECURSIVE - Fix test isolation: use temp dirs for JOIN, migration, and RLS tests - All tests passing (0 failures)
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
import std/asyncdispatch
|
||||
import std/asyncnet
|
||||
import std/strutils
|
||||
import std/sequtils
|
||||
import std/tables
|
||||
import std/os
|
||||
import std/endians
|
||||
@@ -73,6 +74,19 @@ proc parseHeader(data: string): (bool, MessageHeader) =
|
||||
# Query Execution (pipeline-based)
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
proc typeToFieldKind*(colType: string): FieldKind =
|
||||
let t = colType.toUpper()
|
||||
if t.startsWith("INT") or t == "SERIAL" or t == "BIGINT" or t == "SMALLINT" or t == "BIGSERIAL" or t == "SMALLSERIAL":
|
||||
return fkInt64
|
||||
elif t.startsWith("FLOAT") or t == "REAL" or t == "DOUBLE" or t == "NUMERIC":
|
||||
return fkFloat64
|
||||
elif t == "BOOLEAN" or t == "BOOL":
|
||||
return fkBool
|
||||
elif t == "JSON" or t == "JSONB":
|
||||
return fkJson
|
||||
else:
|
||||
return fkString
|
||||
|
||||
proc valueToWire(val: string, colType: string): WireValue =
|
||||
if val.len == 0 or val.toLower() == "null":
|
||||
return WireValue(kind: fkNull)
|
||||
@@ -91,6 +105,8 @@ proc valueToWire(val: string, colType: string): WireValue =
|
||||
return WireValue(kind: fkBool, boolVal: true)
|
||||
elif lv in ["false", "f", "no", "0"]:
|
||||
return WireValue(kind: fkBool, boolVal: false)
|
||||
elif t == "JSON" or t == "JSONB":
|
||||
return WireValue(kind: fkJson, jsonVal: val)
|
||||
return WireValue(kind: fkString, strVal: val)
|
||||
|
||||
proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[]): (bool, QueryResult, string) =
|
||||
@@ -127,7 +143,7 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
|
||||
else:
|
||||
colTypes = newSeq[string](result.columns.len)
|
||||
|
||||
qr.columnTypes = newSeq[FieldKind](result.columns.len)
|
||||
qr.columnTypes = colTypes.mapIt(typeToFieldKind(it))
|
||||
qr.rows = @[]
|
||||
for row in result.rows:
|
||||
var wireRow: seq[WireValue] = @[]
|
||||
@@ -154,6 +170,9 @@ proc serializeResult(qr: QueryResult, requestId: uint32): seq[byte] =
|
||||
# Column names
|
||||
for col in qr.columns:
|
||||
payload.writeString(col)
|
||||
# Column types metadata
|
||||
for ct in qr.columnTypes:
|
||||
payload.add(byte(ct))
|
||||
# Row count
|
||||
payload.writeUint32(uint32(qr.rows.len))
|
||||
# Rows
|
||||
|
||||
@@ -44,6 +44,7 @@ type
|
||||
fkArray = 0x0A
|
||||
fkObject = 0x0B
|
||||
fkVector = 0x0C
|
||||
fkJson = 0x0D
|
||||
|
||||
MessageHeader* = object
|
||||
kind*: MsgKind
|
||||
@@ -79,6 +80,7 @@ type
|
||||
of fkArray: arrayVal*: seq[WireValue]
|
||||
of fkObject: objVal*: seq[(string, WireValue)]
|
||||
of fkVector: vecVal*: seq[float32]
|
||||
of fkJson: jsonVal*: string
|
||||
|
||||
QueryResult* = object
|
||||
columns*: seq[string]
|
||||
@@ -175,6 +177,8 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) =
|
||||
var bytes: array[4, byte]
|
||||
copyMem(addr bytes, unsafeAddr fl, 4)
|
||||
buf.add(bytes)
|
||||
of fkJson:
|
||||
buf.writeString(val.jsonVal)
|
||||
|
||||
proc deserializeValue*(buf: openArray[byte], pos: var int): WireValue =
|
||||
let kind = FieldKind(buf[pos])
|
||||
@@ -241,6 +245,8 @@ proc deserializeValue*(buf: openArray[byte], pos: var int): WireValue =
|
||||
pos += 4
|
||||
vec.add(fl)
|
||||
result = WireValue(kind: fkVector, vecVal: vec)
|
||||
of fkJson:
|
||||
result = WireValue(kind: fkJson, jsonVal: readString(buf, pos))
|
||||
|
||||
proc serializeMessage*(msg: WireMessage): seq[byte] =
|
||||
result = @[]
|
||||
|
||||
@@ -149,7 +149,7 @@ type
|
||||
case kind*: NodeKind
|
||||
of nkSelect:
|
||||
selDistinct*: bool
|
||||
selWith*: seq[(string, Node)]
|
||||
selWith*: seq[(string, Node, bool)]
|
||||
selResult*: seq[Node]
|
||||
selFrom*: Node
|
||||
selJoins*: seq[Node]
|
||||
@@ -286,6 +286,7 @@ type
|
||||
of nkCreateIndex:
|
||||
ciTarget*: string
|
||||
ciName*: string
|
||||
ciColumns*: seq[string]
|
||||
ciExpr*: Node
|
||||
ciKind*: IndexKind
|
||||
of nkDropIndex:
|
||||
@@ -309,7 +310,7 @@ type
|
||||
of nkReturning:
|
||||
retExprs*: seq[Node]
|
||||
of nkWith:
|
||||
withBindings*: seq[(string, Node)]
|
||||
withBindings*: seq[(string, Node, bool)]
|
||||
of nkBinOp:
|
||||
binOp*: BinOpKind
|
||||
binLeft*: Node
|
||||
|
||||
+100
-14
@@ -8,6 +8,7 @@ import std/re
|
||||
import checksums/sha2
|
||||
import std/math
|
||||
import std/times
|
||||
import std/json
|
||||
import lexer as qlex
|
||||
import parser as qpar
|
||||
import ast
|
||||
@@ -54,6 +55,7 @@ type
|
||||
tables*: Table[string, TableDef]
|
||||
btrees*: Table[string, BTreeIndex[string, IndexEntry]]
|
||||
views*: Table[string, Node] # view name -> SELECT AST
|
||||
cteTables*: Table[string, seq[Row]] # CTE name -> rows
|
||||
txnManager*: TxnManager
|
||||
pendingTxn*: Transaction
|
||||
onChange*: proc(ev: ChangeEvent) {.closure.}
|
||||
@@ -129,6 +131,7 @@ proc newExecutionContext*(db: LSMTree): ExecutionContext =
|
||||
result = ExecutionContext(db: db, tables: initTable[string, TableDef](),
|
||||
btrees: initTable[string, BTreeIndex[string, IndexEntry]](),
|
||||
views: initTable[string, Node](),
|
||||
cteTables: initTable[string, seq[Row]](),
|
||||
users: initTable[string, UserDef](),
|
||||
policies: initTable[string, seq[PolicyDef]](),
|
||||
currentUser: "", currentRole: "",
|
||||
@@ -190,6 +193,7 @@ proc restoreSchema(ctx: ExecutionContext) =
|
||||
proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext =
|
||||
ExecutionContext(db: ctx.db, tables: ctx.tables,
|
||||
btrees: ctx.btrees, views: ctx.views,
|
||||
cteTables: initTable[string, seq[Row]](),
|
||||
users: ctx.users, policies: ctx.policies,
|
||||
txnManager: ctx.txnManager,
|
||||
currentUser: ctx.currentUser, currentRole: ctx.currentRole,
|
||||
@@ -478,6 +482,9 @@ proc checkInsertPolicy(ctx: ExecutionContext, tableName: string, row: Row): bool
|
||||
|
||||
proc execScan(ctx: ExecutionContext, table: string): seq[Row] =
|
||||
result = @[]
|
||||
# Check CTE tables first
|
||||
if table in ctx.cteTables:
|
||||
return ctx.cteTables[table]
|
||||
let prefix = table & "."
|
||||
for entry in ctx.db.scanMemTable():
|
||||
if entry.deleted: continue
|
||||
@@ -549,10 +556,14 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu
|
||||
|
||||
for colName in ctx.btrees.keys.toSeq():
|
||||
if colName.startsWith(table & "."):
|
||||
let colOnly = colName[table.len + 1..^1]
|
||||
let colVal = getValue(rowVals, fields, colOnly)
|
||||
if colVal.len > 0 and not isNull(colVal):
|
||||
ctx.btrees[colName].insert(colVal, IndexEntry(lsmKey: fullKey, rowValue: valStr))
|
||||
let colsPart = colName[table.len + 1..^1]
|
||||
let idxCols = colsPart.split(".")
|
||||
var colVals: seq[string] = @[]
|
||||
for c in idxCols:
|
||||
colVals.add(getValue(rowVals, fields, c))
|
||||
let idxVal = colVals.join("|")
|
||||
if idxVal.len > 0 and not isNull(idxVal):
|
||||
ctx.btrees[colName].insert(idxVal, IndexEntry(lsmKey: fullKey, rowValue: valStr))
|
||||
|
||||
inc count
|
||||
return count
|
||||
@@ -600,6 +611,25 @@ proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Tab
|
||||
for col, val in parsed:
|
||||
parts.add(col & "=" & val)
|
||||
let newVal = parts.join(",")
|
||||
# Update indexes: remove old, insert new
|
||||
for colName in ctx.btrees.keys.toSeq():
|
||||
if colName.startsWith(table & "."):
|
||||
let colsPart = colName[table.len + 1..^1]
|
||||
let idxCols = colsPart.split(".")
|
||||
var oldVals: seq[string] = @[]
|
||||
var newVals: seq[string] = @[]
|
||||
for c in idxCols:
|
||||
if c in oldRow:
|
||||
oldVals.add(oldRow[c])
|
||||
else:
|
||||
oldVals.add("")
|
||||
if c in parsed:
|
||||
newVals.add(parsed[c])
|
||||
else:
|
||||
newVals.add("")
|
||||
let newIdxVal = newVals.join("|")
|
||||
if newIdxVal.len > 0 and not isNull(newIdxVal):
|
||||
ctx.btrees[colName].insert(newIdxVal, IndexEntry(lsmKey: fullKey, rowValue: newVal))
|
||||
if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive:
|
||||
discard ctx.txnManager.write(ctx.pendingTxn, fullKey, cast[seq[byte]](newVal))
|
||||
else:
|
||||
@@ -626,6 +656,11 @@ proc validateType*(colType: string, value: string): (bool, string) =
|
||||
elif t == "TIMESTAMP" or t == "DATE":
|
||||
if value.len < 8: # minimal date check
|
||||
return (false, "Type mismatch: expected " & t & " but got '" & value & "'")
|
||||
elif t == "JSON" or t == "JSONB":
|
||||
try:
|
||||
discard parseJson(value)
|
||||
except:
|
||||
return (false, "Type mismatch: expected JSON but got '" & value & "'")
|
||||
return (true, "")
|
||||
|
||||
proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult
|
||||
@@ -1319,6 +1354,23 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
|
||||
let stmt = boundAst.stmts[0]
|
||||
case stmt.kind
|
||||
of nkSelect:
|
||||
defer:
|
||||
ctx.cteTables.clear()
|
||||
# Execute CTEs if present
|
||||
if stmt.selWith.len > 0:
|
||||
for (cteName, cteQuery, isRecursive) in stmt.selWith:
|
||||
if isRecursive:
|
||||
# Recursive CTE: not yet fully implemented
|
||||
ctx.cteTables[cteName] = @[]
|
||||
else:
|
||||
var inner = Node(kind: nkStatementList, stmts: @[])
|
||||
inner.stmts.add(cteQuery)
|
||||
let cteRes = executeQuery(ctx, inner)
|
||||
var cteRows: seq[Row] = @[]
|
||||
for row in cteRes.rows:
|
||||
cteRows.add(row)
|
||||
ctx.cteTables[cteName] = cteRows
|
||||
|
||||
# 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]
|
||||
@@ -1365,6 +1417,35 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
|
||||
if stmt.selFrom != nil and stmt.selFrom.fromTable.len > 0:
|
||||
if stmt.selWhere != nil and stmt.selWhere.whereExpr != nil:
|
||||
let w = stmt.selWhere.whereExpr
|
||||
# Multi-column exact match: AND chain of =
|
||||
var eqConds: seq[(string, string)] = @[]
|
||||
proc collectEq(node: Node) =
|
||||
if node.kind == nkBinOp and node.binOp == bkEq and node.binLeft.kind == nkIdent and node.binRight.kind == nkStringLit:
|
||||
eqConds.add((node.binLeft.identName, node.binRight.strVal))
|
||||
elif node.kind == nkBinOp and node.binOp == bkAnd:
|
||||
collectEq(node.binLeft)
|
||||
collectEq(node.binRight)
|
||||
collectEq(w)
|
||||
if eqConds.len >= 2:
|
||||
var idxCols: seq[string] = @[]
|
||||
for c in eqConds: idxCols.add(c[0])
|
||||
let idxName = stmt.selFrom.fromTable & "." & idxCols.join(".")
|
||||
if idxName in ctx.btrees:
|
||||
var idxVals: seq[string] = @[]
|
||||
for c in eqConds: idxVals.add(c[1])
|
||||
let idxVal = idxVals.join("|")
|
||||
let entries = ctx.btrees[idxName].get(idxVal)
|
||||
if entries.len > 0:
|
||||
var rows: seq[Row] = @[]
|
||||
for entry in entries:
|
||||
let (found, val) = ctx.db.get(entry.lsmKey)
|
||||
if found:
|
||||
rows.add(parseRowData(cast[string](val)))
|
||||
let tbl = ctx.getTableDef(stmt.selFrom.fromTable)
|
||||
var cols: seq[string] = @[]
|
||||
for c in tbl.columns: cols.add(c.name)
|
||||
if cols.len == 0: cols = @["key", "value"]
|
||||
return okResult(rows, cols)
|
||||
if w.kind == nkBinOp and w.binOp == bkEq:
|
||||
if w.binLeft.kind == nkIdent and w.binRight.kind == nkStringLit:
|
||||
let colName = w.binLeft.identName
|
||||
@@ -1905,19 +1986,24 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
|
||||
return okResult(msg=msg)
|
||||
|
||||
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]()
|
||||
var colKey = stmt.ciTarget
|
||||
for col in stmt.ciColumns:
|
||||
colKey = colKey & "." & col
|
||||
let idxName = if stmt.ciName.len > 0: stmt.ciName else: colKey
|
||||
ctx.btrees[colKey] = 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: ""))
|
||||
var colVals: seq[string] = @[]
|
||||
for col in stmt.ciColumns:
|
||||
if col in row:
|
||||
colVals.add(row[col])
|
||||
else:
|
||||
colVals.add("")
|
||||
let idxVal = colVals.join("|")
|
||||
if idxVal.len > 0 and not isNull(idxVal):
|
||||
let lsmKey = if "$key" in row: stmt.ciTarget & "." & row["$key"] else: ""
|
||||
ctx.btrees[colKey].insert(idxVal, IndexEntry(lsmKey: lsmKey, rowValue: ""))
|
||||
return okResult(msg="CREATE INDEX " & idxName & " on " & stmt.ciTarget)
|
||||
|
||||
of nkCreateUser:
|
||||
|
||||
@@ -67,6 +67,7 @@ type
|
||||
tkCase
|
||||
tkWhen
|
||||
tkWith
|
||||
tkRecursive
|
||||
tkDistinct
|
||||
tkUnion
|
||||
tkIntersect
|
||||
@@ -231,6 +232,7 @@ const keywords*: Table[string, TokenKind] = {
|
||||
"case": tkCase,
|
||||
"when": tkWhen,
|
||||
"with": tkWith,
|
||||
"recursive": tkRecursive,
|
||||
"distinct": tkDistinct,
|
||||
"union": tkUnion,
|
||||
"intersect": tkIntersect,
|
||||
|
||||
@@ -275,18 +275,23 @@ proc parseJoinType(p: var Parser): JoinKind =
|
||||
return jkInner
|
||||
|
||||
proc parseWith(p: var Parser): Node =
|
||||
# WITH name AS (select), name2 AS (select2) SELECT ...
|
||||
# WITH [RECURSIVE] name AS (select), name2 AS (select2) SELECT ...
|
||||
let tok = p.expect(tkWith)
|
||||
result = Node(kind: nkWith, line: tok.line, col: tok.col)
|
||||
result.withBindings = @[]
|
||||
|
||||
var isRecursive = false
|
||||
if p.peek().kind == tkRecursive:
|
||||
discard p.advance()
|
||||
isRecursive = true
|
||||
|
||||
# Parse first CTE
|
||||
let cteName = p.expect(tkIdent).value
|
||||
discard p.expect(tkAs)
|
||||
discard p.expect(tkLParen)
|
||||
let cteQuery = p.parseSelect()
|
||||
discard p.expect(tkRParen)
|
||||
result.withBindings.add((cteName, cteQuery))
|
||||
result.withBindings.add((cteName, cteQuery, isRecursive))
|
||||
|
||||
# Parse additional CTEs
|
||||
while p.match(tkComma):
|
||||
@@ -295,7 +300,7 @@ proc parseWith(p: var Parser): Node =
|
||||
discard p.expect(tkLParen)
|
||||
let query = p.parseSelect()
|
||||
discard p.expect(tkRParen)
|
||||
result.withBindings.add((name, query))
|
||||
result.withBindings.add((name, query, isRecursive))
|
||||
|
||||
proc parseSelect(p: var Parser): Node =
|
||||
# Handle WITH (CTE)
|
||||
@@ -735,10 +740,13 @@ proc parseCreateIndex(p: var Parser): Node =
|
||||
discard p.match(tkOn)
|
||||
let tableName = p.expect(tkIdent).value
|
||||
discard p.match(tkLParen)
|
||||
let colName = p.expect(tkIdent).value
|
||||
var colNames: seq[string] = @[]
|
||||
colNames.add(p.expect(tkIdent).value)
|
||||
while p.match(tkComma):
|
||||
colNames.add(p.expect(tkIdent).value)
|
||||
discard p.match(tkRParen)
|
||||
result = Node(kind: nkCreateIndex, ciName: idxName, ciTarget: tableName,
|
||||
line: tok.line, col: tok.col)
|
||||
ciColumns: colNames, line: tok.line, col: tok.col)
|
||||
|
||||
proc parseBeginTxn(p: var Parser): Node =
|
||||
let tok = p.expect(tkBegin)
|
||||
|
||||
Reference in New Issue
Block a user