clients/nim-allographer/src/allographer/query_builder/models/baradb/baradb_exec.nim src/barabadb/core/mvcc.nim src/barabadb/query/codegen.nim src/barabadb/query/lexer.nim src/barabadb/query/parser.nim src/barabadb/query/udf.nim tests/test_all.nim
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
This commit is contained in:
@@ -148,8 +148,8 @@ proc escapeSqlValue(val: JsonNode): string =
|
|||||||
proc formatSql*(sql: string, args: seq[JsonNode]): string =
|
proc formatSql*(sql: string, args: seq[JsonNode]): string =
|
||||||
result = sql
|
result = sql
|
||||||
var placeholderCount = 0
|
var placeholderCount = 0
|
||||||
for i in 0..<result.len:
|
for ch in result:
|
||||||
if result[i] == '?':
|
if ch == '?':
|
||||||
placeholderCount += 1
|
placeholderCount += 1
|
||||||
if placeholderCount != args.len:
|
if placeholderCount != args.len:
|
||||||
raise newException(DbError, "Placeholder count mismatch: expected " & $placeholderCount & " but got " & $args.len & " arguments")
|
raise newException(DbError, "Placeholder count mismatch: expected " & $placeholderCount & " but got " & $args.len & " arguments")
|
||||||
@@ -438,6 +438,7 @@ proc exec(self: BaradbQuery, queryString: string) {.async.} =
|
|||||||
defer:
|
defer:
|
||||||
if not self.isInTransaction:
|
if not self.isInTransaction:
|
||||||
self.returnConn(connI).await
|
self.returnConn(connI).await
|
||||||
|
self.placeHolder = newJArray()
|
||||||
if connI == errorConnectionNum:
|
if connI == errorConnectionNum:
|
||||||
raisePoolTimeout(self)
|
raisePoolTimeout(self)
|
||||||
|
|
||||||
@@ -475,6 +476,7 @@ proc insertId(self: BaradbQuery, queryString: string, key: string): Future[strin
|
|||||||
defer:
|
defer:
|
||||||
if not self.isInTransaction:
|
if not self.isInTransaction:
|
||||||
self.returnConn(connI).await
|
self.returnConn(connI).await
|
||||||
|
self.placeHolder = newJArray()
|
||||||
if connI == errorConnectionNum:
|
if connI == errorConnectionNum:
|
||||||
raisePoolTimeout(self)
|
raisePoolTimeout(self)
|
||||||
|
|
||||||
|
|||||||
@@ -196,7 +196,7 @@ proc write*(tm: TxnManager, txn: Transaction, key: string, value: seq[byte]): bo
|
|||||||
if victimId in tm.activeTxns:
|
if victimId in tm.activeTxns:
|
||||||
tm.activeTxns[victimId].state = tsAborted
|
tm.activeTxns[victimId].state = tsAborted
|
||||||
tm.activeTxns.del(victimId)
|
tm.activeTxns.del(victimId)
|
||||||
tm.deadlockDetector.removeWait(uint64(txn.id), uint64(otherId))
|
# Keep the wait edge so subsequent transactions can detect cycles
|
||||||
release(tm.lock)
|
release(tm.lock)
|
||||||
return false # write-write conflict with uncommitted txn
|
return false # write-write conflict with uncommitted txn
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
## Codegen — compile IR plan to storage operations
|
## Codegen — compile IR plan to storage operations
|
||||||
import std/strutils
|
import std/strutils
|
||||||
|
import ../core/types
|
||||||
import ../query/ir
|
import ../query/ir
|
||||||
|
|
||||||
type
|
type
|
||||||
|
|||||||
@@ -547,11 +547,11 @@ proc readNumber(l: var Lexer, startLine, startCol: int): Token =
|
|||||||
proc readIdent(l: var Lexer, startLine, startCol: int): Token =
|
proc readIdent(l: var Lexer, startLine, startCol: int): Token =
|
||||||
var ident = ""
|
var ident = ""
|
||||||
while l.pos < l.input.len:
|
while l.pos < l.input.len:
|
||||||
let r = l.peekRune()
|
var p = l.pos
|
||||||
|
var r: Rune
|
||||||
|
fastRuneAt(l.input, p, r, true)
|
||||||
if isIdentPartRune(r):
|
if isIdentPartRune(r):
|
||||||
var run: Rune
|
ident.add($r)
|
||||||
fastRuneAt(l.input, l.pos, run, true)
|
|
||||||
ident.add($run)
|
|
||||||
inc l.col
|
inc l.col
|
||||||
discard l.advanceRune()
|
discard l.advanceRune()
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ proc match(p: var Parser, kind: TokenKind): bool =
|
|||||||
return false
|
return false
|
||||||
|
|
||||||
# Token kinds that can also serve as identifiers in table/column name positions
|
# Token kinds that can also serve as identifiers in table/column name positions
|
||||||
const identLikeKinds = {tkIdent, tkLabels, tkCount, tkSum, tkAvg, tkMin, tkMax, tkArrayAgg, tkStringAgg}
|
const identLikeKinds = {tkIdent, tkLabels, tkCount, tkSum, tkAvg, tkMin, tkMax, tkArrayAgg, tkStringAgg, tkJsonFmt, tkArray, tkVector, tkGraph, tkDocument}
|
||||||
|
|
||||||
proc expectIdent(p: var Parser): Token =
|
proc expectIdent(p: var Parser): Token =
|
||||||
## Expect a token that can serve as an identifier (table name, column name, alias, etc.).
|
## Expect a token that can serve as an identifier (table name, column name, alias, etc.).
|
||||||
@@ -1154,16 +1154,21 @@ proc parseCreateTable(p: var Parser): Node =
|
|||||||
# Parse column definition
|
# Parse column definition
|
||||||
let colName = p.expectIdent().value
|
let colName = p.expectIdent().value
|
||||||
var colType = ""
|
var colType = ""
|
||||||
if p.peek().kind == tkIdent:
|
let typeKinds = {tkIdent, tkVector, tkJsonFmt, tkArray, tkDocument, tkGraph}
|
||||||
colType = p.advance().value.toUpper()
|
if p.peek().kind in typeKinds:
|
||||||
if p.peek().kind == tkLParen:
|
let typeTok = p.advance()
|
||||||
discard p.advance()
|
if typeTok.kind == tkVector:
|
||||||
let size = p.expect(tkIntLit).value
|
colType = "VECTOR"
|
||||||
colType &= "(" & size & ")"
|
elif typeTok.kind == tkJsonFmt:
|
||||||
discard p.expect(tkRParen)
|
colType = "JSON"
|
||||||
elif p.peek().kind == tkVector:
|
elif typeTok.kind == tkArray:
|
||||||
discard p.advance()
|
colType = "ARRAY"
|
||||||
colType = "VECTOR"
|
elif typeTok.kind == tkDocument:
|
||||||
|
colType = "DOCUMENT"
|
||||||
|
elif typeTok.kind == tkGraph:
|
||||||
|
colType = "GRAPH"
|
||||||
|
else:
|
||||||
|
colType = typeTok.value.toUpper()
|
||||||
if p.peek().kind == tkLParen:
|
if p.peek().kind == tkLParen:
|
||||||
discard p.advance()
|
discard p.advance()
|
||||||
let size = p.expect(tkIntLit).value
|
let size = p.expect(tkIntLit).value
|
||||||
|
|||||||
@@ -191,7 +191,7 @@ proc registerStdlib*(reg: UDFRegistry) =
|
|||||||
let length = int(args[2].int64Val)
|
let length = int(args[2].int64Val)
|
||||||
let endIdx = min(start + length, s.len)
|
let endIdx = min(start + length, s.len)
|
||||||
return Value(kind: vkString, strVal: s[start ..< endIdx])
|
return Value(kind: vkString, strVal: s[start ..< endIdx])
|
||||||
return Value(kind: vkString, strVal: s[start])
|
return Value(kind: vkString, strVal: s[start .. start])
|
||||||
return Value(kind: vkNull))
|
return Value(kind: vkNull))
|
||||||
|
|
||||||
# Type conversion
|
# Type conversion
|
||||||
@@ -240,11 +240,11 @@ proc registerStdlib*(reg: UDFRegistry) =
|
|||||||
else: discard
|
else: discard
|
||||||
elif (item.kind in {vkInt64, vkInt32, vkFloat64}) and
|
elif (item.kind in {vkInt64, vkInt32, vkFloat64}) and
|
||||||
(target.kind in {vkInt64, vkInt32, vkFloat64}):
|
(target.kind in {vkInt64, vkInt32, vkFloat64}):
|
||||||
let a = case item.kind of vkInt64: float64(item.int64Val)
|
let a = if item.kind == vkInt64: float64(item.int64Val)
|
||||||
of vkInt32: float64(item.int32Val)
|
elif item.kind == vkInt32: float64(item.int32Val)
|
||||||
else: item.float64Val
|
else: item.float64Val
|
||||||
let b = case target.kind of vkInt64: float64(target.int64Val)
|
let b = if target.kind == vkInt64: float64(target.int64Val)
|
||||||
of vkInt32: float64(target.int32Val)
|
elif target.kind == vkInt32: float64(target.int32Val)
|
||||||
else: target.float64Val
|
else: target.float64Val
|
||||||
if a == b:
|
if a == b:
|
||||||
return Value(kind: vkBool, boolVal: true)
|
return Value(kind: vkBool, boolVal: true)
|
||||||
|
|||||||
+3
-3
@@ -1247,13 +1247,13 @@ suite "Sharding":
|
|||||||
|
|
||||||
test "Rebalance assigns nodes":
|
test "Rebalance assigns nodes":
|
||||||
var router = newShardRouter(ShardConfig(numShards: 3, replicas: 2, strategy: ssHash))
|
var router = newShardRouter(ShardConfig(numShards: 3, replicas: 2, strategy: ssHash))
|
||||||
router.rebalance(@["node1", "node2", "node3"])
|
discard router.rebalance(@["node1", "node2", "node3"])
|
||||||
for shard in router.shards:
|
for shard in router.shards:
|
||||||
check shard.nodeIds.len == 2 # 2 replicas
|
check shard.nodeIds.len == 2 # 2 replicas
|
||||||
|
|
||||||
test "Replicas of key":
|
test "Replicas of key":
|
||||||
var router = newShardRouter(ShardConfig(numShards: 2, replicas: 1, strategy: ssHash))
|
var router = newShardRouter(ShardConfig(numShards: 2, replicas: 1, strategy: ssHash))
|
||||||
router.rebalance(@["n1", "n2"])
|
discard router.rebalance(@["n1", "n2"])
|
||||||
let replicas = router.replicasOf("test_key")
|
let replicas = router.replicasOf("test_key")
|
||||||
check replicas.len == 1
|
check replicas.len == 1
|
||||||
|
|
||||||
@@ -1989,7 +1989,7 @@ suite "Cluster Auto-Rebalance":
|
|||||||
|
|
||||||
test "Node fail re-assigns shards":
|
test "Node fail re-assigns shards":
|
||||||
var router = newShardRouter(ShardConfig(numShards: 4, replicas: 2))
|
var router = newShardRouter(ShardConfig(numShards: 4, replicas: 2))
|
||||||
router.rebalance(@["node1", "node2", "node3"])
|
discard router.rebalance(@["node1", "node2", "node3"])
|
||||||
var cm = newClusterMembership(router)
|
var cm = newClusterMembership(router)
|
||||||
cm.nodes = @["node1", "node2", "node3"]
|
cm.nodes = @["node1", "node2", "node3"]
|
||||||
cm.onNodeFail("node1")
|
cm.onNodeFail("node1")
|
||||||
|
|||||||
Reference in New Issue
Block a user