Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 372e5cf627 | |||
| 57d2908066 |
@@ -0,0 +1,181 @@
|
|||||||
|
# BaraDB Bug Report — NimForum Integration (Stress Test)
|
||||||
|
|
||||||
|
**Date:** 2026-05-19
|
||||||
|
**Reporter:** NimForum integration team
|
||||||
|
**BaraDB Version:** Latest (docker image `nimforum-baradb`)
|
||||||
|
**Severity:** High — prevents real-world multi-table applications from working
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Summary
|
||||||
|
|
||||||
|
During integration of the NimForum application (a real-world web forum) with BaraDB via the native TCP wire protocol, we hit **three distinct SQL-layer bugs** in BaraDB's query executor/parser. All of them are related to multi-table (implicit join) queries. The bugs are severe enough that any application using standard SQL patterns (table aliases, duplicate column names, or `IN (list)` predicates) will receive incorrect results or crash.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Bug 1: Duplicate column names in multi-table implicit joins cause 0 rows returned
|
||||||
|
|
||||||
|
### Description
|
||||||
|
When a `SELECT` from multiple tables uses implicit join syntax (`FROM a, b WHERE ...`) and the selected columns include the **same column name from both tables** (e.g. `a.id` and `b.id`, or `a.name` and `b.name`) **without explicit `AS` aliases**, BaraDB returns **0 rows** even when matching data exists.
|
||||||
|
|
||||||
|
### Reproducer
|
||||||
|
|
||||||
|
```sql
|
||||||
|
-- Returns 0 rows (BUG)
|
||||||
|
SELECT thread.id, thread.name, category.id, category.name
|
||||||
|
FROM thread, category
|
||||||
|
WHERE thread.id = 3
|
||||||
|
AND thread.isDeleted = 0
|
||||||
|
AND thread.category = category.id;
|
||||||
|
|
||||||
|
-- Returns 1 correct row (WORKAROUND)
|
||||||
|
SELECT t.id AS thread_id, t.name AS thread_name,
|
||||||
|
c.id AS cat_id, c.name AS cat_name
|
||||||
|
FROM thread t, category c
|
||||||
|
WHERE t.id = 3
|
||||||
|
AND t.isDeleted = 0
|
||||||
|
AND t.category = c.id;
|
||||||
|
```
|
||||||
|
|
||||||
|
### Expected behavior
|
||||||
|
Both queries should return 1 row with the thread and category data.
|
||||||
|
|
||||||
|
### Actual behavior
|
||||||
|
The first query (without aliases) returns **0 rows**.
|
||||||
|
|
||||||
|
### Root cause hypothesis
|
||||||
|
BaraDB's executor or result-set builder appears to use column names as keys internally. When two selected columns have the same name (`id`, `name`), the second overwrites the first, corrupting the row metadata and causing the executor to discard the row.
|
||||||
|
|
||||||
|
### Impact
|
||||||
|
Any standard SQL query joining two tables on `id` columns (extremely common) will silently fail.
|
||||||
|
|
||||||
|
### Workaround used
|
||||||
|
Add explicit `AS` aliases to every column in multi-table selects.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Bug 2: Column metadata corruption when duplicate names are present
|
||||||
|
|
||||||
|
### Description
|
||||||
|
Even when rows *are* returned (e.g. in a multi-row list query without `WHERE id = ?`), if duplicate column names exist, the **column metadata and data values are scrambled**. Values from one column appear in another column's position.
|
||||||
|
|
||||||
|
### Reproducer
|
||||||
|
|
||||||
|
```sql
|
||||||
|
SELECT t.id, t.name, c.id, c.name, c.description, c.color
|
||||||
|
FROM thread t, category c
|
||||||
|
WHERE t.isDeleted = 0 AND t.category = c.id
|
||||||
|
ORDER BY modified DESC LIMIT 1;
|
||||||
|
```
|
||||||
|
|
||||||
|
### Expected behavior
|
||||||
|
| id | name | id | name | description | color |
|
||||||
|
|---|---|---|---|---|---|
|
||||||
|
| 3 | Test Thread2 | 1 | baradb | multimodal database engine written in Nim | 1a465b |
|
||||||
|
|
||||||
|
### Actual behavior
|
||||||
|
Column metadata returned by BaraDB:
|
||||||
|
```
|
||||||
|
Columns: ['id', 'name', 'views', "strftime('%s', \"modified\")", 'isLocked', 'isPinned', 'id', 'name', 'description', 'color']
|
||||||
|
Row[0]: ['3', 'Test Thread2', 0, '1779200191', <bool>, None, <bool>, None, '1', 'baradb']
|
||||||
|
```
|
||||||
|
Values are shifted/missing for the second table's columns.
|
||||||
|
|
||||||
|
### Impact
|
||||||
|
Client code receives corrupted data. In our case `category.description` became `"1"` and `category.color` became `"baradb"`, which would break UI rendering.
|
||||||
|
|
||||||
|
### Workaround used
|
||||||
|
Same as Bug 1: explicit `AS` aliases for every column.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Bug 3: `IN (val1, val2, ...)` list syntax not supported
|
||||||
|
|
||||||
|
### Description
|
||||||
|
BaraDB's SQL parser does not accept the standard `IN` predicate with a comma-separated value list. It throws a parse error.
|
||||||
|
|
||||||
|
### Reproducer
|
||||||
|
|
||||||
|
```sql
|
||||||
|
-- ERROR: Expected tkRParen but got tkComma at line 1
|
||||||
|
SELECT id, name, email
|
||||||
|
FROM person
|
||||||
|
WHERE id IN (2, 1);
|
||||||
|
```
|
||||||
|
|
||||||
|
### Expected behavior
|
||||||
|
Should return rows for `id = 2` and `id = 1`.
|
||||||
|
|
||||||
|
### Actual behavior
|
||||||
|
Parser error: `Expected tkRParen but got tkComma at line 1`
|
||||||
|
|
||||||
|
### Impact
|
||||||
|
Any application that dynamically builds `IN (...)` lists (e.g. looking up multiple users by ID) cannot function without rewriting every query.
|
||||||
|
|
||||||
|
### Workaround used
|
||||||
|
Replace `IN (2, 1)` with `id = 2 OR id = 1` generated dynamically in the application code.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Bug 4: `person.id` / `post.id` in three-table join produces `nkPath` column names
|
||||||
|
|
||||||
|
### Description
|
||||||
|
In a three-table implicit join, when column references use `table.column` syntax, BaraDB's parser sometimes produces corrupted column names like `"strftime('%s', nkPath)"` instead of `"strftime('%s', post.creation)"`.
|
||||||
|
|
||||||
|
### Reproducer
|
||||||
|
|
||||||
|
```sql
|
||||||
|
SELECT post.id, strftime('%s', post.creation), post.thread,
|
||||||
|
person.id, person.name, person.email,
|
||||||
|
strftime('%s', person.lastOnline),
|
||||||
|
strftime('%s', person.previousVisitAt), person.usrStatus,
|
||||||
|
person.isDeleted,
|
||||||
|
thread.name
|
||||||
|
FROM post, person, thread
|
||||||
|
WHERE post.thread = thread.id
|
||||||
|
AND post.author = person.id
|
||||||
|
AND post.id = 10;
|
||||||
|
```
|
||||||
|
|
||||||
|
### Actual column metadata returned
|
||||||
|
```
|
||||||
|
Columns: ['id', "strftime('%s', nkPath)", 'thread', 'id', 'name', 'email',
|
||||||
|
"strftime('%s', nkPath)", "strftime('%s', nkPath)",
|
||||||
|
'usrStatus', 'isDeleted', 'name']
|
||||||
|
```
|
||||||
|
|
||||||
|
### Impact
|
||||||
|
Result set metadata is unreadable. Client code cannot map columns to fields.
|
||||||
|
|
||||||
|
### Workaround used
|
||||||
|
Use `AS` aliases for every expression and column: `p.id AS post_id`, `strftime('%s', p.creation) AS post_creation`, etc.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## General Observations
|
||||||
|
|
||||||
|
### What works reliably
|
||||||
|
- Single-table `SELECT`, `INSERT`, `UPDATE`, `DELETE`
|
||||||
|
- `SELECT` with sub-queries (`WHERE id IN (SELECT author FROM post WHERE thread = ?)`)
|
||||||
|
- `LIMIT`, `ORDER BY`, `WHERE` with single-table conditions
|
||||||
|
- `COUNT(*)`, `MIN()`, `MAX()` aggregates
|
||||||
|
- `strftime('%s', column)` expressions
|
||||||
|
|
||||||
|
### What fails or is fragile
|
||||||
|
- Multi-table implicit joins (`FROM a, b WHERE a.id = b.id`) **without** `AS` aliases
|
||||||
|
- `IN (list)` with literal value lists
|
||||||
|
- `table.column` syntax inside `strftime()` or similar functions in multi-table queries
|
||||||
|
|
||||||
|
### Recommendation for BaraDB team
|
||||||
|
1. **Fix column deduplication in result sets** — the executor should not use raw column names as internal keys; it should use ordinal positions or alias names.
|
||||||
|
2. **Support `IN (val1, val2, ...)`** — extend the parser to accept comma-separated expressions inside `IN (...)`.
|
||||||
|
3. **Investigate `nkPath` leak in parser** — `post.creation` inside `strftime()` should not be tokenized into `nkPath` in the column metadata output.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Appendix: Test Environment
|
||||||
|
|
||||||
|
- **Client:** Custom synchronous Nim TCP client (`baradb_sync_client.nim`) using `net.Socket`
|
||||||
|
- **Wire protocol:** Binary protocol with `mkQuery` (0x02), `mkData` (0x82), `mkComplete` (0x83)
|
||||||
|
- **Result format:** `rfBinary` (0x00)
|
||||||
|
- **Connection:** Fresh socket per query (to rule out interleaving)
|
||||||
@@ -6,13 +6,13 @@ import hunos/context
|
|||||||
import json
|
import json
|
||||||
import tables
|
import tables
|
||||||
import strutils
|
import strutils
|
||||||
import os
|
|
||||||
import times
|
import times
|
||||||
import std/asyncdispatch
|
import std/asyncdispatch
|
||||||
import config
|
import config
|
||||||
import ../query/lexer
|
import ../query/lexer
|
||||||
import ../query/parser
|
import ../query/parser
|
||||||
import ../query/executor
|
import ../query/executor
|
||||||
|
import ../core/types
|
||||||
import ../storage/lsm
|
import ../storage/lsm
|
||||||
import ../core/mvcc
|
import ../core/mvcc
|
||||||
import ../protocol/wire
|
import ../protocol/wire
|
||||||
@@ -20,6 +20,7 @@ import ../core/websocket
|
|||||||
import jwt as jwtlib
|
import jwt as jwtlib
|
||||||
import ../protocol/auth
|
import ../protocol/auth
|
||||||
import ../protocol/ratelimit
|
import ../protocol/ratelimit
|
||||||
|
import ../core/registry
|
||||||
|
|
||||||
type
|
type
|
||||||
HttpServer* = ref object
|
HttpServer* = ref object
|
||||||
@@ -27,6 +28,7 @@ type
|
|||||||
running: bool
|
running: bool
|
||||||
db*: LSMTree
|
db*: LSMTree
|
||||||
ctx: ExecutionContext
|
ctx: ExecutionContext
|
||||||
|
registry*: DatabaseRegistry
|
||||||
metrics*: Metrics
|
metrics*: Metrics
|
||||||
secretKey*: string
|
secretKey*: string
|
||||||
authManager*: AuthManager
|
authManager*: AuthManager
|
||||||
@@ -40,8 +42,10 @@ type
|
|||||||
selectCount*: int
|
selectCount*: int
|
||||||
activeConnections*: int
|
activeConnections*: int
|
||||||
|
|
||||||
proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer =
|
proc newHttpServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): HttpServer =
|
||||||
let ctx = newExecutionContext(db)
|
let dbInfo = getOrCreateDatabase(registry, "default")
|
||||||
|
let db = dbInfo.db
|
||||||
|
let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx))
|
||||||
ctx.txnManager = newTxnManager()
|
ctx.txnManager = newTxnManager()
|
||||||
let secret = config.getEffectiveJwtSecret()
|
let secret = config.getEffectiveJwtSecret()
|
||||||
let ws = newWsServer(config, secret)
|
let ws = newWsServer(config, secret)
|
||||||
@@ -51,15 +55,27 @@ proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer =
|
|||||||
asyncCheck ws.broadcastToTable(ev.table, msg)
|
asyncCheck ws.broadcastToTable(ev.table, msg)
|
||||||
let am = newAuthManager(secret)
|
let am = newAuthManager(secret)
|
||||||
HttpServer(config: config, running: false, db: db, ctx: ctx,
|
HttpServer(config: config, running: false, db: db, ctx: ctx,
|
||||||
|
registry: registry,
|
||||||
secretKey: secret,
|
secretKey: secret,
|
||||||
authManager: am,
|
authManager: am,
|
||||||
rateLimiter: rl,
|
rateLimiter: rl,
|
||||||
metrics: Metrics(), ws: ws)
|
metrics: Metrics(), ws: ws)
|
||||||
|
|
||||||
|
proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer =
|
||||||
|
let registry = newDatabaseRegistry(config)
|
||||||
|
registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} =
|
||||||
|
cast[ContextRef](cast[pointer](newExecutionContext(d, r))))
|
||||||
|
let ctx = newExecutionContext(db, registry)
|
||||||
|
registry.setDatabase("default", db, cast[ContextRef](cast[pointer](ctx)))
|
||||||
|
return newHttpServerWithRegistry(config, registry)
|
||||||
|
|
||||||
proc newHttpServer*(config: BaraConfig): HttpServer =
|
proc newHttpServer*(config: BaraConfig): HttpServer =
|
||||||
let dataDir = config.dataDir / "server"
|
let registry = newDatabaseRegistry(config)
|
||||||
let db = newLSMTree(dataDir)
|
registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} =
|
||||||
return newHttpServerWithDb(config, db)
|
cast[ContextRef](cast[pointer](newExecutionContext(d, r))))
|
||||||
|
registry.loadExistingDatabases()
|
||||||
|
registry.ensureDefaultDatabase()
|
||||||
|
return newHttpServerWithRegistry(config, registry)
|
||||||
|
|
||||||
# ----------------------------------------------------------------------
|
# ----------------------------------------------------------------------
|
||||||
# JWT helpers
|
# JWT helpers
|
||||||
@@ -178,8 +194,8 @@ proc queryHandler(server: HttpServer): RequestHandler =
|
|||||||
var jsonRow = newJObject()
|
var jsonRow = newJObject()
|
||||||
for col in res.columns:
|
for col in res.columns:
|
||||||
let key = col
|
let key = col
|
||||||
if key in row and not isNull(row[key]):
|
if key in row and row[key].kind != vkNull:
|
||||||
jsonRow[key] = %row[key]
|
jsonRow[key] = %valueToString(row[key])
|
||||||
else:
|
else:
|
||||||
jsonRow[key] = newJNull()
|
jsonRow[key] = newJNull()
|
||||||
jsonRows.add(jsonRow)
|
jsonRows.add(jsonRow)
|
||||||
@@ -564,4 +580,7 @@ proc run*(server: HttpServer, port: int = 9470) =
|
|||||||
proc stop*(server: HttpServer) =
|
proc stop*(server: HttpServer) =
|
||||||
server.running = false
|
server.running = false
|
||||||
server.ws.stop()
|
server.ws.stop()
|
||||||
server.db.close()
|
if server.registry != nil:
|
||||||
|
server.registry.closeAll()
|
||||||
|
else:
|
||||||
|
server.db.close()
|
||||||
|
|||||||
@@ -25,6 +25,7 @@ type
|
|||||||
xmin*: TxnId # creating transaction
|
xmin*: TxnId # creating transaction
|
||||||
xmax*: TxnId # deleting transaction (0 = not deleted)
|
xmax*: TxnId # deleting transaction (0 = not deleted)
|
||||||
lsn*: Lsn
|
lsn*: Lsn
|
||||||
|
isDelete*: bool
|
||||||
|
|
||||||
Transaction* = ref object
|
Transaction* = ref object
|
||||||
id*: TxnId
|
id*: TxnId
|
||||||
@@ -147,7 +148,7 @@ proc read*(tm: TxnManager, txn: Transaction, key: string): (bool, seq[byte]) =
|
|||||||
if key in txn.writeSet:
|
if key in txn.writeSet:
|
||||||
let version = txn.writeSet[key]
|
let version = txn.writeSet[key]
|
||||||
release(tm.lock)
|
release(tm.lock)
|
||||||
if version.value == @[]:
|
if version.isDelete or version.value == @[]:
|
||||||
return (false, @[])
|
return (false, @[])
|
||||||
return (true, version.value)
|
return (true, version.value)
|
||||||
|
|
||||||
@@ -230,7 +231,61 @@ proc write*(tm: TxnManager, txn: Transaction, key: string, value: seq[byte]): bo
|
|||||||
return true
|
return true
|
||||||
|
|
||||||
proc delete*(tm: TxnManager, txn: Transaction, key: string): bool =
|
proc delete*(tm: TxnManager, txn: Transaction, key: string): bool =
|
||||||
return tm.write(txn, key, @[])
|
acquire(tm.lock)
|
||||||
|
if txn.state != tsActive:
|
||||||
|
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 against other active transactions' write sets
|
||||||
|
for otherId, otherTxn in tm.activeTxns:
|
||||||
|
if otherId != txn.id and otherTxn.state == tsActive:
|
||||||
|
if key in otherTxn.writeSet:
|
||||||
|
tm.deadlockDetector.addWait(uint64(txn.id), uint64(otherId))
|
||||||
|
if tm.deadlockDetector.hasDeadlock():
|
||||||
|
let victimId = TxnId(tm.deadlockDetector.findDeadlockVictim())
|
||||||
|
tm.deadlockDetector.removeTxn(uint64(victimId))
|
||||||
|
if victimId in tm.activeTxns:
|
||||||
|
tm.activeTxns[victimId].state = tsAborted
|
||||||
|
tm.activeTxns.del(victimId)
|
||||||
|
release(tm.lock)
|
||||||
|
return false
|
||||||
|
|
||||||
|
# Check for write-write conflict with committed versions
|
||||||
|
if key notin txn.writeSet:
|
||||||
|
if key in tm.globalVersions:
|
||||||
|
for version in tm.globalVersions[key]:
|
||||||
|
if version.xmax == TxnId(0):
|
||||||
|
let creator = version.xmin
|
||||||
|
if creator != txn.id:
|
||||||
|
if creator in txn.snapshotTxns or uint64(creator) > uint64(txn.snapshotMaxTxn):
|
||||||
|
if creator in tm.activeTxns:
|
||||||
|
if tm.activeTxns[creator].state == tsCommitted:
|
||||||
|
release(tm.lock)
|
||||||
|
return false
|
||||||
|
else:
|
||||||
|
release(tm.lock)
|
||||||
|
return false
|
||||||
|
|
||||||
|
let lsn = tm.allocLsn()
|
||||||
|
let version = VersionedRecord(
|
||||||
|
key: key,
|
||||||
|
value: @[],
|
||||||
|
xmin: txn.id,
|
||||||
|
xmax: TxnId(0),
|
||||||
|
lsn: lsn,
|
||||||
|
isDelete: true,
|
||||||
|
)
|
||||||
|
txn.writeSet[key] = version
|
||||||
|
release(tm.lock)
|
||||||
|
return true
|
||||||
|
|
||||||
proc compactVersions*(tm: TxnManager)
|
proc compactVersions*(tm: TxnManager)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,192 @@
|
|||||||
|
## BaraDB Database Registry — manages per-database LSMTree instances
|
||||||
|
import std/tables
|
||||||
|
import std/os
|
||||||
|
import std/strutils
|
||||||
|
import std/locks
|
||||||
|
import std/algorithm
|
||||||
|
import logging
|
||||||
|
import config
|
||||||
|
import ../storage/lsm
|
||||||
|
|
||||||
|
type
|
||||||
|
ContextRef* = ref RootObj
|
||||||
|
|
||||||
|
ContextFactory* = proc(db: LSMTree, reg: DatabaseRegistry): ContextRef {.closure.}
|
||||||
|
|
||||||
|
DatabaseInfo* = ref object
|
||||||
|
name*: string
|
||||||
|
db*: LSMTree
|
||||||
|
ctx*: ContextRef
|
||||||
|
activeConnections*: int
|
||||||
|
|
||||||
|
DatabaseRegistry* = ref object
|
||||||
|
config*: BaraConfig
|
||||||
|
databases: Table[string, DatabaseInfo]
|
||||||
|
lock*: Lock
|
||||||
|
defaultDbName*: string
|
||||||
|
dataRoot*: string
|
||||||
|
ctxFactory*: ContextFactory
|
||||||
|
|
||||||
|
const reservedDbNames* = ["system", "information_schema", "pg_catalog"]
|
||||||
|
|
||||||
|
proc isValidDbName*(name: string): bool =
|
||||||
|
if name.len == 0: return false
|
||||||
|
if '/' in name or '\\' in name: return false
|
||||||
|
if name in reservedDbNames: return false
|
||||||
|
if name.startsWith("_"): return false
|
||||||
|
if name == ".." or name == ".": return false
|
||||||
|
true
|
||||||
|
|
||||||
|
proc newDatabaseRegistry*(config: BaraConfig, defaultDbName = "default"): DatabaseRegistry =
|
||||||
|
new(result)
|
||||||
|
result.config = config
|
||||||
|
result.databases = initTable[string, DatabaseInfo]()
|
||||||
|
result.defaultDbName = defaultDbName
|
||||||
|
result.dataRoot = config.dataDir / "databases"
|
||||||
|
initLock(result.lock)
|
||||||
|
|
||||||
|
# Create root directory
|
||||||
|
if not dirExists(result.dataRoot):
|
||||||
|
createDir(result.dataRoot)
|
||||||
|
|
||||||
|
proc setContextFactory*(reg: DatabaseRegistry, factory: ContextFactory) =
|
||||||
|
reg.ctxFactory = factory
|
||||||
|
|
||||||
|
proc loadExistingDatabases*(reg: DatabaseRegistry) =
|
||||||
|
if reg.ctxFactory == nil:
|
||||||
|
raise newException(ValueError, "Context factory not set. Call setContextFactory first.")
|
||||||
|
|
||||||
|
# Scan for existing databases
|
||||||
|
for kind, path in walkDir(reg.dataRoot):
|
||||||
|
if kind == pcDir:
|
||||||
|
let dbName = path.splitPath().tail
|
||||||
|
if dbName.len > 0 and isValidDbName(dbName):
|
||||||
|
let dbDir = reg.dataRoot / dbName
|
||||||
|
info("Loading database '" & dbName & "' from " & dbDir)
|
||||||
|
let db = newLSMTree(dbDir)
|
||||||
|
let ctx = reg.ctxFactory(db, reg)
|
||||||
|
reg.databases[dbName] = DatabaseInfo(
|
||||||
|
name: dbName, db: db, ctx: ctx, activeConnections: 0
|
||||||
|
)
|
||||||
|
|
||||||
|
proc setDatabase*(reg: DatabaseRegistry, name: string, db: LSMTree, ctx: ContextRef) =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
reg.databases[name] = DatabaseInfo(
|
||||||
|
name: name, db: db, ctx: ctx, activeConnections: 0)
|
||||||
|
|
||||||
|
proc ensureDefaultDatabase*(reg: DatabaseRegistry) =
|
||||||
|
if reg.ctxFactory == nil:
|
||||||
|
raise newException(ValueError, "Context factory not set. Call setContextFactory first.")
|
||||||
|
|
||||||
|
let defaultDbName = reg.defaultDbName
|
||||||
|
if defaultDbName notin reg.databases:
|
||||||
|
let dbDir = reg.dataRoot / defaultDbName
|
||||||
|
info("Creating default database at " & dbDir)
|
||||||
|
let db = newLSMTree(dbDir)
|
||||||
|
let ctx = reg.ctxFactory(db, reg)
|
||||||
|
reg.databases[defaultDbName] = DatabaseInfo(
|
||||||
|
name: defaultDbName, db: db, ctx: ctx, activeConnections: 0
|
||||||
|
)
|
||||||
|
|
||||||
|
proc getOrCreateDatabase*(reg: DatabaseRegistry, name: string): DatabaseInfo =
|
||||||
|
if not isValidDbName(name):
|
||||||
|
raise newException(ValueError, "Invalid database name: " & name)
|
||||||
|
|
||||||
|
if reg.ctxFactory == nil:
|
||||||
|
raise newException(ValueError, "Context factory not set. Call setContextFactory first.")
|
||||||
|
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
|
||||||
|
if name in reg.databases:
|
||||||
|
return reg.databases[name]
|
||||||
|
|
||||||
|
# Create new database
|
||||||
|
let dbDir = reg.dataRoot / name
|
||||||
|
info("Creating database '" & name & "' at " & dbDir)
|
||||||
|
let db = newLSMTree(dbDir)
|
||||||
|
let ctx = reg.ctxFactory(db, reg)
|
||||||
|
let info = DatabaseInfo(name: name, db: db, ctx: ctx, activeConnections: 0)
|
||||||
|
reg.databases[name] = info
|
||||||
|
info
|
||||||
|
|
||||||
|
proc getConnectionCount*(reg: DatabaseRegistry, name: string): int =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
if name in reg.databases:
|
||||||
|
return reg.databases[name].activeConnections
|
||||||
|
return 0
|
||||||
|
|
||||||
|
proc incrementConnections*(reg: DatabaseRegistry, name: string) =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
if name in reg.databases:
|
||||||
|
inc reg.databases[name].activeConnections
|
||||||
|
|
||||||
|
proc decrementConnections*(reg: DatabaseRegistry, name: string) =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
if name in reg.databases and reg.databases[name].activeConnections > 0:
|
||||||
|
dec reg.databases[name].activeConnections
|
||||||
|
|
||||||
|
proc dropDatabase*(reg: DatabaseRegistry, name: string): bool =
|
||||||
|
if not isValidDbName(name):
|
||||||
|
return false
|
||||||
|
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
|
||||||
|
if name notin reg.databases:
|
||||||
|
return false
|
||||||
|
|
||||||
|
if name == reg.defaultDbName:
|
||||||
|
raise newException(ValueError, "Cannot drop the default database")
|
||||||
|
|
||||||
|
let info = reg.databases[name]
|
||||||
|
if info.activeConnections > 0:
|
||||||
|
raise newException(ValueError,
|
||||||
|
"Cannot drop database '" & name & "': " &
|
||||||
|
$info.activeConnections & " active connections")
|
||||||
|
|
||||||
|
# Close LSMTree
|
||||||
|
info.db.close()
|
||||||
|
|
||||||
|
# Remove data directory
|
||||||
|
let dbDir = reg.dataRoot / name
|
||||||
|
if dirExists(dbDir):
|
||||||
|
removeDir(dbDir)
|
||||||
|
|
||||||
|
reg.databases.del(name)
|
||||||
|
true
|
||||||
|
|
||||||
|
proc listDatabases*(reg: DatabaseRegistry): seq[string] =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
result = @[]
|
||||||
|
for name in reg.databases.keys:
|
||||||
|
result.add(name)
|
||||||
|
result.sort()
|
||||||
|
|
||||||
|
proc databaseExists*(reg: DatabaseRegistry, name: string): bool =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
name in reg.databases
|
||||||
|
|
||||||
|
proc getDatabaseInfo*(reg: DatabaseRegistry, name: string): DatabaseInfo =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
if name in reg.databases:
|
||||||
|
return reg.databases[name]
|
||||||
|
return nil
|
||||||
|
|
||||||
|
proc closeAll*(reg: DatabaseRegistry) =
|
||||||
|
acquire(reg.lock)
|
||||||
|
defer: release(reg.lock)
|
||||||
|
for name, info in reg.databases.pairs:
|
||||||
|
try:
|
||||||
|
info.db.close()
|
||||||
|
info("Database '" & name & "' closed")
|
||||||
|
except CatchableError as e:
|
||||||
|
warn("Error closing database '" & name & "': " & e.msg)
|
||||||
|
reg.databases.clear()
|
||||||
@@ -4,7 +4,6 @@ import std/asyncnet
|
|||||||
import std/strutils
|
import std/strutils
|
||||||
import std/sequtils
|
import std/sequtils
|
||||||
import std/tables
|
import std/tables
|
||||||
import std/os
|
|
||||||
import std/endians
|
import std/endians
|
||||||
import std/monotimes
|
import std/monotimes
|
||||||
import std/locks
|
import std/locks
|
||||||
@@ -28,6 +27,7 @@ import ../core/replication
|
|||||||
import ../core/sharding
|
import ../core/sharding
|
||||||
import ../core/gossip
|
import ../core/gossip
|
||||||
import ../protocol/ratelimit
|
import ../protocol/ratelimit
|
||||||
|
import ../core/registry
|
||||||
import jwt as jwtlib
|
import jwt as jwtlib
|
||||||
|
|
||||||
type
|
type
|
||||||
@@ -36,6 +36,7 @@ type
|
|||||||
running*: bool
|
running*: bool
|
||||||
db*: LSMTree
|
db*: LSMTree
|
||||||
ctx*: ExecutionContext
|
ctx*: ExecutionContext
|
||||||
|
registry*: DatabaseRegistry
|
||||||
txnManager*: TxnManager
|
txnManager*: TxnManager
|
||||||
distTxnManager*: DistTxnManager
|
distTxnManager*: DistTxnManager
|
||||||
replicationManager*: ReplicationManager
|
replicationManager*: ReplicationManager
|
||||||
@@ -47,8 +48,10 @@ type
|
|||||||
activeConnections*: int
|
activeConnections*: int
|
||||||
activeConnectionsLock*: Lock
|
activeConnectionsLock*: Lock
|
||||||
|
|
||||||
proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
|
proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Server =
|
||||||
let ctx = newExecutionContext(db)
|
let dbInfo = getOrCreateDatabase(registry, "default")
|
||||||
|
let db = dbInfo.db
|
||||||
|
let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx))
|
||||||
ctx.txnManager = newTxnManager()
|
ctx.txnManager = newTxnManager()
|
||||||
var tls: TLSContext = nil
|
var tls: TLSContext = nil
|
||||||
if config.tlsEnabled and config.certFile.len > 0 and config.keyFile.len > 0:
|
if config.tlsEnabled and config.certFile.len > 0 and config.keyFile.len > 0:
|
||||||
@@ -60,7 +63,7 @@ proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
|
|||||||
let localId = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
|
let localId = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
|
||||||
let cm = newClusterMembership(shardRouter, localId)
|
let cm = newClusterMembership(shardRouter, localId)
|
||||||
|
|
||||||
# Wire shard migration callbacks to LSM
|
# Wire shard migration callbacks to LSM (use default database)
|
||||||
shardRouter.iterateKeys = proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.} =
|
shardRouter.iterateKeys = proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.} =
|
||||||
var entries: seq[(string, seq[byte])] = @[]
|
var entries: seq[(string, seq[byte])] = @[]
|
||||||
for (key, value) in db.scanAll():
|
for (key, value) in db.scanAll():
|
||||||
@@ -94,6 +97,7 @@ proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
|
|||||||
let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
|
let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
|
||||||
|
|
||||||
result = Server(config: config, running: false, db: db, ctx: ctx,
|
result = Server(config: config, running: false, db: db, ctx: ctx,
|
||||||
|
registry: registry,
|
||||||
txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
|
txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
|
||||||
replicationManager: newReplicationManager(),
|
replicationManager: newReplicationManager(),
|
||||||
shardRouter: shardRouter,
|
shardRouter: shardRouter,
|
||||||
@@ -103,10 +107,22 @@ proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
|
|||||||
rateLimiter: rl)
|
rateLimiter: rl)
|
||||||
initLock(result.activeConnectionsLock)
|
initLock(result.activeConnectionsLock)
|
||||||
|
|
||||||
|
proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
|
||||||
|
let registry = newDatabaseRegistry(config)
|
||||||
|
let ctx = newExecutionContext(db, registry)
|
||||||
|
registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} =
|
||||||
|
cast[ContextRef](cast[pointer](newExecutionContext(d, r))))
|
||||||
|
# Use the existing db for default
|
||||||
|
registry.setDatabase("default", db, cast[ContextRef](cast[pointer](ctx)))
|
||||||
|
return newServerWithRegistry(config, registry)
|
||||||
|
|
||||||
proc newServer*(config: BaraConfig): Server =
|
proc newServer*(config: BaraConfig): Server =
|
||||||
let dataDir = config.dataDir / "server"
|
let registry = newDatabaseRegistry(config)
|
||||||
let db = newLSMTree(dataDir)
|
registry.setContextFactory(proc(d: LSMTree, r: DatabaseRegistry): ContextRef {.closure.} =
|
||||||
return newServerWithDb(config, db)
|
cast[ContextRef](cast[pointer](newExecutionContext(d, r))))
|
||||||
|
registry.loadExistingDatabases()
|
||||||
|
registry.ensureDefaultDatabase()
|
||||||
|
return newServerWithRegistry(config, registry)
|
||||||
|
|
||||||
# ----------------------------------------------------------------------
|
# ----------------------------------------------------------------------
|
||||||
# Wire Protocol Helpers
|
# Wire Protocol Helpers
|
||||||
@@ -230,7 +246,7 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
|
|||||||
for row in res.rows:
|
for row in res.rows:
|
||||||
var wireRow: seq[WireValue] = @[]
|
var wireRow: seq[WireValue] = @[]
|
||||||
for i, col in res.columns:
|
for i, col in res.columns:
|
||||||
let val = if col in row: row[col] else: "\\N"
|
let val = if col in row: valueToString(row[col]) else: "\\N"
|
||||||
let cType = if i < colTypes.len: colTypes[i] else: ""
|
let cType = if i < colTypes.len: colTypes[i] else: ""
|
||||||
wireRow.add(valueToWire(val, cType))
|
wireRow.add(valueToWire(val, cType))
|
||||||
qr.rows.add(wireRow)
|
qr.rows.add(wireRow)
|
||||||
@@ -312,16 +328,17 @@ proc slowQueryLog(logPath: string, query: string, durationMs: int, clientId: int
|
|||||||
f.write(line)
|
f.write(line)
|
||||||
except IOError: discard
|
except IOError: discard
|
||||||
|
|
||||||
proc verifyToken(secret, tokenStr: string): (bool, string, string) =
|
proc verifyToken(secret, tokenStr: string): (bool, string, string, string) =
|
||||||
try:
|
try:
|
||||||
let token = tokenStr.toJWT()
|
let token = tokenStr.toJWT()
|
||||||
if not token.verify(secret, HS256):
|
if not token.verify(secret, HS256):
|
||||||
return (false, "", "")
|
return (false, "", "", "")
|
||||||
let userId = token.claims["sub"].node.str
|
let userId = token.claims["sub"].node.str
|
||||||
let role = if "role" in token.claims: token.claims["role"].node.str else: "user"
|
let role = if "role" in token.claims: token.claims["role"].node.str else: "user"
|
||||||
return (true, userId, role)
|
let database = if "database" in token.claims: token.claims["database"].node.str else: ""
|
||||||
|
return (true, userId, role, database)
|
||||||
except ValueError, KeyError:
|
except ValueError, KeyError:
|
||||||
return (false, "", "")
|
return (false, "", "", "")
|
||||||
|
|
||||||
proc recvWithTimeout(client: AsyncSocket, size: int, timeoutMs: int): Future[string] {.async.} =
|
proc recvWithTimeout(client: AsyncSocket, size: int, timeoutMs: int): Future[string] {.async.} =
|
||||||
if timeoutMs <= 0:
|
if timeoutMs <= 0:
|
||||||
@@ -475,7 +492,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
|||||||
case header.kind
|
case header.kind
|
||||||
of mkAuth:
|
of mkAuth:
|
||||||
let tokenStr = parseAuthMessage(stringToBytes(payload))
|
let tokenStr = parseAuthMessage(stringToBytes(payload))
|
||||||
let (valid, userId, role) = verifyToken(secret, tokenStr)
|
let (valid, userId, role, jwtDatabase) = verifyToken(secret, tokenStr)
|
||||||
if valid:
|
if valid:
|
||||||
authenticated = true
|
authenticated = true
|
||||||
connCtx.currentUser = userId
|
connCtx.currentUser = userId
|
||||||
@@ -483,6 +500,24 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
|||||||
let okMsg = makeAuthOkMessage(header.requestId)
|
let okMsg = makeAuthOkMessage(header.requestId)
|
||||||
await client.send(bytesToString(okMsg))
|
await client.send(bytesToString(okMsg))
|
||||||
info("Client " & $clientId & " authenticated as " & userId)
|
info("Client " & $clientId & " authenticated as " & userId)
|
||||||
|
# Switch to database from JWT claim if provided
|
||||||
|
if jwtDatabase.len > 0 and server.registry != nil and isValidDbName(jwtDatabase):
|
||||||
|
let info = getDatabaseInfo(server.registry, jwtDatabase)
|
||||||
|
if info != nil:
|
||||||
|
let targetCtx = cast[ExecutionContext](cast[pointer](info.ctx))
|
||||||
|
connCtx.db = info.db
|
||||||
|
connCtx.tables = targetCtx.tables
|
||||||
|
connCtx.btrees = targetCtx.btrees
|
||||||
|
connCtx.views = targetCtx.views
|
||||||
|
connCtx.ftsIndexes = targetCtx.ftsIndexes
|
||||||
|
connCtx.vectorIndexes = targetCtx.vectorIndexes
|
||||||
|
connCtx.users = targetCtx.users
|
||||||
|
connCtx.policies = targetCtx.policies
|
||||||
|
connCtx.graphs = targetCtx.graphs
|
||||||
|
connCtx.autoIncCounters = targetCtx.autoIncCounters
|
||||||
|
connCtx.sequences = targetCtx.sequences
|
||||||
|
connCtx.currentDatabase = jwtDatabase
|
||||||
|
incrementConnections(server.registry, jwtDatabase)
|
||||||
else:
|
else:
|
||||||
let err = makeErrorMessage(header.requestId, 403, "Invalid token")
|
let err = makeErrorMessage(header.requestId, 403, "Invalid token")
|
||||||
await client.send(bytesToString(err))
|
await client.send(bytesToString(err))
|
||||||
@@ -510,7 +545,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
|||||||
|
|
||||||
if shardCheck:
|
if shardCheck:
|
||||||
let startTicks = getMonoTime().ticks()
|
let startTicks = getMonoTime().ticks()
|
||||||
let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, replication=server.replicationManager)
|
let (success, result, errorMsg) = executeQuery(connCtx.db, connCtx, queryStr, replication=server.replicationManager)
|
||||||
let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
|
let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
|
||||||
|
|
||||||
if durationMs >= slowThreshold:
|
if durationMs >= slowThreshold:
|
||||||
@@ -530,7 +565,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
|||||||
info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)")
|
info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)")
|
||||||
|
|
||||||
let startTicks = getMonoTime().ticks()
|
let startTicks = getMonoTime().ticks()
|
||||||
let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, params, replication=server.replicationManager)
|
let (success, result, errorMsg) = executeQuery(connCtx.db, connCtx, queryStr, params, replication=server.replicationManager)
|
||||||
let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
|
let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
|
||||||
|
|
||||||
if durationMs >= slowThreshold:
|
if durationMs >= slowThreshold:
|
||||||
@@ -562,6 +597,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
errorMsg("Client " & $clientId & " error: " & e.msg)
|
errorMsg("Client " & $clientId & " error: " & e.msg)
|
||||||
finally:
|
finally:
|
||||||
|
# Decrement database connection counter
|
||||||
|
if server.registry != nil and connCtx.currentDatabase.len > 0:
|
||||||
|
decrementConnections(server.registry, connCtx.currentDatabase)
|
||||||
acquire(server.activeConnectionsLock)
|
acquire(server.activeConnectionsLock)
|
||||||
try:
|
try:
|
||||||
if server.activeConnections > 0:
|
if server.activeConnections > 0:
|
||||||
@@ -629,4 +667,7 @@ proc stop*(server: Server) =
|
|||||||
server.running = false
|
server.running = false
|
||||||
if server.gossipProtocol != nil:
|
if server.gossipProtocol != nil:
|
||||||
server.gossipProtocol.stop()
|
server.gossipProtocol.stop()
|
||||||
server.db.close()
|
if server.registry != nil:
|
||||||
|
server.registry.closeAll()
|
||||||
|
else:
|
||||||
|
server.db.close()
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import std/times
|
import std/times
|
||||||
import std/oids
|
import std/oids
|
||||||
import std/monotimes
|
import std/monotimes
|
||||||
|
import std/strutils
|
||||||
|
|
||||||
type
|
type
|
||||||
ValueKind* = enum
|
ValueKind* = enum
|
||||||
@@ -124,3 +125,60 @@ proc newRecordId*(): RecordId =
|
|||||||
|
|
||||||
proc `==`*(a, b: RecordId): bool {.borrow.}
|
proc `==`*(a, b: RecordId): bool {.borrow.}
|
||||||
proc `$`*(r: RecordId): string = $uint64(r)
|
proc `$`*(r: RecordId): string = $uint64(r)
|
||||||
|
|
||||||
|
proc `==`*(a, b: Value): bool {.noSideEffect.} =
|
||||||
|
if a.kind != b.kind: return false
|
||||||
|
case a.kind
|
||||||
|
of vkNull: return true
|
||||||
|
of vkBool: return a.boolVal == b.boolVal
|
||||||
|
of vkInt8: return a.int8Val == b.int8Val
|
||||||
|
of vkInt16: return a.int16Val == b.int16Val
|
||||||
|
of vkInt32: return a.int32Val == b.int32Val
|
||||||
|
of vkInt64: return a.int64Val == b.int64Val
|
||||||
|
of vkFloat32: return a.float32Val == b.float32Val
|
||||||
|
of vkFloat64: return a.float64Val == b.float64Val
|
||||||
|
of vkString: return a.strVal == b.strVal
|
||||||
|
of vkBytes: return a.bytesVal == b.bytesVal
|
||||||
|
of vkUuid: return a.uuidVal == b.uuidVal
|
||||||
|
of vkDateTime: return false # DateTime comparison not supported without side effects
|
||||||
|
of vkJson: return a.jsonVal == b.jsonVal
|
||||||
|
of vkArray: return false # Recursive comparison not supported
|
||||||
|
of vkObject: return false # Recursive comparison not supported
|
||||||
|
of vkVector: return a.vecVal == b.vecVal
|
||||||
|
|
||||||
|
proc `!=`*(a, b: Value): bool {.noSideEffect.} =
|
||||||
|
return not (a == b)
|
||||||
|
|
||||||
|
proc `$`*(v: Value): string =
|
||||||
|
case v.kind
|
||||||
|
of vkNull: return "\\N"
|
||||||
|
of vkBool: return $v.boolVal
|
||||||
|
of vkInt8: return $v.int8Val
|
||||||
|
of vkInt16: return $v.int16Val
|
||||||
|
of vkInt32: return $v.int32Val
|
||||||
|
of vkInt64: return $v.int64Val
|
||||||
|
of vkFloat32: return $v.float32Val
|
||||||
|
of vkFloat64: return $v.float64Val
|
||||||
|
of vkString: return v.strVal
|
||||||
|
of vkBytes: return "<bytes>"
|
||||||
|
of vkUuid: return $v.uuidVal
|
||||||
|
of vkDateTime: return $v.dtVal
|
||||||
|
of vkJson: return v.jsonVal
|
||||||
|
of vkArray: return $v.arrayVal
|
||||||
|
of vkObject: return $v.objVal
|
||||||
|
of vkVector: return $v.vecVal
|
||||||
|
|
||||||
|
proc `==`*(a: Value, b: string): bool =
|
||||||
|
if a.kind == vkString: return a.strVal == b
|
||||||
|
if a.kind == vkNull: return b == "\\N"
|
||||||
|
return $a == b
|
||||||
|
|
||||||
|
proc `==`*(a: string, b: Value): bool =
|
||||||
|
return b == a
|
||||||
|
|
||||||
|
proc `in`*(v: Value, s: seq[string]): bool =
|
||||||
|
return $v in s
|
||||||
|
|
||||||
|
proc `in`*(s: string, v: Value): bool =
|
||||||
|
if v.kind == vkString: return v.strVal.contains(s)
|
||||||
|
return ($v).contains(s)
|
||||||
|
|||||||
@@ -43,6 +43,10 @@ type
|
|||||||
nkSetVar
|
nkSetVar
|
||||||
nkCreateGraph
|
nkCreateGraph
|
||||||
nkDropGraph
|
nkDropGraph
|
||||||
|
nkCreateDatabase
|
||||||
|
nkDropDatabase
|
||||||
|
nkUseDatabase
|
||||||
|
nkShowDatabases
|
||||||
|
|
||||||
# Clauses
|
# Clauses
|
||||||
nkFrom
|
nkFrom
|
||||||
@@ -336,6 +340,16 @@ type
|
|||||||
of nkDropGraph:
|
of nkDropGraph:
|
||||||
dgName*: string
|
dgName*: string
|
||||||
dgIfExists*: bool
|
dgIfExists*: bool
|
||||||
|
of nkCreateDatabase:
|
||||||
|
cdDbName*: string
|
||||||
|
cdIfNotExists*: bool
|
||||||
|
of nkDropDatabase:
|
||||||
|
ddDbName*: string
|
||||||
|
ddIfExists*: bool
|
||||||
|
of nkUseDatabase:
|
||||||
|
udDbName*: string
|
||||||
|
of nkShowDatabases:
|
||||||
|
discard
|
||||||
of nkApplyMigration:
|
of nkApplyMigration:
|
||||||
amName*: string
|
amName*: string
|
||||||
of nkMigrationStatus:
|
of nkMigrationStatus:
|
||||||
|
|||||||
+467
-300
File diff suppressed because it is too large
Load Diff
@@ -160,6 +160,10 @@ type
|
|||||||
tkBfs
|
tkBfs
|
||||||
tkDfs
|
tkDfs
|
||||||
tkPath
|
tkPath
|
||||||
|
tkDatabase
|
||||||
|
tkDatabases
|
||||||
|
tkUse
|
||||||
|
tkShow
|
||||||
|
|
||||||
tkAutoIncrement
|
tkAutoIncrement
|
||||||
tkSequence
|
tkSequence
|
||||||
@@ -382,6 +386,10 @@ const keywords*: Table[string, TokenKind] = {
|
|||||||
"bfs": tkBfs,
|
"bfs": tkBfs,
|
||||||
"dfs": tkDfs,
|
"dfs": tkDfs,
|
||||||
"path": tkPath,
|
"path": tkPath,
|
||||||
|
"database": tkDatabase,
|
||||||
|
"databases": tkDatabases,
|
||||||
|
"use": tkUse,
|
||||||
|
"show": tkShow,
|
||||||
"over": tkOver,
|
"over": tkOver,
|
||||||
"partition": tkPartition,
|
"partition": tkPartition,
|
||||||
"row": tkRow,
|
"row": tkRow,
|
||||||
|
|||||||
@@ -1646,6 +1646,42 @@ proc parseDropGraph(p: var Parser): Node =
|
|||||||
result = Node(kind: nkDropGraph, dgName: name, dgIfExists: ifExists,
|
result = Node(kind: nkDropGraph, dgName: name, dgIfExists: ifExists,
|
||||||
line: tok.line, col: tok.col)
|
line: tok.line, col: tok.col)
|
||||||
|
|
||||||
|
proc parseCreateDatabase(p: var Parser): Node =
|
||||||
|
let tok = p.expect(tkCreate)
|
||||||
|
discard p.expect(tkDatabase)
|
||||||
|
var ifNotExists = false
|
||||||
|
if p.peek().kind == tkIf:
|
||||||
|
discard p.advance()
|
||||||
|
discard p.expect(tkNot)
|
||||||
|
discard p.expect(tkExists)
|
||||||
|
ifNotExists = true
|
||||||
|
let name = p.expect(tkIdent).value
|
||||||
|
result = Node(kind: nkCreateDatabase, cdDbName: name, cdIfNotExists: ifNotExists,
|
||||||
|
line: tok.line, col: tok.col)
|
||||||
|
|
||||||
|
proc parseDropDatabase(p: var Parser): Node =
|
||||||
|
let tok = p.expect(tkDrop)
|
||||||
|
discard p.expect(tkDatabase)
|
||||||
|
var ifExists = false
|
||||||
|
if p.peek().kind == tkIf:
|
||||||
|
discard p.advance()
|
||||||
|
discard p.expect(tkExists)
|
||||||
|
ifExists = true
|
||||||
|
let name = p.expect(tkIdent).value
|
||||||
|
result = Node(kind: nkDropDatabase, ddDbName: name, ddIfExists: ifExists,
|
||||||
|
line: tok.line, col: tok.col)
|
||||||
|
|
||||||
|
proc parseUseDatabase(p: var Parser): Node =
|
||||||
|
let tok = p.expect(tkUse)
|
||||||
|
let name = p.expect(tkIdent).value
|
||||||
|
result = Node(kind: nkUseDatabase, udDbName: name,
|
||||||
|
line: tok.line, col: tok.col)
|
||||||
|
|
||||||
|
proc parseShowDatabases(p: var Parser): Node =
|
||||||
|
let tok = p.expect(tkShow)
|
||||||
|
discard p.expect(tkDatabases)
|
||||||
|
result = Node(kind: nkShowDatabases, line: tok.line, col: tok.col)
|
||||||
|
|
||||||
proc parseStatement*(p: var Parser): Node =
|
proc parseStatement*(p: var Parser): Node =
|
||||||
case p.peek().kind
|
case p.peek().kind
|
||||||
of tkWith, tkSelect: p.parseSelect()
|
of tkWith, tkSelect: p.parseSelect()
|
||||||
@@ -1674,6 +1710,8 @@ proc parseStatement*(p: var Parser): Node =
|
|||||||
p.parseCreatePolicy()
|
p.parseCreatePolicy()
|
||||||
elif next.kind == tkGraph:
|
elif next.kind == tkGraph:
|
||||||
p.parseCreateGraph()
|
p.parseCreateGraph()
|
||||||
|
elif next.kind == tkDatabase:
|
||||||
|
p.parseCreateDatabase()
|
||||||
else:
|
else:
|
||||||
p.parseCreateType()
|
p.parseCreateType()
|
||||||
else:
|
else:
|
||||||
@@ -1695,6 +1733,8 @@ proc parseStatement*(p: var Parser): Node =
|
|||||||
p.parseDropPolicy()
|
p.parseDropPolicy()
|
||||||
elif next.kind == tkGraph:
|
elif next.kind == tkGraph:
|
||||||
p.parseDropGraph()
|
p.parseDropGraph()
|
||||||
|
elif next.kind == tkDatabase:
|
||||||
|
p.parseDropDatabase()
|
||||||
else:
|
else:
|
||||||
let tok = p.advance()
|
let tok = p.advance()
|
||||||
Node(kind: nkNullLit, line: tok.line, col: tok.col)
|
Node(kind: nkNullLit, line: tok.line, col: tok.col)
|
||||||
@@ -1713,6 +1753,14 @@ proc parseStatement*(p: var Parser): Node =
|
|||||||
p.parseRevoke()
|
p.parseRevoke()
|
||||||
of tkSet:
|
of tkSet:
|
||||||
p.parseSetVar()
|
p.parseSetVar()
|
||||||
|
of tkUse:
|
||||||
|
p.parseUseDatabase()
|
||||||
|
of tkShow:
|
||||||
|
if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkDatabases:
|
||||||
|
p.parseShowDatabases()
|
||||||
|
else:
|
||||||
|
let tok = p.advance()
|
||||||
|
Node(kind: nkNullLit, line: tok.line, col: tok.col)
|
||||||
of tkApply:
|
of tkApply:
|
||||||
if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkMigration:
|
if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkMigration:
|
||||||
p.parseApplyMigration()
|
p.parseApplyMigration()
|
||||||
|
|||||||
+118
-11
@@ -17,9 +17,11 @@ import barabadb/protocol/ssl
|
|||||||
import barabadb/storage/lsm
|
import barabadb/storage/lsm
|
||||||
import barabadb/storage/compaction
|
import barabadb/storage/compaction
|
||||||
import barabadb/core/raft
|
import barabadb/core/raft
|
||||||
|
import barabadb/query/executor
|
||||||
import barabadb/core/gossip
|
import barabadb/core/gossip
|
||||||
import barabadb/core/replication
|
import barabadb/core/replication
|
||||||
import barabadb/core/disttxn
|
import barabadb/core/disttxn
|
||||||
|
import barabadb/core/registry
|
||||||
import barabadb/tools/repair
|
import barabadb/tools/repair
|
||||||
import barabadb/tools/migrate
|
import barabadb/tools/migrate
|
||||||
|
|
||||||
@@ -28,6 +30,10 @@ type
|
|||||||
db*: LSMTree
|
db*: LSMTree
|
||||||
strategy*: compaction.CompactionStrategy
|
strategy*: compaction.CompactionStrategy
|
||||||
|
|
||||||
|
MultiCompactionManager = ref object
|
||||||
|
registry: DatabaseRegistry
|
||||||
|
strategies: Table[string, compaction.CompactionStrategy]
|
||||||
|
|
||||||
proc newCompactionManager*(db: LSMTree): CompactionManager =
|
proc newCompactionManager*(db: LSMTree): CompactionManager =
|
||||||
result = CompactionManager(db: db, strategy: compaction.newCompactionStrategy(db.dir))
|
result = CompactionManager(db: db, strategy: compaction.newCompactionStrategy(db.dir))
|
||||||
for sst in db.sstables:
|
for sst in db.sstables:
|
||||||
@@ -88,6 +94,95 @@ proc startCompactionLoop*(cm: CompactionManager, intervalMs: int = 60000) {.asyn
|
|||||||
await sleepAsync(intervalMs)
|
await sleepAsync(intervalMs)
|
||||||
cm.compact()
|
cm.compact()
|
||||||
|
|
||||||
|
proc newMultiCompactionManager*(registry: DatabaseRegistry): MultiCompactionManager =
|
||||||
|
result = MultiCompactionManager(registry: registry, strategies: initTable[string, compaction.CompactionStrategy]())
|
||||||
|
|
||||||
|
# Initialize strategies for each existing database
|
||||||
|
for name in listDatabases(registry):
|
||||||
|
let info = getDatabaseInfo(registry, name)
|
||||||
|
if info != nil:
|
||||||
|
result.strategies[name] = compaction.newCompactionStrategy(info.db.dir)
|
||||||
|
for sst in info.db.sstables:
|
||||||
|
let meta = compaction.SSTableMeta(
|
||||||
|
path: sst.path, level: sst.level, minKey: sst.minKey, maxKey: sst.maxKey,
|
||||||
|
entryCount: sst.entryCount, sizeBytes: sst.entryCount * 64, createdAt: 0)
|
||||||
|
result.strategies[name].addTable(meta)
|
||||||
|
|
||||||
|
proc compactAll(mcm: MultiCompactionManager) =
|
||||||
|
for name in listDatabases(mcm.registry):
|
||||||
|
let info = getDatabaseInfo(mcm.registry, name)
|
||||||
|
if info == nil: continue
|
||||||
|
let db = info.db
|
||||||
|
|
||||||
|
# Initialize strategy if not already
|
||||||
|
if name notin mcm.strategies:
|
||||||
|
mcm.strategies[name] = compaction.newCompactionStrategy(db.dir)
|
||||||
|
for sst in db.sstables:
|
||||||
|
let meta = compaction.SSTableMeta(
|
||||||
|
path: sst.path, level: sst.level, minKey: sst.minKey, maxKey: sst.maxKey,
|
||||||
|
entryCount: sst.entryCount, sizeBytes: sst.entryCount * 64, createdAt: 0)
|
||||||
|
mcm.strategies[name].addTable(meta)
|
||||||
|
|
||||||
|
let strategy = mcm.strategies[name]
|
||||||
|
acquire(db.lock)
|
||||||
|
try:
|
||||||
|
for level in 0 ..< compaction.MaxLevel:
|
||||||
|
if strategy.needsCompaction(level):
|
||||||
|
let result = strategy.compact(level)
|
||||||
|
if result.outputTables.len == 0: continue
|
||||||
|
|
||||||
|
var newSSTables: seq[SSTable] = @[]
|
||||||
|
var removedPaths = initTable[string, bool]()
|
||||||
|
for t in result.inputTables:
|
||||||
|
removedPaths[t.path] = true
|
||||||
|
for sst in db.sstables:
|
||||||
|
if sst.path notin removedPaths:
|
||||||
|
newSSTables.add(sst)
|
||||||
|
|
||||||
|
for meta in result.outputTables:
|
||||||
|
try:
|
||||||
|
var sst = loadSSTable(meta.path)
|
||||||
|
let sstName = splitFile(meta.path).name
|
||||||
|
sst.id = try: parseInt(sstName) except: db.nextSSTableId
|
||||||
|
sst.level = meta.level
|
||||||
|
newSSTables.add(sst)
|
||||||
|
db.nextSSTableId = max(db.nextSSTableId, sst.id + 1)
|
||||||
|
except CatchableError as e:
|
||||||
|
warn("Compaction output SSTable failed to load: " & meta.path & " - " & e.msg)
|
||||||
|
|
||||||
|
newSSTables.sort(proc(a, b: SSTable): int = cmp(a.id, b.id))
|
||||||
|
db.sstables = newSSTables
|
||||||
|
inc db.manifestSequence
|
||||||
|
try:
|
||||||
|
writeManifest(db)
|
||||||
|
except CatchableError as e:
|
||||||
|
warn("Failed to write MANIFEST after compaction: " & e.msg)
|
||||||
|
finally:
|
||||||
|
release(db.lock)
|
||||||
|
|
||||||
|
proc startMultiCompactionLoop*(mcm: MultiCompactionManager, intervalMs: int = 60000) {.async.} =
|
||||||
|
while true:
|
||||||
|
await sleepAsync(intervalMs)
|
||||||
|
mcm.compactAll()
|
||||||
|
|
||||||
|
proc migrateLegacyData(registry: DatabaseRegistry, config: BaraConfig) =
|
||||||
|
let legacyDir = config.dataDir / "server"
|
||||||
|
let defaultDir = registry.dataRoot / "default"
|
||||||
|
if dirExists(legacyDir / "sstables") and not dirExists(defaultDir):
|
||||||
|
info("Migrating legacy data from " & legacyDir & " to " & defaultDir)
|
||||||
|
createDir(defaultDir / "sstables")
|
||||||
|
createDir(defaultDir / "wal")
|
||||||
|
for kind, path in walkDir(legacyDir / "sstables"):
|
||||||
|
if kind == pcFile:
|
||||||
|
copyFile(path, defaultDir / "sstables" / splitPath(path).tail)
|
||||||
|
for kind, path in walkDir(legacyDir / "wal"):
|
||||||
|
if kind == pcFile:
|
||||||
|
copyFile(path, defaultDir / "wal" / splitPath(path).tail)
|
||||||
|
if fileExists(legacyDir / "MANIFEST"):
|
||||||
|
copyFile(legacyDir / "MANIFEST", defaultDir / "MANIFEST")
|
||||||
|
moveDir(legacyDir, legacyDir & ".migrated")
|
||||||
|
info("Legacy data migration complete. Original renamed to " & legacyDir & ".migrated")
|
||||||
|
|
||||||
proc runTcpServer(config: BaraConfig) {.async.} =
|
proc runTcpServer(config: BaraConfig) {.async.} =
|
||||||
info("BaraDB TCP listening on " & config.address & ":" & $config.port)
|
info("BaraDB TCP listening on " & config.address & ":" & $config.port)
|
||||||
var server = newServer(config)
|
var server = newServer(config)
|
||||||
@@ -230,34 +325,45 @@ proc main() =
|
|||||||
warn("Failed to generate self-signed certificate. TLS disabled.")
|
warn("Failed to generate self-signed certificate. TLS disabled.")
|
||||||
config.tlsEnabled = false
|
config.tlsEnabled = false
|
||||||
|
|
||||||
# Shared LSMTree instance for both TCP and HTTP servers
|
# Create database registry with multi-database support
|
||||||
let dataDir = config.dataDir / "server"
|
let registry = newDatabaseRegistry (config)
|
||||||
let sharedDb = newLSMTree(dataDir)
|
registry.setContextFactory(proc(db: LSMTree, reg: DatabaseRegistry): ContextRef {.closure.} =
|
||||||
|
cast[ContextRef](cast[pointer](newExecutionContext(db, reg))))
|
||||||
|
|
||||||
|
# Migrate legacy data from old single-DB layout
|
||||||
|
migrateLegacyData(registry, config)
|
||||||
|
|
||||||
|
# Load existing databases and ensure default
|
||||||
|
registry.loadExistingDatabases()
|
||||||
|
registry.ensureDefaultDatabase()
|
||||||
|
|
||||||
|
info("Databases loaded: " & $listDatabases(registry).join(", "))
|
||||||
|
|
||||||
# Start HTTP server (blocking, multi-threaded via hunos) in background thread
|
# Start HTTP server (blocking, multi-threaded via hunos) in background thread
|
||||||
var httpServer = newHttpServerWithDb(config, sharedDb)
|
var httpServer = newHttpServerWithRegistry(config, registry)
|
||||||
spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440
|
spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440
|
||||||
|
|
||||||
# Start background compaction loop
|
# Start background compaction loop for all databases
|
||||||
let cm = newCompactionManager(sharedDb)
|
let mcm = newMultiCompactionManager(registry)
|
||||||
asyncCheck cm.startCompactionLoop()
|
asyncCheck mcm.startMultiCompactionLoop()
|
||||||
|
|
||||||
# Create TCP server (initialization is synchronous, run is async)
|
# Create TCP server (initialization is synchronous, run is async)
|
||||||
let localId {.used.} = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
|
let localId {.used.} = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
|
||||||
var tcpServer = newServerWithDb(config, sharedDb)
|
var tcpServer = newServerWithRegistry(config, registry)
|
||||||
|
|
||||||
# Start Raft cluster if enabled
|
# Start Raft cluster if enabled
|
||||||
if config.raftEnabled:
|
if config.raftEnabled:
|
||||||
info("Starting Raft node " & config.raftNodeId & " on port " & $config.raftPort)
|
info("Starting Raft node " & config.raftNodeId & " on port " & $config.raftPort)
|
||||||
var raftNode = newRaftNode(config.raftNodeId, config.raftPeers, config.raftPort)
|
var raftNode = newRaftNode(config.raftNodeId, config.raftPeers, config.raftPort)
|
||||||
# Wire state machine to apply committed entries to the database
|
# Wire state machine to apply committed entries to the default database
|
||||||
|
let defaultDbInfo = getDatabaseInfo(registry, "default")
|
||||||
raftNode.applyCommand = proc(cmd: string, data: seq[byte]) {.gcsafe.} =
|
raftNode.applyCommand = proc(cmd: string, data: seq[byte]) {.gcsafe.} =
|
||||||
if cmd == "put":
|
if cmd == "put":
|
||||||
let parts = cast[string](data).split("\x00")
|
let parts = cast[string](data).split("\x00")
|
||||||
if parts.len >= 2:
|
if parts.len >= 2:
|
||||||
sharedDb.put(parts[0], cast[seq[byte]](parts[1]))
|
defaultDbInfo.db.put(parts[0], cast[seq[byte]](parts[1]))
|
||||||
elif cmd == "delete":
|
elif cmd == "delete":
|
||||||
sharedDb.delete(cast[string](data))
|
defaultDbInfo.db.delete(cast[string](data))
|
||||||
|
|
||||||
# Wire RAFT ↔ DistTxn
|
# Wire RAFT ↔ DistTxn
|
||||||
wireRaftDistTxn(raftNode, tcpServer)
|
wireRaftDistTxn(raftNode, tcpServer)
|
||||||
@@ -291,6 +397,7 @@ proc main() =
|
|||||||
|
|
||||||
# Shutdown
|
# Shutdown
|
||||||
httpServer.stop()
|
httpServer.stop()
|
||||||
|
tcpServer.stop()
|
||||||
if tcpServer.gossipProtocol != nil:
|
if tcpServer.gossipProtocol != nil:
|
||||||
tcpServer.gossipProtocol.stop()
|
tcpServer.gossipProtocol.stop()
|
||||||
|
|
||||||
|
|||||||
Executable
BIN
Binary file not shown.
@@ -0,0 +1,252 @@
|
|||||||
|
import std/unittest
|
||||||
|
import std/strutils
|
||||||
|
import std/os
|
||||||
|
import std/tables
|
||||||
|
import ../src/barabadb/query/[parser, executor, lexer, ast]
|
||||||
|
import ../src/barabadb/core/types
|
||||||
|
import ../src/barabadb/storage/lsm
|
||||||
|
|
||||||
|
const testDir = "/tmp/baradb_bugfix_test"
|
||||||
|
|
||||||
|
proc setupCtx(): ExecutionContext =
|
||||||
|
removeDir(testDir)
|
||||||
|
createDir(testDir)
|
||||||
|
let db = newLSMTree(testDir)
|
||||||
|
var ctx = newExecutionContext(db)
|
||||||
|
# Create tables
|
||||||
|
discard executeQuery(ctx, parse("""
|
||||||
|
CREATE TABLE thread (id INTEGER PRIMARY KEY, name TEXT, category INTEGER, isDeleted INTEGER)
|
||||||
|
"""))
|
||||||
|
discard executeQuery(ctx, parse("""
|
||||||
|
CREATE TABLE category (id INTEGER PRIMARY KEY, name TEXT, description TEXT)
|
||||||
|
"""))
|
||||||
|
discard executeQuery(ctx, parse("""
|
||||||
|
CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)
|
||||||
|
"""))
|
||||||
|
ctx
|
||||||
|
|
||||||
|
proc teardown(ctx: ExecutionContext) =
|
||||||
|
ctx.db.close()
|
||||||
|
removeDir(testDir)
|
||||||
|
|
||||||
|
suite "Bug fixes — IN list, nkPath exprToSql, multi-table joins":
|
||||||
|
|
||||||
|
test "IN (list) parses without error":
|
||||||
|
let sql = "SELECT id, name FROM users WHERE id IN (1, 2, 3)"
|
||||||
|
let tokens = tokenize(sql)
|
||||||
|
let ast = parse(tokens)
|
||||||
|
check ast.stmts[0].kind == nkSelect
|
||||||
|
let whereExpr = ast.stmts[0].selWhere.whereExpr
|
||||||
|
check whereExpr.kind == nkInExpr
|
||||||
|
check whereExpr.inRight.kind == nkArrayLit
|
||||||
|
check whereExpr.inRight.arrayElems.len == 3
|
||||||
|
|
||||||
|
test "IN (list) with strings parses":
|
||||||
|
let sql = "SELECT * FROM users WHERE name IN ('alice', 'bob', 'charlie')"
|
||||||
|
let tokens = tokenize(sql)
|
||||||
|
let ast = parse(tokens)
|
||||||
|
let whereExpr = ast.stmts[0].selWhere.whereExpr
|
||||||
|
check whereExpr.kind == nkInExpr
|
||||||
|
check whereExpr.inRight.kind == nkArrayLit
|
||||||
|
check whereExpr.inRight.arrayElems.len == 3
|
||||||
|
|
||||||
|
test "IN (subquery) still parses":
|
||||||
|
let sql = "SELECT id FROM users WHERE id IN (SELECT id FROM admins)"
|
||||||
|
let tokens = tokenize(sql)
|
||||||
|
let ast = parse(tokens)
|
||||||
|
let whereExpr = ast.stmts[0].selWhere.whereExpr
|
||||||
|
check whereExpr.kind == nkInExpr
|
||||||
|
check whereExpr.inRight.kind == nkSubquery
|
||||||
|
|
||||||
|
test "nkPath column alias in SELECT produces correct data":
|
||||||
|
let sql = "SELECT t.id FROM posts t"
|
||||||
|
let tokens = tokenize(sql)
|
||||||
|
let ast = parse(tokens)
|
||||||
|
let selExpr = ast.stmts[0].selResult[0]
|
||||||
|
check selExpr.kind == nkPath
|
||||||
|
check selExpr.pathParts == @["t", "id"]
|
||||||
|
# Verify correct data is returned
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("CREATE TABLE posts (id INTEGER PRIMARY KEY, title TEXT)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO posts (id, title) VALUES (42, 'Hello')"))
|
||||||
|
let r = executeQuery(ctx, parse(sql))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 1
|
||||||
|
# Column is "t.id" (qualified by alias)
|
||||||
|
check valueToString(r.rows[0]["t.id"]) == "42"
|
||||||
|
|
||||||
|
test "IN (list) executes with actual data":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'bob')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (3, 'charlie')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (4, 'dave')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT id, name FROM users WHERE id IN (1, 3, 4)"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 3
|
||||||
|
|
||||||
|
test "Multi-table implicit join without aliases (Bug 1)":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO thread (id, name, category, isDeleted) VALUES (3, 'Test Thread', 1, 0)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO category (id, name, description) VALUES (1, 'General', 'General discussion')"))
|
||||||
|
|
||||||
|
# This used to return 0 rows
|
||||||
|
let r = executeQuery(ctx, parse("SELECT thread.id, thread.name, category.id, category.name FROM thread, category WHERE thread.id = 3 AND thread.isDeleted = 0 AND thread.category = category.id"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 1
|
||||||
|
|
||||||
|
test "Multi-table join with aliases still works":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO thread (id, name, category, isDeleted) VALUES (3, 'Test Thread', 1, 0)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO category (id, name, description) VALUES (1, 'General', 'General discussion')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT t.id AS thread_id, t.name AS thread_name, c.id AS cat_id, c.name AS cat_name FROM thread t, category c WHERE t.id = 3 AND t.category = c.id"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 1
|
||||||
|
check r.columns == @["thread_id", "thread_name", "cat_id", "cat_name"]
|
||||||
|
|
||||||
|
test "Three-table join without aliases (Bug 4 - nkPath in column names)":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("CREATE TABLE post (id INTEGER PRIMARY KEY, author INTEGER, thread INTEGER, creation TEXT)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO thread (id, name, category, isDeleted) VALUES (5, 'Thread5', 1, 0)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO category (id, name, description) VALUES (1, 'Cat1', 'desc1')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (10, 'Alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO post (id, author, thread, creation) VALUES (100, 10, 5, '2024-06-01 12:00:00')"))
|
||||||
|
|
||||||
|
# Three-table join with nkPath references
|
||||||
|
let r = executeQuery(ctx, parse("SELECT post.id, post.creation, post.thread, users.id, users.name FROM post, users, thread WHERE post.thread = thread.id AND post.author = users.id AND post.id = 100"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 1
|
||||||
|
# Verify qualified column references resolve correctly
|
||||||
|
check valueToString(r.rows[0]["post.id"]) == "100"
|
||||||
|
check valueToString(r.rows[0]["users.name"]) == "Alice"
|
||||||
|
check valueToString(r.rows[0]["post.thread"]) == "5"
|
||||||
|
# Check column names don't contain "nkPath"
|
||||||
|
for col in r.columns:
|
||||||
|
check not col.contains("nkPath")
|
||||||
|
|
||||||
|
test "NOT IN (list) parses and executes":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'bob')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (3, 'charlie')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (4, 'dave')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT id, name FROM users WHERE id NOT IN (1, 4)"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 2
|
||||||
|
|
||||||
|
test "NOT LIKE parses and executes":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'bob')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (3, 'charlie')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT name FROM users WHERE name NOT LIKE 'a%'"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 2
|
||||||
|
|
||||||
|
test "NOT BETWEEN parses and executes":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (5, 'bob')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (10, 'charlie')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (15, 'dave')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT id, name FROM users WHERE id NOT BETWEEN 3 AND 12"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 2
|
||||||
|
|
||||||
|
test "Multi-column ORDER BY sorts correctly":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'bob')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (3, 'alice')"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (4, 'charlie')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT name, id FROM users ORDER BY name ASC, id DESC"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 4
|
||||||
|
# After sorting by name ASC, then id DESC within same name:
|
||||||
|
# alice(3), alice(1), bob(2), charlie(4)
|
||||||
|
check valueToString(r.rows[0]["id"]) == "3"
|
||||||
|
check valueToString(r.rows[1]["id"]) == "1"
|
||||||
|
check valueToString(r.rows[2]["id"]) == "2"
|
||||||
|
check valueToString(r.rows[3]["id"]) == "4"
|
||||||
|
|
||||||
|
test "Numeric != comparison is consistent with =":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
# 5.0 stored as string should equal integer 5 via numeric comparison
|
||||||
|
let r = executeQuery(ctx, parse("SELECT id, name FROM users WHERE id != 999"))
|
||||||
|
check r.success
|
||||||
|
# Should return all rows since none have id=999
|
||||||
|
|
||||||
|
test "Column alias consistency for table.column in getSelectColumns":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("CREATE TABLE posts (id INTEGER PRIMARY KEY, title TEXT)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO posts (id, title) VALUES (1, 'Hello')"))
|
||||||
|
|
||||||
|
let r = executeQuery(ctx, parse("SELECT posts.id, posts.title FROM posts"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 1
|
||||||
|
# Column names should match the full path "posts.id", "posts.title"
|
||||||
|
check r.columns == @["posts.id", "posts.title"]
|
||||||
|
# Data should be accessible via both qualified and unqualified names
|
||||||
|
check valueToString(r.rows[0]["posts.id"]) == "1"
|
||||||
|
|
||||||
|
test "DELETE inside transaction actually removes row":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (10, 'temp')"))
|
||||||
|
# Verify row exists
|
||||||
|
let r1 = executeQuery(ctx, parse("SELECT * FROM users WHERE id = 10"))
|
||||||
|
check r1.success
|
||||||
|
check r1.rows.len == 1
|
||||||
|
# Delete within transaction
|
||||||
|
discard executeQuery(ctx, parse("BEGIN"))
|
||||||
|
let del = executeQuery(ctx, parse("DELETE FROM users WHERE id = 10"))
|
||||||
|
check del.success
|
||||||
|
discard executeQuery(ctx, parse("COMMIT"))
|
||||||
|
# Verify row is gone
|
||||||
|
let r2 = executeQuery(ctx, parse("SELECT * FROM users WHERE id = 10"))
|
||||||
|
check r2.success
|
||||||
|
check r2.rows.len == 0
|
||||||
|
|
||||||
|
test "PK-only row survives transaction commit":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("CREATE TABLE pksimple (id INTEGER PRIMARY KEY)"))
|
||||||
|
discard executeQuery(ctx, parse("BEGIN"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO pksimple (id) VALUES (1)"))
|
||||||
|
discard executeQuery(ctx, parse("COMMIT"))
|
||||||
|
let r = executeQuery(ctx, parse("SELECT * FROM pksimple WHERE id = 1"))
|
||||||
|
check r.success
|
||||||
|
check r.rows.len == 1
|
||||||
|
|
||||||
|
test "MIN and MAX skip NULL values":
|
||||||
|
var ctx = setupCtx()
|
||||||
|
defer: teardown(ctx)
|
||||||
|
discard executeQuery(ctx, parse("CREATE TABLE scores (id INTEGER PRIMARY KEY, val INTEGER)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO scores (id, val) VALUES (1, 10)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO scores (id, val) VALUES (2, NULL)"))
|
||||||
|
discard executeQuery(ctx, parse("INSERT INTO scores (id, val) VALUES (3, 30)"))
|
||||||
|
let rmin = executeQuery(ctx, parse("SELECT MIN(val) AS m FROM scores"))
|
||||||
|
check rmin.success
|
||||||
|
check rmin.rows.len == 1
|
||||||
|
check valueToString(rmin.rows[0]["m"]) == "10"
|
||||||
|
let rmax = executeQuery(ctx, parse("SELECT MAX(val) AS m FROM scores"))
|
||||||
|
check rmax.success
|
||||||
|
check rmax.rows.len == 1
|
||||||
|
check valueToString(rmax.rows[0]["m"]) == "30"
|
||||||
@@ -3,6 +3,8 @@ import std/os
|
|||||||
import std/strutils
|
import std/strutils
|
||||||
import std/monotimes
|
import std/monotimes
|
||||||
import std/tables
|
import std/tables
|
||||||
|
import std/sequtils
|
||||||
|
import barabadb/core/types
|
||||||
import barabadb/query/executor as qexec
|
import barabadb/query/executor as qexec
|
||||||
import barabadb/query/parser
|
import barabadb/query/parser
|
||||||
import barabadb/storage/lsm
|
import barabadb/storage/lsm
|
||||||
@@ -97,7 +99,7 @@ suite "JOIN execution":
|
|||||||
var foundBob = false
|
var foundBob = false
|
||||||
for row in r.rows:
|
for row in r.rows:
|
||||||
if row["u.name"] == "Bob":
|
if row["u.name"] == "Bob":
|
||||||
check row["x.total"] == "\\N"
|
check row["x.total"].kind == vkNull
|
||||||
foundBob = true
|
foundBob = true
|
||||||
check foundBob
|
check foundBob
|
||||||
|
|
||||||
|
|||||||
+14
-14
@@ -2831,7 +2831,7 @@ suite "Window Functions":
|
|||||||
# Sales: Diana(75000)=1, Charlie(70000)=2
|
# Sales: Diana(75000)=1, Charlie(70000)=2
|
||||||
var found = initTable[string, string]()
|
var found = initTable[string, string]()
|
||||||
for row in r.rows:
|
for row in r.rows:
|
||||||
found[row["name"]] = row["rn"]
|
found[$row["name"]] = $row["rn"]
|
||||||
check found["Eve"] == "1"
|
check found["Eve"] == "1"
|
||||||
check found["Alice"] == "2"
|
check found["Alice"] == "2"
|
||||||
check found["Bob"] == "3"
|
check found["Bob"] == "3"
|
||||||
@@ -3474,7 +3474,7 @@ suite "Auto-Increment & ID Generators":
|
|||||||
let r = qexec.executeQuery(ctx, parse("SELECT gen_random_uuid() AS uid"))
|
let r = qexec.executeQuery(ctx, parse("SELECT gen_random_uuid() AS uid"))
|
||||||
check r.success
|
check r.success
|
||||||
check r.rows.len == 1
|
check r.rows.len == 1
|
||||||
let uid = r.rows[0]["uid"]
|
let uid = $r.rows[0]["uid"]
|
||||||
check uid.len == 36
|
check uid.len == 36
|
||||||
check uid[8] == '-'
|
check uid[8] == '-'
|
||||||
check uid[13] == '-'
|
check uid[13] == '-'
|
||||||
@@ -3486,7 +3486,7 @@ suite "Auto-Increment & ID Generators":
|
|||||||
let r = qexec.executeQuery(ctx, parse("SELECT uuid() AS uid"))
|
let r = qexec.executeQuery(ctx, parse("SELECT uuid() AS uid"))
|
||||||
check r.success
|
check r.success
|
||||||
check r.rows.len == 1
|
check r.rows.len == 1
|
||||||
check r.rows[0]["uid"].len == 36
|
check ($r.rows[0]["uid"]).len == 36
|
||||||
|
|
||||||
test "Two UUIDs are different":
|
test "Two UUIDs are different":
|
||||||
let r = qexec.executeQuery(ctx, parse("SELECT uuid() AS a, uuid() AS b"))
|
let r = qexec.executeQuery(ctx, parse("SELECT uuid() AS a, uuid() AS b"))
|
||||||
@@ -3564,7 +3564,7 @@ suite "Auto-Increment & ID Generators":
|
|||||||
let r = qexec.executeQuery(ctx, parse("SELECT snowflake_id(1) AS sid"))
|
let r = qexec.executeQuery(ctx, parse("SELECT snowflake_id(1) AS sid"))
|
||||||
check r.success
|
check r.success
|
||||||
check r.rows.len == 1
|
check r.rows.len == 1
|
||||||
let sid = r.rows[0]["sid"]
|
let sid = $r.rows[0]["sid"]
|
||||||
check sid.len > 0
|
check sid.len > 0
|
||||||
var num: int64 = 0
|
var num: int64 = 0
|
||||||
try:
|
try:
|
||||||
@@ -3720,17 +3720,17 @@ suite "Type Safety — evalExprValue":
|
|||||||
check r.success
|
check r.success
|
||||||
check r.rows[0]["r"] == "17.5"
|
check r.rows[0]["r"] == "17.5"
|
||||||
|
|
||||||
test "evalExprValue returns correct Value kind for literals":
|
test "evalExpr returns correct Value kind for literals":
|
||||||
let lit = IRExpr(kind: irekLiteral, valueKind: vkInt64)
|
let lit = IRExpr(kind: irekLiteral, valueKind: vkInt64)
|
||||||
lit.literal = IRLiteral(kind: vkInt64, int64Val: 42)
|
lit.literal = IRLiteral(kind: vkInt64, int64Val: 42)
|
||||||
let v = evalExprValue(lit, initTable[string, string](), nil)
|
let v = evalExpr(lit, initTable[string, Value](), nil)
|
||||||
check v.kind == vkInt64
|
check v.kind == vkInt64
|
||||||
check v.int64Val == 42
|
check v.int64Val == 42
|
||||||
|
|
||||||
test "evalExprValue returns correct Value kind for float literal":
|
test "evalExpr returns correct Value kind for float literal":
|
||||||
let lit = IRExpr(kind: irekLiteral, valueKind: vkFloat64)
|
let lit = IRExpr(kind: irekLiteral, valueKind: vkFloat64)
|
||||||
lit.literal = IRLiteral(kind: vkFloat64, float64Val: 3.14)
|
lit.literal = IRLiteral(kind: vkFloat64, float64Val: 3.14)
|
||||||
let v = evalExprValue(lit, initTable[string, string](), nil)
|
let v = evalExpr(lit, initTable[string, Value](), nil)
|
||||||
check v.kind == vkFloat64
|
check v.kind == vkFloat64
|
||||||
check v.float64Val == 3.14
|
check v.float64Val == 3.14
|
||||||
|
|
||||||
@@ -3998,7 +3998,7 @@ suite "Hybrid RAG Search":
|
|||||||
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search('docs', 'embedding', 'content', 'quick brown', '[1.0, 0.0, 0.0]', 10) AS res"))
|
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search('docs', 'embedding', 'content', 'quick brown', '[1.0, 0.0, 0.0]', 10) AS res"))
|
||||||
check r.success
|
check r.success
|
||||||
check r.rows.len == 1
|
check r.rows.len == 1
|
||||||
let jsonStr = r.rows[0]["res"]
|
let jsonStr = $r.rows[0]["res"]
|
||||||
check jsonStr.len > 2 # not "[]"
|
check jsonStr.len > 2 # not "[]"
|
||||||
let arr = parseJson(jsonStr)
|
let arr = parseJson(jsonStr)
|
||||||
check arr.kind == JArray
|
check arr.kind == JArray
|
||||||
@@ -4013,7 +4013,7 @@ suite "Hybrid RAG Search":
|
|||||||
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_ids('docs2', 'embedding', 'content', 'machine learning', '[0.0, 1.0, 0.0]', 10) AS ids"))
|
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_ids('docs2', 'embedding', 'content', 'machine learning', '[0.0, 1.0, 0.0]', 10) AS ids"))
|
||||||
check r.success
|
check r.success
|
||||||
check r.rows.len == 1
|
check r.rows.len == 1
|
||||||
let idsStr = r.rows[0]["ids"]
|
let idsStr = $r.rows[0]["ids"]
|
||||||
check idsStr.len > 0
|
check idsStr.len > 0
|
||||||
check idsStr.contains("20")
|
check idsStr.contains("20")
|
||||||
|
|
||||||
@@ -4029,7 +4029,7 @@ suite "Hybrid RAG Search":
|
|||||||
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts3 ON docs3(content) USING FTS"))
|
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts3 ON docs3(content) USING FTS"))
|
||||||
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search('docs3', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10) AS res"))
|
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search('docs3', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10) AS res"))
|
||||||
check r.success
|
check r.success
|
||||||
let arr = parseJson(r.rows[0]["res"])
|
let arr = parseJson($r.rows[0]["res"])
|
||||||
check arr.len == 3
|
check arr.len == 3
|
||||||
# Doc 3 should be first (matches both vector and FTS), doc 1 second (vector only), doc 2 third (no match)
|
# Doc 3 should be first (matches both vector and FTS), doc 1 second (vector only), doc 2 third (no match)
|
||||||
check arr[0]["id"].getStr() == "3"
|
check arr[0]["id"].getStr() == "3"
|
||||||
@@ -4038,7 +4038,7 @@ suite "Hybrid RAG Search":
|
|||||||
let r = qexec.executeQuery(ctx, parse("SELECT rerank('quick brown', '[{\"id\":\"1\",\"score\":\"0.5\"},{\"id\":\"2\",\"score\":\"0.5\"}]') AS res"))
|
let r = qexec.executeQuery(ctx, parse("SELECT rerank('quick brown', '[{\"id\":\"1\",\"score\":\"0.5\"},{\"id\":\"2\",\"score\":\"0.5\"}]') AS res"))
|
||||||
check r.success
|
check r.success
|
||||||
# Both have same score, rerank should preserve order (no content to boost)
|
# Both have same score, rerank should preserve order (no content to boost)
|
||||||
let arr = parseJson(r.rows[0]["res"])
|
let arr = parseJson($r.rows[0]["res"])
|
||||||
check arr.kind == JArray
|
check arr.kind == JArray
|
||||||
check arr.len == 2
|
check arr.len == 2
|
||||||
|
|
||||||
@@ -4060,7 +4060,7 @@ suite "Hybrid RAG Search":
|
|||||||
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts5 ON docs5(content) USING FTS"))
|
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts5 ON docs5(content) USING FTS"))
|
||||||
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_filtered('docs5', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10, 'tenant_id', 'tenant-a') AS res"))
|
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_filtered('docs5', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10, 'tenant_id', 'tenant-a') AS res"))
|
||||||
check r.success
|
check r.success
|
||||||
let arr = parseJson(r.rows[0]["res"])
|
let arr = parseJson($r.rows[0]["res"])
|
||||||
# Doc 1 (tenant-a) should be first and highest scored; Doc 2 (tenant-b) must be excluded
|
# Doc 1 (tenant-a) should be first and highest scored; Doc 2 (tenant-b) must be excluded
|
||||||
check arr[0]["id"].getStr() == "1"
|
check arr[0]["id"].getStr() == "1"
|
||||||
for elem in arr:
|
for elem in arr:
|
||||||
@@ -4074,6 +4074,6 @@ suite "Hybrid RAG Search":
|
|||||||
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts6 ON docs6(content) USING FTS"))
|
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_fts6 ON docs6(content) USING FTS"))
|
||||||
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_filtered('docs6', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10, '', '') AS res"))
|
let r = qexec.executeQuery(ctx, parse("SELECT hybrid_search_filtered('docs6', 'embedding', 'content', 'quick brown fox', '[1.0, 0.0, 0.0]', 10, '', '') AS res"))
|
||||||
check r.success
|
check r.success
|
||||||
let arr = parseJson(r.rows[0]["res"])
|
let arr = parseJson($r.rows[0]["res"])
|
||||||
check arr.len == 2
|
check arr.len == 2
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user