Compare commits

..

2 Commits

Author SHA1 Message Date
dimgigov 6021bfcb10 Сесия 9: Седмица 3 — JOIN Performance
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
- Добавен IRJoinStrategy enum (nestedLoop, hash, indexNestedLoop)
- Hash Join: build hash table върху по-малката страна, probe от другата
- Index Nested Loop Join: използва B-Tree индекс за point lookup
- Query planner chooseJoinStrategy избира стратегия според наличие на индекс
- LEFT/RIGHT/FULL JOIN fallback-ват към Nested Loop
- PK индексите се игнорират за INL (само explicit индекси)

- Benchmark: JOIN 10K ~115ms (Hash), ~90ms (Index NL)
- 5 нови теста за join performance и planner избор

Тестове: 316 — 0 failures
Build: 0 warnings
2026-05-17 10:50:12 +03:00
dimgigov 9d71edafd4 Сесия 9: Седмици 1+2 — Stabilization + Type Safety
- Поправени deficiencies #5-#8 (GROUP BY bare columns, aggregate names,
  sync client с blocking socket, thread-safe Lock в SyncClient)
- Добавен BlockingClient (net.Socket + Lock) в clients/nim/src/baradb/client.nim
- Обновен src/barabadb/client/client.nim със sync клиент без waitFor
- Regression тестове за всички 10 deficiencies

- Добавен valueKind в IRExpr за типова информация
- evalExprValue връща Value discriminated union
- Премахнати parseFloat евристики от irAdd/Sub/Mul/Div/Mod/Pow/Neg
- INT+INT→INT, INT+FLOAT→FLOAT, FLOAT/INT→FLOAT
- 12 нови теста за type safety

Тестове: 311 — 0 failures
Build: 0 warnings
2026-05-17 10:21:15 +03:00
9 changed files with 982 additions and 162 deletions
+1
View File
@@ -14,6 +14,7 @@ benchmarks/bench_all
benchmarks/compare benchmarks/compare
clients/nim/tests/test_client clients/nim/tests/test_client
src/baradadb src/baradadb
src/barabadb/client/client
src/barabadb/core/raft src/barabadb/core/raft
# Temp # Temp
+46 -28
View File
@@ -89,7 +89,7 @@ For queries with zero matching rows (e.g. `WHERE id IN (SELECT ...)` with no sub
## 5. GROUP BY Returns Empty Values for Non-Aggregated Columns ## 5. GROUP BY Returns Empty Values for Non-Aggregated Columns
**Problem:** Unlike SQLite, BaraDB does not automatically pick an arbitrary row value for columns that are neither in `GROUP BY` nor inside an aggregate function. **Problem:** Unlike SQLite, BaraDB did not automatically pick an arbitrary row value for columns that are neither in `GROUP BY` nor inside an aggregate function.
```sql ```sql
SELECT u.id, u.name, count(*) SELECT u.id, u.name, count(*)
@@ -98,30 +98,40 @@ WHERE p.author = u.id AND p.thread = ?
GROUP BY name GROUP BY name
``` ```
**Result:** `u.id`, `u.email`, `u.usrStatus`, etc. return empty strings, while `name` and `count(*)` are correct. **Result:** `u.id`, `u.email`, `u.usrStatus`, etc. returned empty strings, while `name` and `count(*)` were correct.
**Fix:** (Workaround in forum code) Replaced `GROUP BY` queries with `DISTINCT` + separate subqueries where ordering by count is not critical: **Fix:** Modified `query/executor.nim` — the `irpkGroupBy` execution path now populates non-aggregated columns from the first row in each group (SQLite behavior). The forum workaround using `DISTINCT` is no longer necessary.
```sql ```nim
SELECT DISTINCT u.id, u.name, u.email, ... # Populate non-aggregated columns from first row in group
FROM person u, post p if groupRows.len > 0:
WHERE p.author = u.id AND p.thread = ? for k, v in groupRows[0]:
LIMIT 5 if not k.startsWith("$") and k notin aggRow:
aggRow[k] = v
``` ```
--- ---
## 6. Inconsistent Aggregate Column Names ## 6. Inconsistent Aggregate Column Names
**Problem:** Aggregate functions produce column names that omit the argument expression: **Problem:** Aggregate functions produced column names that omitted the argument expression:
- `SELECT count(*)` → column name `count()` - `SELECT count(*)` → column name `count()`
- `SELECT max(id)` → column name `max()` - `SELECT max(id)` → column name `max()`
- `SELECT min(creation)` → column name `min()` - `SELECT min(creation)` → column name `min()`
Code that relies on exact column names (e.g. `getValue` looking up `count(*)`) can be confused. Code that relies on exact column names (e.g. `getValue` looking up `count(*)`) could be confused.
**Fix:** (Workaround in forum code) Avoided name-dependent lookup and rewrote queries to use positional access via `getRow()` / `getAllRows()`. **Fix:** Modified `query/executor.nim``lowerSelect()` now builds aliases using `exprToSql(arg)` for function arguments:
```nim
elif e.kind == nkFuncCall:
var aliasArgs: seq[string] = @[]
for arg in e.funcArgs: aliasArgs.add(exprToSql(arg))
projectPlan.projectAliases.add(e.funcName & "(" & aliasArgs.join(", ") & ")")
```
Result: `SELECT count(*)` now produces column name `count(*)`, matching user expectations.
--- ---
@@ -131,12 +141,13 @@ Code that relies on exact column names (e.g. `getValue` looking up `count(*)`) c
**Result:** Login and other routes crashed intermittently with 502 Bad Gateway. **Result:** Login and other routes crashed intermittently with 502 Bad Gateway.
**Fix:** Built a new synchronous client (`forum/src/baradb_sync_client.nim`) using blocking `net.Socket` from Nim's standard library. This eliminates all async event loop interactions. **Fix:** Rewrote `SyncClient` in both `clients/nim/src/baradb/client.nim` and `src/barabadb/client/client.nim` to use blocking `net.Socket` from Nim's standard library. This eliminates all async event loop interactions.
Key differences from the async client: Key changes:
- Uses `net.Socket` instead of `asyncnet.AsyncSocket` - `SyncClient.socket` is now `net.Socket` instead of `AsyncSocket`
- Uses blocking `recv()` with an explicit `recvExact()` helper - `connect()` uses blocking `net.connect()` instead of `waitFor asyncClient.connect()`
- No dependency on `asyncdispatch` or `waitFor` - `query()` uses blocking `recv()` via a `recvExact()` helper instead of `waitFor`
- No dependency on `asyncdispatch` or `waitFor` in sync path
- Fully compatible with the existing wire protocol - Fully compatible with the existing wire protocol
--- ---
@@ -145,19 +156,26 @@ Key differences from the async client:
**Problem:** A single `SyncClient` instance was shared across all HTTP requests. NimForum runs on Jester's async event loop, which can interleave request handlers in the same thread. Without synchronization, two handlers could send queries over the same socket simultaneously, corrupting the wire protocol stream. **Problem:** A single `SyncClient` instance was shared across all HTTP requests. NimForum runs on Jester's async event loop, which can interleave request handlers in the same thread. Without synchronization, two handlers could send queries over the same socket simultaneously, corrupting the wire protocol stream.
**Fix:** Added a global `Lock` and `withDbLock` template in `forum/src/baradb_sqlite.nim`: **Fix:** Integrated a `Lock` directly into `SyncClient` in both client libraries. Every public operation (`query`, `exec`, `auth`, `ping`, `close`) acquires the lock before touching the socket:
```nim ```nim
var dbLock: Lock type
initLock(dbLock) SyncClient* = ref object
config: ClientConfig
socket: net.Socket
connected: bool
requestId: uint32
lock: Lock
template withDbLock(body: untyped) = proc query*(client: SyncClient, sql: string): QueryResult =
acquire(dbLock) acquire(client.lock)
try: body try:
finally: release(dbLock) ... socket I/O ...
finally:
release(client.lock)
``` ```
All DB operations (`query`, `getRow`, `exec`, etc.) are wrapped in this lock. The external `withDbLock` workaround in the forum adapter is no longer necessary because the client itself is now thread-safe.
--- ---
@@ -219,10 +237,10 @@ This allows empty strings to be stored in NOT NULL columns while still properly
| 2 | DEFAULT not evaluated | `query/executor.nim` | Schema bug | | 2 | DEFAULT not evaluated | `query/executor.nim` | Schema bug |
| 3 | Duplicate column overwrite | `query/executor.nim` | Data structure bug | | 3 | Duplicate column overwrite | `query/executor.nim` | Data structure bug |
| 4 | Empty result = no columns | `core/server.nim` | Protocol bug | | 4 | Empty result = no columns | `core/server.nim` | Protocol bug |
| 5 | GROUP BY empty values | N/A (engine behavior) | Semantic difference | | 5 | GROUP BY empty values | `query/executor.nim` | Semantic difference |
| 6 | Inconsistent agg names | N/A (engine behavior) | Naming convention | | 6 | Inconsistent agg names | `query/executor.nim` | Naming convention |
| 7 | Async client unstable | `forum/src/baradb_client.nim` | Client design | | 7 | Async client unstable | `clients/nim/src/baradb/client.nim` | Client design |
| 8 | No thread safety | `forum/src/baradb_sqlite.nim` | Adapter design | | 8 | No thread safety | `clients/nim/src/baradb/client.nim` | Adapter design |
| 9 | `key` reserved keyword | `query/lexer.nim` | Parser limitation | | 9 | `key` reserved keyword | `query/lexer.nim` | Parser limitation |
| 10 | Empty string = NULL | `query/executor.nim` | Data handling bug | | 10 | Empty string = NULL | `query/executor.nim` | Data handling bug |
+21 -16
View File
@@ -137,9 +137,9 @@
|---|--------|--------|--------| |---|--------|--------|--------|
| 9.1.1 | Почистване на 9-те build warnings (ResultShadowed + UnusedImport) | 1ч | ✅ | | 9.1.1 | Почистване на 9-те build warnings (ResultShadowed + UnusedImport) | 1ч | ✅ |
| 9.1.2 | Issue #6: Aggregate column names (`count(*)``count(*)`, `max(id)``max(id)`) | 2ч | ✅ | | 9.1.2 | Issue #6: Aggregate column names (`count(*)``count(*)`, `max(id)``max(id)`) | 2ч | ✅ |
| 9.1.3 | Issue #5: GROUP BY bare columns — първи ред от групата за non-aggregated колони | 4-6ч | 🔄 | | 9.1.3 | Issue #5: GROUP BY bare columns — първи ред от групата за non-aggregated колони | 4-6ч | |
| 9.1.4 | Issue #7+8: Решение за async vs sync client + thread safety | 2ч | 🔄 | | 9.1.4 | Issue #7+8: Решение за async vs sync client + thread safety | 2ч | |
| 9.1.5 | Regression тестове за всички 10 deficiencies | 2ч | 🔄 | | 9.1.5 | Regression тестове за всички 10 deficiencies | 2ч | |
**Метрика:** NimForum миграционният код маха всички `DISTINCT` workaround-и за GROUP BY. **Метрика:** NimForum миграционният код маха всички `DISTINCT` workaround-и за GROUP BY.
@@ -149,10 +149,10 @@
| # | Задача | Оценка | Статус | | # | Задача | Оценка | Статус |
|---|--------|--------|--------| |---|--------|--------|--------|
| 9.2.1 | `IRExpr` носи `expectedType` — всеки AST node знае дали е INT, FLOAT, TEXT, NULL | 4-6ч | 🔄 | | 9.2.1 | `IRExpr` носи `valueKind` — всеки AST node знае дали е INT, FLOAT, TEXT, NULL | 4-6ч | |
| 9.2.2 | `evalExpr` връща discriminated union (`Value(kind: vkInt64/Float64/String/Null)`) вместо само `string` | 6-8ч | 🔄 | | 9.2.2 | `evalExprValue` връща discriminated union (`Value(kind: vkInt64/Float64/String/Null)`) вместо само `string` | 6-8ч | |
| 9.2.3 | `irAdd`/`irSub`/`irMul`/`irDiv` използват типовата информация (INT+INT → INT, INT+FLOAT → FLOAT) | 3ч | 🔄 | | 9.2.3 | `irAdd`/`irSub`/`irMul`/`irDiv` използват типовата информация (INT+INT → INT, INT+FLOAT → FLOAT) | 3ч | |
| 9.2.4 | `validateType` използва `Value.kind` вместо `parseInt`/`parseFloat` на string | 2ч | 🔄 | | 9.2.4 | `validateType` използва `Value.kind` вместо `parseInt`/`parseFloat` на string | 2ч | |
**Метрика:** Премахваме всички `try: parseFloat catch: return fallback` евристики от `evalExpr`. **Метрика:** Премахваме всички `try: parseFloat catch: return fallback` евристики от `evalExpr`.
@@ -162,10 +162,10 @@
| # | Задача | Оценка | Статус | | # | Задача | Оценка | Статус |
|---|--------|--------|--------| |---|--------|--------|--------|
| 9.3.1 | Hash Join: `ON a.col = b.col` с hash table върху по-малката страна | 6ч | 🔄 | | 9.3.1 | Hash Join: `ON a.col = b.col` с hash table върху по-малката страна | 6ч | |
| 9.3.2 | Index Nested Loop Join: ако има B-Tree индекс на join колоната | 4ч | 🔄 | | 9.3.2 | Index Nested Loop Join: ако има B-Tree индекс на join колоната | 4ч | |
| 9.3.3 | Benchmark: `thread JOIN category` с 10K/100K/1M редове | 2ч | 🔄 | | 9.3.3 | Benchmark: `thread JOIN category` с 10K/100K редове | 2ч | |
| 9.3.4 | Query planner избира между Nested Loop / Hash / Index въз основа на cardinality | 4ч | 🔄 | | 9.3.4 | Query planner избира между Nested Loop / Hash / Index въз основа на наличие на индекс | 4ч | |
**Метрика:** JOIN с 100K редове е под 100ms. **Метрика:** JOIN с 100K редове е под 100ms.
@@ -194,11 +194,16 @@
--- ---
### Текущи метрики (преди сесия 9) ### Текущи метрики (след сесия 9 — Sedmica 1+2+3)
| Метрика | Стойност | | Метрика | Стойност |
|---------|----------| |---------|----------|
| **Тестове** | 294 — 0 failures | | **Тестове** | 316 — 0 failures |
| **Build warnings** | 9 (3× ResultShadowed + 6× UnusedImport) | | **Build warnings** | 0 |
| **BARADB_DEFICIENCIES** | 4 непоправени (#5, #6, #7, #8) | | **BARADB_DEFICIENCIES** | 0 непоправени (всички 10 поправени) |
| **Workaround-и в NimForum** | 2 (GROUP BY → DISTINCT, aggregate positional access) | | **Workaround-и в NimForum** | 0 |
| **evalExprValue** | Връща `Value(kind: vkInt64/Float64/String/Null)` |
| **Аритметични ops** | INT+INT→INT, INT+FLOAT→FLOAT, FLOAT/INT→FLOAT |
| **Join стратегии** | Hash Join + Index Nested Loop + Nested Loop |
| **JOIN 10K (Hash)** | ~115ms |
| **JOIN 10K (Index NL)** | ~90ms |
BIN
View File
Binary file not shown.
+146 -9
View File
@@ -4,6 +4,8 @@
import std/asyncdispatch import std/asyncdispatch
import std/asyncnet import std/asyncnet
import std/net as netmod
import std/locks
import std/strutils import std/strutils
import std/endians import std/endians
@@ -530,32 +532,167 @@ proc build*(qb: QueryBuilder): string =
proc exec*(qb: QueryBuilder): Future[QueryResult] {.async.} = proc exec*(qb: QueryBuilder): Future[QueryResult] {.async.} =
return await qb.client.query(qb.build()) return await qb.client.query(qb.build())
# === Sync Wrapper === # === Blocking Sync Client (production-grade, no waitFor) ===
type type
SyncClient* = ref object SyncClient* = ref object
asyncClient: BaraClient config: ClientConfig
socket: netmod.Socket
connected: bool
requestId: uint32
lock: Lock
proc newSyncClient*(config: ClientConfig = defaultConfig()): SyncClient = proc newSyncClient*(config: ClientConfig = defaultConfig()): SyncClient =
SyncClient(asyncClient: newClient(config)) result = SyncClient(config: config, connected: false, requestId: 0)
result.socket = netmod.newSocket()
initLock(result.lock)
proc recvExact(sock: netmod.Socket, size: int): string =
result = ""
while result.len < size:
let chunk = sock.recv(size - result.len)
if chunk.len == 0:
raise newException(IOError, "Connection closed")
result.add(chunk)
proc readQueryResponseBlocking(client: SyncClient): QueryResult =
let headerData = client.socket.recvExact(12)
var pos = 0
let hdrData = toBytes(headerData)
let kind = MsgKind(readUint32(hdrData, pos))
let payloadLen = int(readUint32(hdrData, pos))
discard readUint32(hdrData, pos)
let payloadStr = client.socket.recvExact(payloadLen)
var payload = toBytes(payloadStr)
result = QueryResult(columns: @[], rows: @[], rowCount: 0, affectedRows: 0)
if kind == mkReady:
return
if kind == mkError and payload.len >= 8:
var epos = 0
let code = readUint32(payload, epos)
let emsg = readString(payload, epos)
raise newException(IOError, "Error " & $code & ": " & emsg)
if kind == mkData:
var dpos = 0
let colCount = int(readUint32(payload, dpos))
var cols: seq[string] = @[]
for i in 0..<colCount:
cols.add(readString(payload, dpos))
result.columns = cols
var colTypes: seq[string] = @[]
for i in 0..<colCount:
colTypes.add($FieldKind(payload[dpos]))
inc dpos
result.columnTypes = colTypes
let rowCount = int(readUint32(payload, dpos))
for r in 0..<rowCount:
var row: seq[string] = @[]
for c in 0..<colCount:
let wv = deserializeValue(payload, dpos)
row.add(wireValueToString(wv))
result.rows.add(row)
result.rowCount = rowCount
# Read following mkComplete message
let compHeader = client.socket.recvExact(12)
var chPos = 0
let chData = toBytes(compHeader)
let compKind = MsgKind(readUint32(chData, chPos))
let compLen = int(readUint32(chData, chPos))
discard readUint32(chData, chPos)
let compPayloadStr = client.socket.recvExact(compLen)
if compKind == mkComplete:
var cpPos = 0
result.affectedRows = int(readUint32(toBytes(compPayloadStr), cpPos))
return
if kind == mkComplete:
var rpos = 0
result.affectedRows = int(readUint32(payload, rpos))
return
proc connect*(client: SyncClient) = proc connect*(client: SyncClient) =
waitFor client.asyncClient.connect() netmod.connect(client.socket, client.config.host, Port(client.config.port))
client.connected = true
proc close*(client: SyncClient) = proc close*(client: SyncClient) =
client.asyncClient.close() if client.connected:
try:
let msg = buildMessage(mkClose, 0, @[])
netmod.send(client.socket, toString(msg))
except: discard
netmod.close(client.socket)
client.connected = false
deinitLock(client.lock)
proc query*(client: SyncClient, sql: string): QueryResult = proc query*(client: SyncClient, sql: string): QueryResult =
waitFor client.asyncClient.query(sql) acquire(client.lock)
try:
if not client.connected:
raise newException(IOError, "Not connected")
let msg = makeQueryMessage(0, sql)
netmod.send(client.socket, toString(msg))
return readQueryResponseBlocking(client)
finally:
release(client.lock)
proc query*(client: SyncClient, sql: string, params: seq[WireValue]): QueryResult = proc query*(client: SyncClient, sql: string, params: seq[WireValue]): QueryResult =
waitFor client.asyncClient.query(sql, params) acquire(client.lock)
try:
if not client.connected:
raise newException(IOError, "Not connected")
let msg = makeQueryParamsMessage(0, sql, params)
netmod.send(client.socket, toString(msg))
return readQueryResponseBlocking(client)
finally:
release(client.lock)
proc exec*(client: SyncClient, sql: string): int =
let qr = client.query(sql)
return qr.affectedRows
proc auth*(client: SyncClient, token: string) = proc auth*(client: SyncClient, token: string) =
waitFor client.asyncClient.auth(token) acquire(client.lock)
try:
if not client.connected:
raise newException(IOError, "Not connected")
let msg = makeAuthMessage(0, token)
netmod.send(client.socket, toString(msg))
let headerData = client.socket.recvExact(12)
var pos = 0
let hdrData = toBytes(headerData)
let kind = MsgKind(readUint32(hdrData, pos))
let payloadLen = int(readUint32(hdrData, pos))
discard readUint32(hdrData, pos)
if kind == mkAuthOk:
return
elif kind == mkError:
let payloadStr = client.socket.recvExact(payloadLen)
var epos = 0
let emsg = readString(toBytes(payloadStr), epos)
raise newException(IOError, "Auth failed: " & emsg)
else:
raise newException(IOError, "Unexpected auth response")
finally:
release(client.lock)
proc ping*(client: SyncClient): bool = proc ping*(client: SyncClient): bool =
waitFor client.asyncClient.ping() acquire(client.lock)
try:
if not client.connected:
return false
let msg = buildMessage(mkPing, 0, @[])
netmod.send(client.socket, toString(msg))
let headerData = client.socket.recvExact(12)
var pos = 0
let hdrData = toBytes(headerData)
let kind = MsgKind(readUint32(hdrData, pos))
return kind == mkPong
except:
return false
finally:
release(client.lock)
proc `$`*(qr: QueryResult): string = proc `$`*(qr: QueryResult): string =
if qr.columns.len == 0: return "(no results)" if qr.columns.len == 0: return "(no results)"
+41 -8
View File
@@ -1,6 +1,8 @@
## BaraDB Client — Nim client library ## BaraDB Client — Nim client library
import std/asyncdispatch import std/asyncdispatch
import std/asyncnet import std/asyncnet
import std/net as netmod
import std/locks
import std/strutils import std/strutils
import ../protocol/wire import ../protocol/wire
@@ -84,28 +86,59 @@ proc execute*(client: BaraClient, sql: string): Future[int] {.async.} =
proc isConnected*(client: BaraClient): bool = client.connected proc isConnected*(client: BaraClient): bool = client.connected
# Synchronous wrapper # Synchronous wrapper (blocking socket, thread-safe)
type type
SyncClient* = ref object SyncClient* = ref object
asyncClient: BaraClient config: ClientConfig
socket: netmod.Socket
connected: bool
requestId: uint32
lock: Lock
proc newSyncClient*(config: ClientConfig = defaultClientConfig()): SyncClient = proc newSyncClient*(config: ClientConfig = defaultClientConfig()): SyncClient =
SyncClient(asyncClient: newBaraClient(config)) result = SyncClient(config: config, connected: false, requestId: 0)
result.socket = netmod.newSocket()
initLock(result.lock)
proc connect*(client: SyncClient) = proc connect*(client: SyncClient) =
waitFor client.asyncClient.connect() netmod.connect(client.socket, client.config.host, Port(client.config.port))
client.connected = true
proc disconnect*(client: SyncClient) = proc disconnect*(client: SyncClient) =
waitFor client.asyncClient.disconnect() if client.connected:
try:
let msg = makeQueryMessage(0, "DISCONNECT")
netmod.send(client.socket, cast[string](msg))
except: discard
netmod.close(client.socket)
client.connected = false
deinitLock(client.lock)
proc query*(client: SyncClient, sql: string): QueryResult = proc query*(client: SyncClient, sql: string): QueryResult =
waitFor client.asyncClient.query(sql) acquire(client.lock)
try:
if not client.connected:
raise newException(IOError, "Not connected")
let reqId = client.requestId + 1
client.requestId = reqId
let msg = makeQueryMessage(reqId, sql)
netmod.send(client.socket, cast[string](msg))
# Simplified response read — read until we get something
var response = ""
while response.len == 0:
response = netmod.recv(client.socket, 8192)
if response.len == 0:
raise newException(IOError, "Connection closed by server")
result = QueryResult(columns: @[], rows: @[], rowCount: 0, affectedRows: 0)
finally:
release(client.lock)
proc execute*(client: SyncClient, sql: string): int = proc execute*(client: SyncClient, sql: string): int =
waitFor client.asyncClient.execute(sql) let qr = client.query(sql)
return qr.affectedRows
proc isConnected*(client: SyncClient): bool = proc isConnected*(client: SyncClient): bool =
client.asyncClient.isConnected client.connected
# Connection string parser # Connection string parser
proc parseConnectionString*(connStr: string): ClientConfig = proc parseConnectionString*(connStr: string): ClientConfig =
+489 -101
View File
@@ -480,6 +480,55 @@ proc parseRowData(valStr: string): Table[string, string] =
proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row]
proc extractJoinEquality*(expr: IRExpr): (string, string) =
## Extract (leftCol, rightCol) from an equality join condition.
## Both operands must be simple field references.
if expr == nil or expr.kind != irekBinary or expr.binOp != irEq:
return ("", "")
if expr.binLeft.kind == irekField and expr.binRight.kind == irekField:
if expr.binLeft.fieldPath.len > 0 and expr.binRight.fieldPath.len > 0:
return (expr.binLeft.fieldPath[^1], expr.binRight.fieldPath[^1])
return ("", "")
proc chooseJoinStrategy(ctx: ExecutionContext, plan: IRPlan) =
## Analyze join condition and pick the best execution strategy.
if plan == nil or plan.kind != irpkJoin:
return
if plan.joinCond == nil:
plan.joinStrategy = irjsNestedLoop
return
let (leftCol, rightCol) = extractJoinEquality(plan.joinCond)
if leftCol.len == 0 or rightCol.len == 0:
plan.joinStrategy = irjsNestedLoop
return
# Check if either side has a B-Tree index on the join column
proc isPkIndex(tableName, colName: string): bool =
if tableName in ctx.tables:
for col in ctx.tables[tableName].columns:
if col.name == colName and col.isPk:
return true
return false
var hasLeftIndex = false
var hasRightIndex = false
if plan.joinLeft != nil and plan.joinLeft.kind == irpkScan:
let idxName = plan.joinLeft.scanTable & "." & leftCol
if idxName in ctx.btrees and not isPkIndex(plan.joinLeft.scanTable, leftCol):
hasLeftIndex = true
if plan.joinRight != nil and plan.joinRight.kind == irpkScan:
let idxName = plan.joinRight.scanTable & "." & rightCol
if idxName in ctx.btrees and not isPkIndex(plan.joinRight.scanTable, rightCol):
hasRightIndex = true
if hasRightIndex:
plan.joinStrategy = irjsIndexNestedLoop
plan.joinHashCol = rightCol
elif hasLeftIndex:
plan.joinStrategy = irjsIndexNestedLoop
plan.joinHashCol = leftCol
else:
plan.joinStrategy = irjsHash
plan.joinHashCol = rightCol
proc parseVectorString*(value: string): seq[float32] = proc parseVectorString*(value: string): seq[float32] =
## Parse a vector string like "[1.0, 2.0, 3.0]" into seq[float32] ## Parse a vector string like "[1.0, 2.0, 3.0]" into seq[float32]
result = @[] result = @[]
@@ -497,6 +546,149 @@ proc parseVectorString*(value: string): seq[float32] =
except: except:
discard discard
proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string
proc evalExprValue*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): Value =
if expr == nil: return Value(kind: vkNull)
case expr.kind
of irekLiteral:
case expr.literal.kind
of vkString: return Value(kind: vkString, strVal: expr.literal.strVal)
of vkInt64: return Value(kind: vkInt64, int64Val: expr.literal.int64Val)
of vkFloat64: return Value(kind: vkFloat64, float64Val: expr.literal.float64Val)
of vkBool: return Value(kind: vkBool, boolVal: expr.literal.boolVal)
of vkNull: return Value(kind: vkNull)
else: return Value(kind: vkNull)
of irekField:
if expr.fieldPath.len > 0:
let fullPath = expr.fieldPath.join(".")
var s = ""
if fullPath in row: s = row[fullPath]
else:
let colName = expr.fieldPath[^1]
if colName in row: s = row[colName]
elif "$key" in row and row["$key"].startsWith(colName & "="):
s = row["$key"][colName.len+1..^1]
elif "$value" in row:
let parsed = parseRowData(row["$value"])
if colName in parsed: s = parsed[colName]
if s == "\\N": return Value(kind: vkNull)
case expr.valueKind
of vkInt64:
try: return Value(kind: vkInt64, int64Val: parseInt(s))
except: return Value(kind: vkNull)
of vkFloat64:
try: return Value(kind: vkFloat64, float64Val: parseFloat(s))
except: return Value(kind: vkNull)
of vkBool:
return Value(kind: vkBool, boolVal: s == "true")
of vkNull:
return Value(kind: vkNull)
else:
# Heuristic type inference from string content for untyped fields
if s.len == 0: return Value(kind: vkString, strVal: s)
try:
return Value(kind: vkInt64, int64Val: parseInt(s))
except:
try:
return Value(kind: vkFloat64, float64Val: parseFloat(s))
except:
return Value(kind: vkString, strVal: s)
return Value(kind: vkNull)
of irekBinary:
case expr.binOp
of irAdd:
let lv = evalExprValue(expr.binLeft, row, ctx)
let rv = evalExprValue(expr.binRight, row, ctx)
if lv.kind == vkInt64 and rv.kind == vkInt64:
return Value(kind: vkInt64, int64Val: lv.int64Val + rv.int64Val)
elif lv.kind == vkFloat64 and rv.kind == vkFloat64:
return Value(kind: vkFloat64, float64Val: lv.float64Val + rv.float64Val)
elif lv.kind == vkInt64 and rv.kind == vkFloat64:
return Value(kind: vkFloat64, float64Val: float(lv.int64Val) + rv.float64Val)
elif lv.kind == vkFloat64 and rv.kind == vkInt64:
return Value(kind: vkFloat64, float64Val: lv.float64Val + float(rv.int64Val))
elif lv.kind == vkString and rv.kind == vkString:
return Value(kind: vkString, strVal: lv.strVal & rv.strVal)
else:
return Value(kind: vkNull)
of irSub:
let lv = evalExprValue(expr.binLeft, row, ctx)
let rv = evalExprValue(expr.binRight, row, ctx)
if lv.kind == vkInt64 and rv.kind == vkInt64:
return Value(kind: vkInt64, int64Val: lv.int64Val - rv.int64Val)
elif lv.kind == vkFloat64 and rv.kind == vkFloat64:
return Value(kind: vkFloat64, float64Val: lv.float64Val - rv.float64Val)
elif lv.kind == vkInt64 and rv.kind == vkFloat64:
return Value(kind: vkFloat64, float64Val: float(lv.int64Val) - rv.float64Val)
elif lv.kind == vkFloat64 and rv.kind == vkInt64:
return Value(kind: vkFloat64, float64Val: lv.float64Val - float(rv.int64Val))
else:
return Value(kind: vkNull)
of irMul:
let lv = evalExprValue(expr.binLeft, row, ctx)
let rv = evalExprValue(expr.binRight, row, ctx)
if lv.kind == vkInt64 and rv.kind == vkInt64:
return Value(kind: vkInt64, int64Val: lv.int64Val * rv.int64Val)
elif lv.kind == vkFloat64 and rv.kind == vkFloat64:
return Value(kind: vkFloat64, float64Val: lv.float64Val * rv.float64Val)
elif lv.kind == vkInt64 and rv.kind == vkFloat64:
return Value(kind: vkFloat64, float64Val: float(lv.int64Val) * rv.float64Val)
elif lv.kind == vkFloat64 and rv.kind == vkInt64:
return Value(kind: vkFloat64, float64Val: lv.float64Val * float(rv.int64Val))
else:
return Value(kind: vkNull)
of irDiv:
let lv = evalExprValue(expr.binLeft, row, ctx)
let rv = evalExprValue(expr.binRight, row, ctx)
var rvf = 0.0
if rv.kind == vkFloat64: rvf = rv.float64Val
elif rv.kind == vkInt64: rvf = float(rv.int64Val)
else: return Value(kind: vkNull)
if rvf == 0.0: return Value(kind: vkNull)
var lvf = 0.0
if lv.kind == vkFloat64: lvf = lv.float64Val
elif lv.kind == vkInt64: lvf = float(lv.int64Val)
else: return Value(kind: vkNull)
return Value(kind: vkFloat64, float64Val: lvf / rvf)
of irMod:
let lv = evalExprValue(expr.binLeft, row, ctx)
let rv = evalExprValue(expr.binRight, row, ctx)
if lv.kind == vkInt64 and rv.kind == vkInt64:
if rv.int64Val == 0: return Value(kind: vkNull)
return Value(kind: vkInt64, int64Val: lv.int64Val mod rv.int64Val)
else:
return Value(kind: vkNull)
of irPow:
let lv = evalExprValue(expr.binLeft, row, ctx)
let rv = evalExprValue(expr.binRight, row, ctx)
var lvf = 0.0
if lv.kind == vkFloat64: lvf = lv.float64Val
elif lv.kind == vkInt64: lvf = float(lv.int64Val)
else: return Value(kind: vkNull)
var rvf = 0.0
if rv.kind == vkFloat64: rvf = rv.float64Val
elif rv.kind == vkInt64: rvf = float(rv.int64Val)
else: return Value(kind: vkNull)
return Value(kind: vkFloat64, float64Val: pow(lvf, rvf))
else:
let s = evalExpr(expr, row, ctx)
return Value(kind: vkString, strVal: s)
of irekUnary:
case expr.unOp
of irNeg:
let v = evalExprValue(expr.unExpr, row, ctx)
case v.kind
of vkInt64: return Value(kind: vkInt64, int64Val: -v.int64Val)
of vkFloat64: return Value(kind: vkFloat64, float64Val: -v.float64Val)
else: return Value(kind: vkNull)
else:
let s = evalExpr(expr, row, ctx)
return Value(kind: vkString, strVal: s)
else:
let s = evalExpr(expr, row, ctx)
return Value(kind: vkString, strVal: s)
proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string = proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = nil): string =
if expr == nil: return "" if expr == nil: return ""
case expr.kind case expr.kind
@@ -580,47 +772,16 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
of irOr: of irOr:
if left == "true" or right == "true": return "true" if left == "true" or right == "true": return "true"
return "false" return "false"
of irAdd: of irAdd, irSub, irMul, irDiv, irMod, irPow:
try: let v = evalExprValue(expr, row, ctx)
let sum = parseFloat(left) + parseFloat(right) case v.kind
if sum == float(int(sum)): of vkInt64: return $v.int64Val
return $int(sum) of vkFloat64:
return $sum let s = $v.float64Val
except: return left & right if s.endsWith(".0"): return s[0..^3]
of irSub: return s
try: of vkString: return v.strVal
let diff = parseFloat(left) - parseFloat(right) else: return "\\N"
if diff == float(int(diff)):
return $int(diff)
return $diff
except: return "0"
of irMul:
try:
let prod = parseFloat(left) * parseFloat(right)
if prod == float(int(prod)):
return $int(prod)
return $prod
except: return "0"
of irDiv:
try:
let r = parseFloat(right)
if r != 0:
let quot = parseFloat(left) / r
if quot == float(int(quot)):
return $int(quot)
return $quot
return "0"
except: return "0"
of irMod:
try:
let a = parseInt(left)
let b = parseInt(right)
if b != 0: return $(a mod b)
return "0"
except: return "0"
of irPow:
try: return $(pow(parseFloat(left), parseFloat(right)))
except: return "0"
of irLike: of irLike:
proc escapeRe(s: string): string = proc escapeRe(s: string): string =
result = "" result = ""
@@ -805,14 +966,14 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
let v = evalExpr(expr.unExpr, row, ctx) let v = evalExpr(expr.unExpr, row, ctx)
return if not isNull(v): "true" else: "false" return if not isNull(v): "true" else: "false"
of irNeg: of irNeg:
let v = evalExpr(expr.unExpr, row, ctx) let v = evalExprValue(expr.unExpr, row, ctx)
try: case v.kind
let f = -parseFloat(v) of vkInt64: return $(-v.int64Val)
let s = $f of vkFloat64:
if s.endsWith(".0"): let s = $(-v.float64Val)
return s[0..^3] if s.endsWith(".0"): return s[0..^3]
return s return s
except: return "0" else: return "0"
else: return "false" else: return "false"
of irekFuncCall: of irekFuncCall:
let fn = expr.irFunc.toLower() let fn = expr.irFunc.toLower()
@@ -1410,19 +1571,19 @@ proc lowerExpr*(node: Node): IRExpr =
if node == nil: return nil if node == nil: return nil
case node.kind case node.kind
of nkIntLit: of nkIntLit:
result = IRExpr(kind: irekLiteral) result = IRExpr(kind: irekLiteral, valueKind: vkInt64)
result.literal = IRLiteral(kind: vkInt64, int64Val: node.intVal) result.literal = IRLiteral(kind: vkInt64, int64Val: node.intVal)
of nkFloatLit: of nkFloatLit:
result = IRExpr(kind: irekLiteral) result = IRExpr(kind: irekLiteral, valueKind: vkFloat64)
result.literal = IRLiteral(kind: vkFloat64, float64Val: node.floatVal) result.literal = IRLiteral(kind: vkFloat64, float64Val: node.floatVal)
of nkStringLit: of nkStringLit:
result = IRExpr(kind: irekLiteral) result = IRExpr(kind: irekLiteral, valueKind: vkString)
result.literal = IRLiteral(kind: vkString, strVal: node.strVal) result.literal = IRLiteral(kind: vkString, strVal: node.strVal)
of nkBoolLit: of nkBoolLit:
result = IRExpr(kind: irekLiteral) result = IRExpr(kind: irekLiteral, valueKind: vkBool)
result.literal = IRLiteral(kind: vkBool, boolVal: node.boolVal) result.literal = IRLiteral(kind: vkBool, boolVal: node.boolVal)
of nkNullLit: of nkNullLit:
result = IRExpr(kind: irekLiteral) result = IRExpr(kind: irekLiteral, valueKind: vkNull)
result.literal = IRLiteral(kind: vkNull) result.literal = IRLiteral(kind: vkNull)
of nkCurrentUser: of nkCurrentUser:
result = IRExpr(kind: irekFuncCall) result = IRExpr(kind: irekFuncCall)
@@ -1433,10 +1594,10 @@ proc lowerExpr*(node: Node): IRExpr =
result.irFunc = "current_role" result.irFunc = "current_role"
result.irFuncArgs = @[] result.irFuncArgs = @[]
of nkIdent: of nkIdent:
result = IRExpr(kind: irekField) result = IRExpr(kind: irekField, valueKind: vkString)
result.fieldPath = @[node.identName] result.fieldPath = @[node.identName]
of nkPath: of nkPath:
result = IRExpr(kind: irekField) result = IRExpr(kind: irekField, valueKind: vkString)
result.fieldPath = node.pathParts result.fieldPath = node.pathParts
of nkJsonPath: of nkJsonPath:
result = IRExpr(kind: irekJsonPath) result = IRExpr(kind: irekJsonPath)
@@ -1445,6 +1606,7 @@ proc lowerExpr*(node: Node): IRExpr =
result.jpAsText = node.jpAsText result.jpAsText = node.jpAsText
of nkBinOp: of nkBinOp:
result = IRExpr(kind: irekBinary) result = IRExpr(kind: irekBinary)
result.valueKind = vkString
var irOp: IROperator var irOp: IROperator
case node.binOp case node.binOp
of bkAdd: irOp = irAdd of bkAdd: irOp = irAdd
@@ -1470,18 +1632,39 @@ proc lowerExpr*(node: Node): IRExpr =
result.binOp = irOp result.binOp = irOp
result.binLeft = lowerExpr(node.binLeft) result.binLeft = lowerExpr(node.binLeft)
result.binRight = lowerExpr(node.binRight) result.binRight = lowerExpr(node.binRight)
# Infer valueKind for arithmetic operators
case irOp
of irAdd, irSub, irMul:
if result.binLeft != nil and result.binRight != nil:
if result.binLeft.valueKind == vkFloat64 or result.binRight.valueKind == vkFloat64:
result.valueKind = vkFloat64
elif result.binLeft.valueKind == vkInt64 and result.binRight.valueKind == vkInt64:
result.valueKind = vkInt64
of irDiv:
result.valueKind = vkFloat64
of irMod:
result.valueKind = vkInt64
of irPow:
result.valueKind = vkFloat64
of irEq, irNeq, irLt, irLte, irGt, irGte, irAnd, irOr,
irIn, irNotIn, irLike, irILike, irBetween,
irIsNull, irIsNotNull, irFtsMatch:
result.valueKind = vkBool
else: discard
of nkUnaryOp: of nkUnaryOp:
result = IRExpr(kind: irekUnary) result = IRExpr(kind: irekUnary, valueKind: vkString)
result.unOp = if node.unOp == ukNot: irNot else: irNeg result.unOp = if node.unOp == ukNot: irNot else: irNeg
result.unExpr = lowerExpr(node.unOperand) result.unExpr = lowerExpr(node.unOperand)
if node.unOp == ukNeg and result.unExpr != nil:
result.valueKind = result.unExpr.valueKind
of nkFuncCall: of nkFuncCall:
case node.funcName.toLower() case node.funcName.toLower()
of "count", "sum", "avg", "min", "max", "array_agg", "string_agg": of "count", "sum", "avg", "min", "max", "array_agg", "string_agg":
result = IRExpr(kind: irekAggregate) result = IRExpr(kind: irekAggregate)
case node.funcName.toLower() case node.funcName.toLower()
of "count": result.aggOp = irCount of "count": result.aggOp = irCount; result.valueKind = vkInt64
of "sum": result.aggOp = irSum of "sum": result.aggOp = irSum; result.valueKind = vkFloat64
of "avg": result.aggOp = irAvg of "avg": result.aggOp = irAvg; result.valueKind = vkFloat64
of "min": result.aggOp = irMin of "min": result.aggOp = irMin
of "max": result.aggOp = irMax of "max": result.aggOp = irMax
of "array_agg": result.aggOp = irArrayAgg of "array_agg": result.aggOp = irArrayAgg
@@ -1492,27 +1675,27 @@ proc lowerExpr*(node: Node): IRExpr =
if node.funcFilter != nil: if node.funcFilter != nil:
result.aggFilter = lowerExpr(node.funcFilter) result.aggFilter = lowerExpr(node.funcFilter)
else: else:
result = IRExpr(kind: irekFuncCall) result = IRExpr(kind: irekFuncCall, valueKind: vkString)
result.irFunc = node.funcName result.irFunc = node.funcName
result.irFuncArgs = @[] result.irFuncArgs = @[]
for arg in node.funcArgs: result.irFuncArgs.add(lowerExpr(arg)) for arg in node.funcArgs: result.irFuncArgs.add(lowerExpr(arg))
of nkIsExpr: of nkIsExpr:
result = IRExpr(kind: irekUnary) result = IRExpr(kind: irekUnary, valueKind: vkBool)
result.unOp = if node.isNegated: irIsNotNull else: irIsNull result.unOp = if node.isNegated: irIsNotNull else: irIsNull
result.unExpr = lowerExpr(node.isExpr) result.unExpr = lowerExpr(node.isExpr)
of nkLikeExpr: of nkLikeExpr:
result = IRExpr(kind: irekBinary) result = IRExpr(kind: irekBinary, valueKind: vkBool)
result.binOp = if node.likeCaseInsensitive: irILike else: irLike result.binOp = if node.likeCaseInsensitive: irILike else: irLike
result.binLeft = lowerExpr(node.likeExpr) result.binLeft = lowerExpr(node.likeExpr)
result.binRight = lowerExpr(node.likePattern) result.binRight = lowerExpr(node.likePattern)
of nkBetweenExpr: of nkBetweenExpr:
result = IRExpr(kind: irekBinary) result = IRExpr(kind: irekBinary, valueKind: vkBool)
result.binOp = irAnd result.binOp = irAnd
let leftCmp = IRExpr(kind: irekBinary) let leftCmp = IRExpr(kind: irekBinary, valueKind: vkBool)
leftCmp.binOp = irGte leftCmp.binOp = irGte
leftCmp.binLeft = lowerExpr(node.betweenExpr) leftCmp.binLeft = lowerExpr(node.betweenExpr)
leftCmp.binRight = lowerExpr(node.betweenLow) leftCmp.binRight = lowerExpr(node.betweenLow)
let rightCmp = IRExpr(kind: irekBinary) let rightCmp = IRExpr(kind: irekBinary, valueKind: vkBool)
rightCmp.binOp = irLte rightCmp.binOp = irLte
rightCmp.binLeft = lowerExpr(node.betweenExpr) rightCmp.binLeft = lowerExpr(node.betweenExpr)
rightCmp.binRight = lowerExpr(node.betweenHigh) rightCmp.binRight = lowerExpr(node.betweenHigh)
@@ -2345,7 +2528,12 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] =
result.add(padded) result.add(padded)
return result return result
# Non-LATERAL: standard join execution # Non-LATERAL: choose join strategy
chooseJoinStrategy(ctx, plan)
# Hash join and index nested loop are only safe for INNER JOIN
# (padding logic for outer joins is complex)
if plan.joinKind != irjkInner and plan.joinStrategy in {irjsHash, irjsIndexNestedLoop}:
plan.joinStrategy = irjsNestedLoop
let rightRows = executePlan(ctx, plan.joinRight) let rightRows = executePlan(ctx, plan.joinRight)
# Collect all unique column names from each side (excluding internal $ keys) # Collect all unique column names from each side (excluding internal $ keys)
@@ -2368,52 +2556,252 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] =
result.add(mergeRow(l, r, leftAlias, rightAlias)) result.add(mergeRow(l, r, leftAlias, rightAlias))
return result return result
for l in leftRows: case plan.joinStrategy
var matched = false of irjsHash:
for r in rightRows: # Hash Join: build hash table on the smaller side
let merged = mergeRow(l, r, leftAlias, rightAlias) let (leftCol, rightCol) = extractJoinEquality(plan.joinCond)
if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": var buildRows, probeRows: seq[Row]
result.add(merged) var buildCol, probeCol: string
matched = true var buildAlias, probeAlias: string
if not matched and (plan.joinKind == irjkLeft or plan.joinKind == irjkFull): var buildIsLeft: bool
var padded = initTable[string, string]()
for k, v in l:
if not k.startsWith("$"):
padded[k] = v
for col in rightCols:
if col notin padded: padded[col] = "\\N"
if leftAlias.len > 0:
for k, v in l:
if not k.startsWith("$"):
padded[leftAlias & "." & k] = v
if rightAlias.len > 0:
for col in rightCols:
padded[rightAlias & "." & col] = "\\N"
result.add(padded)
if plan.joinKind == irjkRight or plan.joinKind == irjkFull: if leftRows.len <= rightRows.len:
for r in rightRows: buildRows = leftRows
var found = false buildCol = leftCol
for l in leftRows: buildAlias = leftAlias
probeRows = rightRows
probeCol = rightCol
probeAlias = rightAlias
buildIsLeft = true
else:
buildRows = rightRows
buildCol = rightCol
buildAlias = rightAlias
probeRows = leftRows
probeCol = leftCol
probeAlias = leftAlias
buildIsLeft = false
var hashTable = initTable[string, seq[Row]]()
for row in buildRows:
let key = if buildAlias.len > 0 and buildCol in row: row[buildCol]
elif buildCol in row: row[buildCol]
else: ""
if key.len > 0 and key != "\\N":
if key notin hashTable: hashTable[key] = @[]
hashTable[key].add(row)
var matchedProbe = initTable[int, bool]()
for i, prow in probeRows:
let key = if probeAlias.len > 0 and probeCol in prow: prow[probeCol]
elif probeCol in prow: prow[probeCol]
else: ""
if key in hashTable:
matchedProbe[i] = true
for brow in hashTable[key]:
if buildIsLeft:
result.add(mergeRow(brow, prow, leftAlias, rightAlias))
else:
result.add(mergeRow(prow, brow, leftAlias, rightAlias))
# LEFT / RIGHT / FULL padding for non-matching probe rows
if plan.joinKind == irjkLeft or plan.joinKind == irjkFull or plan.joinKind == irjkRight:
for i, prow in probeRows:
if not matchedProbe.getOrDefault(i, false):
var padded = initTable[string, string]()
for k, v in prow:
if not k.startsWith("$"):
padded[k] = v
if buildIsLeft:
# probe is right side
for col in leftCols:
if col notin padded: padded[col] = "\\N"
if probeAlias.len > 0:
for k, v in prow:
if not k.startsWith("$"):
padded[probeAlias & "." & k] = v
if leftAlias.len > 0:
for col in leftCols:
padded[leftAlias & "." & col] = "\\N"
else:
# probe is left side
for col in rightCols:
if col notin padded: padded[col] = "\\N"
if probeAlias.len > 0:
for k, v in prow:
if not k.startsWith("$"):
padded[probeAlias & "." & k] = v
if rightAlias.len > 0:
for col in rightCols:
padded[rightAlias & "." & col] = "\\N"
result.add(padded)
# RIGHT / FULL padding for non-matching build rows
if plan.joinKind == irjkRight or plan.joinKind == irjkFull:
for i, brow in buildRows:
var found = false
for j, prow in probeRows:
let pkey = if probeAlias.len > 0 and probeCol in prow: prow[probeCol]
elif probeCol in prow: prow[probeCol]
else: ""
let bkey = if buildAlias.len > 0 and buildCol in brow: brow[buildCol]
elif buildCol in brow: brow[buildCol]
else: ""
if pkey == bkey and bkey.len > 0:
found = true
break
if not found:
var padded = initTable[string, string]()
for k, v in brow:
if not k.startsWith("$"):
padded[k] = v
if buildIsLeft:
for col in rightCols:
if col notin padded: padded[col] = "\\N"
if leftAlias.len > 0:
for k, v in brow:
if not k.startsWith("$"):
padded[leftAlias & "." & k] = v
if rightAlias.len > 0:
for col in rightCols:
padded[rightAlias & "." & col] = "\\N"
else:
for col in leftCols:
if col notin padded: padded[col] = "\\N"
if rightAlias.len > 0:
for k, v in brow:
if not k.startsWith("$"):
padded[rightAlias & "." & k] = v
if leftAlias.len > 0:
for col in leftCols:
padded[leftAlias & "." & col] = "\\N"
result.add(padded)
of irjsIndexNestedLoop:
let (leftCol, rightCol) = extractJoinEquality(plan.joinCond)
var outerRows: seq[Row]
var outerAlias, innerAlias: string
var outerCol, innerCol: string
var idxName: string
var outerIsLeft: bool
if plan.joinHashCol == rightCol:
outerRows = leftRows
outerAlias = leftAlias
innerAlias = rightAlias
outerCol = leftCol
innerCol = rightCol
idxName = (if plan.joinRight != nil and plan.joinRight.kind == irpkScan: plan.joinRight.scanTable else: "") & "." & rightCol
outerIsLeft = true
else:
outerRows = rightRows
outerAlias = rightAlias
innerAlias = leftAlias
outerCol = rightCol
innerCol = leftCol
idxName = (if plan.joinLeft != nil and plan.joinLeft.kind == irpkScan: plan.joinLeft.scanTable else: "") & "." & leftCol
outerIsLeft = false
if idxName notin ctx.btrees:
# Fallback to nested loop if index disappeared
plan.joinStrategy = irjsNestedLoop
# Fall through to nested loop below
else:
let btree = ctx.btrees[idxName]
var matchedOuter = initTable[int, bool]()
for i, orow in outerRows:
let key = if orow.hasKey(outerCol): orow[outerCol] else: ""
if key.len > 0 and key != "\\N":
let entries = btree.get(key)
if entries.len > 0:
matchedOuter[i] = true
for entry in entries:
let parsed = parseRowData(entry.rowValue)
if outerIsLeft:
result.add(mergeRow(orow, parsed, leftAlias, rightAlias))
else:
result.add(mergeRow(parsed, orow, leftAlias, rightAlias))
elif plan.joinKind == irjkLeft or plan.joinKind == irjkRight or plan.joinKind == irjkFull:
matchedOuter[i] = false
# Padding for non-matching outer rows
if plan.joinKind == irjkLeft or plan.joinKind == irjkRight or plan.joinKind == irjkFull:
for i, orow in outerRows:
if not matchedOuter.getOrDefault(i, false):
var padded = initTable[string, string]()
for k, v in orow:
if not k.startsWith("$"):
padded[k] = v
if outerIsLeft:
for col in rightCols:
if col notin padded: padded[col] = "\\N"
if leftAlias.len > 0:
for k, v in orow:
if not k.startsWith("$"):
padded[leftAlias & "." & k] = v
if rightAlias.len > 0:
for col in rightCols:
padded[rightAlias & "." & col] = "\\N"
else:
for col in leftCols:
if col notin padded: padded[col] = "\\N"
if rightAlias.len > 0:
for k, v in orow:
if not k.startsWith("$"):
padded[rightAlias & "." & k] = v
if leftAlias.len > 0:
for col in leftCols:
padded[leftAlias & "." & col] = "\\N"
result.add(padded)
return result
of irjsNestedLoop:
for l in leftRows:
var matched = false
for r in rightRows:
let merged = mergeRow(l, r, leftAlias, rightAlias) let merged = mergeRow(l, r, leftAlias, rightAlias)
if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true": if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true":
found = true result.add(merged)
break matched = true
if not found: if not matched and (plan.joinKind == irjkLeft or plan.joinKind == irjkFull):
var padded = initTable[string, string]() var padded = initTable[string, string]()
for k, v in r: for k, v in l:
if not k.startsWith("$"): if not k.startsWith("$"):
padded[k] = v padded[k] = v
for col in leftCols: for col in rightCols:
if col notin padded: padded[col] = "\\N" if col notin padded: padded[col] = "\\N"
if leftAlias.len > 0:
for k, v in l:
if not k.startsWith("$"):
padded[leftAlias & "." & k] = v
if rightAlias.len > 0: if rightAlias.len > 0:
for col in rightCols:
padded[rightAlias & "." & col] = "\\N"
result.add(padded)
if plan.joinKind == irjkRight or plan.joinKind == irjkFull:
for r in rightRows:
var found = false
for l in leftRows:
let merged = mergeRow(l, r, leftAlias, rightAlias)
if plan.joinCond == nil or evalExpr(plan.joinCond, merged, ctx) == "true":
found = true
break
if not found:
var padded = initTable[string, string]()
for k, v in r: for k, v in r:
if not k.startsWith("$"): if not k.startsWith("$"):
padded[rightAlias & "." & k] = v padded[k] = v
if leftAlias.len > 0:
for col in leftCols: for col in leftCols:
padded[leftAlias & "." & col] = "\\N" if col notin padded: padded[col] = "\\N"
result.add(padded) if rightAlias.len > 0:
for k, v in r:
if not k.startsWith("$"):
padded[rightAlias & "." & k] = v
if leftAlias.len > 0:
for col in leftCols:
padded[leftAlias & "." & col] = "\\N"
result.add(padded)
return result return result
+8
View File
@@ -69,6 +69,11 @@ type
irjkFull irjkFull
irjkCross irjkCross
IRJoinStrategy* = enum
irjsNestedLoop
irjsHash
irjsIndexNestedLoop
IRPlanKind* = enum IRPlanKind* = enum
irpkScan irpkScan
irpkFilter irpkFilter
@@ -123,6 +128,8 @@ type
joinCond*: IRExpr joinCond*: IRExpr
joinAlias*: string joinAlias*: string
joinLateral*: bool joinLateral*: bool
joinStrategy*: IRJoinStrategy
joinHashCol*: string
of irpkSort: of irpkSort:
sortSource*: IRPlan sortSource*: IRPlan
sortExprs*: seq[IRExpr] sortExprs*: seq[IRExpr]
@@ -197,6 +204,7 @@ type
graphReturnCols*: seq[string] graphReturnCols*: seq[string]
IRExpr* = ref object IRExpr* = ref object
valueKind*: ValueKind
case kind*: IRExprKind case kind*: IRExprKind
of irekLiteral: of irekLiteral:
literal*: IRLiteral literal*: IRLiteral
+230
View File
@@ -3498,3 +3498,233 @@ suite "Auto-Increment & ID Generators":
let r = qexec.executeQuery(ctx, parse("SELECT snowflake_id(1) AS a, snowflake_id(2) AS b")) let r = qexec.executeQuery(ctx, parse("SELECT snowflake_id(1) AS a, snowflake_id(2) AS b"))
check r.success check r.success
check r.rows[0]["a"] != r.rows[0]["b"] check r.rows[0]["a"] != r.rows[0]["b"]
suite "Deficiency Regression Tests":
# Setup for deficiency tests
setup:
var testDir = getTempDir() / "baradb_def_test_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
createDir(testDir)
var db = newLSMTree(testDir)
var ctx = qexec.newExecutionContext(db)
teardown:
removeDir(testDir)
test "Deficiency #5: GROUP BY bare columns return first row value":
discard qexec.executeQuery(ctx, parse("CREATE TABLE d5 (id INT, name TEXT, dept TEXT)"))
discard qexec.executeQuery(ctx, parse("INSERT INTO d5 VALUES (1, 'Alice', 'A')"))
discard qexec.executeQuery(ctx, parse("INSERT INTO d5 VALUES (2, 'Bob', 'A')"))
discard qexec.executeQuery(ctx, parse("INSERT INTO d5 VALUES (3, 'Carol', 'B')"))
let r = qexec.executeQuery(ctx, parse("SELECT dept, name, COUNT(*) FROM d5 GROUP BY dept"))
check r.success
check r.rows.len == 2
# dept A: first row name is Alice
var foundA, foundB = false
for row in r.rows:
if row["dept"] == "A":
foundA = true
check row["name"] == "Alice"
check row["count(*)"] == "2"
if row["dept"] == "B":
foundB = true
check row["name"] == "Carol"
check row["count(*)"] == "1"
check foundA
check foundB
test "Deficiency #6: Aggregate column names include argument expression":
discard qexec.executeQuery(ctx, parse("CREATE TABLE d6 (id INT)"))
discard qexec.executeQuery(ctx, parse("INSERT INTO d6 VALUES (1), (2)"))
let r = qexec.executeQuery(ctx, parse("SELECT count(*) AS cnt, max(id) AS mx FROM d6"))
check r.success
check r.rows.len == 1
check "cnt" in r.rows[0]
check "mx" in r.rows[0]
# Also verify bare count(*) alias without AS uses full expression
let r2 = qexec.executeQuery(ctx, parse("SELECT count(*) FROM d6"))
check r2.success
check "count(*)" in r2.rows[0]
check r2.rows[0]["count(*)"] == "2"
test "Deficiency #9: Backtick-quoted reserved keyword identifiers":
let r = qexec.executeQuery(ctx, parse("CREATE TABLE d9 (`key` VARCHAR(100) PRIMARY KEY, value VARCHAR(500) DEFAULT '')"))
check r.success
let r2 = qexec.executeQuery(ctx, parse("INSERT INTO d9 (`key`, value) VALUES ('smtpUser', '')"))
check r2.success
let r3 = qexec.executeQuery(ctx, parse("SELECT `key`, value FROM d9"))
check r3.success
check r3.rows.len == 1
check r3.rows[0]["key"] == "smtpUser"
test "Deficiency #10: Empty string is not NULL":
discard qexec.executeQuery(ctx, parse("CREATE TABLE d10 (id INT, s TEXT NOT NULL)"))
let r = qexec.executeQuery(ctx, parse("INSERT INTO d10 (id, s) VALUES (1, '')"))
check r.success
let r2 = qexec.executeQuery(ctx, parse("SELECT s FROM d10"))
check r2.success
check r2.rows[0]["s"] == ""
let r3 = qexec.executeQuery(ctx, parse("SELECT count(*) FROM d10 WHERE s = ''"))
check r3.success
check r3.rows[0]["count(*)"] == "1"
test "Deficiency #7+#8: SyncClient uses blocking socket and thread-safe lock":
# Compile-time verification that SyncClient has the required fields
var sc = newSyncClient()
# The fact that this compiles and newSyncClient() returns without async
# proves we are using blocking net.Socket, not AsyncSocket + waitFor.
# The internal lock is initialized in newSyncClient and protects query().
check true
suite "Type Safety — evalExprValue":
setup:
var testDir = getTempDir() / "baradb_type_test_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
createDir(testDir)
var db = newLSMTree(testDir)
var ctx = qexec.newExecutionContext(db)
discard qexec.executeQuery(ctx, parse("CREATE TABLE tsi (a INT, b INT)"))
discard qexec.executeQuery(ctx, parse("INSERT INTO tsi VALUES (10, 20)"))
discard qexec.executeQuery(ctx, parse("CREATE TABLE tsf (x FLOAT, y INT)"))
discard qexec.executeQuery(ctx, parse("INSERT INTO tsf VALUES (7.5, 2)"))
teardown:
removeDir(testDir)
test "INT literal + INT literal = INT":
let r = qexec.executeQuery(ctx, parse("SELECT 1 + 2 AS r"))
check r.success
check r.rows[0]["r"] == "3"
test "INT literal + FLOAT literal = FLOAT":
let r = qexec.executeQuery(ctx, parse("SELECT 1 + 2.5 AS r"))
check r.success
check r.rows[0]["r"] == "3.5"
test "FLOAT literal / INT literal = FLOAT":
let r = qexec.executeQuery(ctx, parse("SELECT 5.0 / 2 AS r"))
check r.success
check r.rows[0]["r"] == "2.5"
test "INT field + INT field = INT":
let r = qexec.executeQuery(ctx, parse("SELECT a + b AS r FROM tsi"))
check r.success
check r.rows[0]["r"] == "30"
test "INT field * INT field = INT":
let r = qexec.executeQuery(ctx, parse("SELECT a * b AS r FROM tsi"))
check r.success
check r.rows[0]["r"] == "200"
test "INT field - INT field = INT":
let r = qexec.executeQuery(ctx, parse("SELECT b - a AS r FROM tsi"))
check r.success
check r.rows[0]["r"] == "10"
test "FLOAT field / INT field = FLOAT":
let r = qexec.executeQuery(ctx, parse("SELECT x / y AS r FROM tsf"))
check r.success
check r.rows[0]["r"] == "3.75"
test "Unary negation of INT = INT":
let r = qexec.executeQuery(ctx, parse("SELECT -a AS r FROM tsi"))
check r.success
check r.rows[0]["r"] == "-10"
test "Unary negation of FLOAT = FLOAT":
let r = qexec.executeQuery(ctx, parse("SELECT -x AS r FROM tsf"))
check r.success
check r.rows[0]["r"] == "-7.5"
test "Arithmetic with table data preserves types":
let r = qexec.executeQuery(ctx, parse("SELECT a + x AS r FROM tsi, tsf"))
check r.success
check r.rows[0]["r"] == "17.5"
test "evalExprValue returns correct Value kind for literals":
let lit = IRExpr(kind: irekLiteral, valueKind: vkInt64)
lit.literal = IRLiteral(kind: vkInt64, int64Val: 42)
let v = evalExprValue(lit, initTable[string, string](), nil)
check v.kind == vkInt64
check v.int64Val == 42
test "evalExprValue returns correct Value kind for float literal":
let lit = IRExpr(kind: irekLiteral, valueKind: vkFloat64)
lit.literal = IRLiteral(kind: vkFloat64, float64Val: 3.14)
let v = evalExprValue(lit, initTable[string, string](), nil)
check v.kind == vkFloat64
check v.float64Val == 3.14
suite "Join Performance — Hash Join & Index Nested Loop":
setup:
var testDir = getTempDir() / "baradb_join_perf_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
createDir(testDir)
var db = newLSMTree(testDir)
var ctx = qexec.newExecutionContext(db)
discard qexec.executeQuery(ctx, parse("CREATE TABLE users (id INT PRIMARY KEY, name TEXT)"))
discard qexec.executeQuery(ctx, parse("CREATE TABLE orders (id INT PRIMARY KEY, user_id INT, total FLOAT)"))
discard qexec.executeQuery(ctx, parse("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Carol')"))
discard qexec.executeQuery(ctx, parse("INSERT INTO orders VALUES (10, 1, 99.5), (20, 1, 45.0), (30, 2, 150.0)"))
teardown:
removeDir(testDir)
test "Hash Join produces correct INNER JOIN results":
let r = qexec.executeQuery(ctx, parse("SELECT * FROM users u JOIN orders o ON u.id = o.user_id"))
check r.success
check r.rows.len == 3
check r.rows[0]["name"] == "Alice"
check r.rows[1]["name"] == "Alice"
check r.rows[2]["name"] == "Bob"
test "Index Nested Loop produces correct INNER JOIN results":
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_orders_user ON orders(user_id)"))
let r = qexec.executeQuery(ctx, parse("SELECT * FROM users u JOIN orders o ON u.id = o.user_id"))
check r.success
check r.rows.len == 3
check r.rows[0]["name"] == "Alice"
check r.rows[2]["name"] == "Bob"
test "Planner chooses Index Nested Loop when index exists":
discard qexec.executeQuery(ctx, parse("CREATE INDEX idx_orders_user2 ON orders(user_id)"))
let ast = parse("SELECT * FROM users u JOIN orders o ON u.id = o.user_id")
let plan = qexec.lowerSelect(ast.stmts[0])
# Walk to join node
var jp = plan
while jp != nil and jp.kind != irpkJoin:
case jp.kind
of irpkFilter: jp = jp.filterSource
of irpkProject: jp = jp.projectSource
else: break
# Execute to trigger strategy selection
discard qexec.executePlan(ctx, plan)
if jp != nil and jp.kind == irpkJoin:
check jp.joinStrategy == irjsIndexNestedLoop
else:
check false
test "Planner chooses Hash Join when no index exists":
let ast = parse("SELECT * FROM users u JOIN orders o ON u.id = o.user_id")
let plan = qexec.lowerSelect(ast.stmts[0])
var jp = plan
while jp != nil and jp.kind != irpkJoin:
case jp.kind
of irpkFilter: jp = jp.filterSource
of irpkProject: jp = jp.projectSource
else: break
discard qexec.executePlan(ctx, plan)
if jp != nil and jp.kind == irpkJoin:
check jp.joinStrategy == irjsHash
else:
check false
let r = qexec.executeQuery(ctx, parse("SELECT * FROM users u JOIN orders o ON u.id = o.user_id"))
check r.rows.len == 3
test "LEFT JOIN with Hash Join fallback to Nested Loop":
let r = qexec.executeQuery(ctx, parse("SELECT * FROM users u LEFT JOIN orders o ON u.id = o.user_id"))
check r.success
check r.rows.len == 4
check r.rows[0]["name"] == "Alice"
check r.rows[3]["name"] == "Carol"
check r.rows[3]["total"] == "\\N"