diff --git a/.gitignore b/.gitignore index 1472fa6..47087ae 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ benchmarks/bench_all benchmarks/compare clients/nim/tests/test_client src/baradadb +src/barabadb/core/raft # Temp *.tmp diff --git a/PLAN.md b/PLAN.md index 6a910fe..02e8495 100644 --- a/PLAN.md +++ b/PLAN.md @@ -1,216 +1,294 @@ -# BaraDB — План към Production (Web & ERP) +# BaraDB — Production Roadmap (Web & ERP) -## Визия -BaraDB да стане production-ready база данни, с която да се изграждат: -- **Web приложения** (блогове, е-магазини, SaaS продукти) -- **Малки ERP системи** (CRM, склад, счетоводство, invoicing) +## Vision +BaraDB as a production-ready database for: +- **Web applications** (blogs, e-commerce, SaaS) +- **Small ERP systems** (CRM, warehouse, accounting, invoicing) -> Целеви потребител: solo-dev / малък екип, който иска лесна за deploy и бърза локална база, без зависимост от PostgreSQL/MySQL. +> Target user: solo-dev / small team wanting a fast local DB without PostgreSQL/MySQL dependency. --- -## Текущо състояние (база) +## Current State (Baseline) -| Компонент | Статус | +| Component | Status | |-----------|--------| -| LSM-Tree KV store | ✅ Стабилен, thread-safe, persistent | -| HNSW векторен search | ✅ Работи с recall > 0.9 | -| TCP wire protocol | ✅ Бинарен, SELECT/INSERT/DELETE | -| Raft consensus | ✅ TCP transport, leader election | -| Graph engine | ✅ In-memory + persistence | -| Thread-safety | ✅ Coarse-grained locks | -| CI/CD | ✅ GitHub Actions | +| LSM-Tree KV store | Stable, thread-safe, persistent | +| HNSW vector search | Working, recall > 0.9 | +| TCP wire protocol | Binary, SELECT/INSERT/DELETE | +| Raft consensus | TCP transport, leader election | +| Graph engine | In-memory + persistence | +| CI/CD | GitHub Actions | +| Test suite | 56 suites, ~250 tests | -**Липсва за Web/ERP:** SQL съвместимост, ACID транзакции, HTTP API, auth, ORM, миграции, backup. +**Critical gaps for production:** +- Server bypasses IR/codegen/MVCC/schema — `executeQuery()` does lex→parse→raw LSMTree calls +- INSERT parser incomplete (no VALUES, column list, RETURNING) +- No CREATE TABLE/ALTER TABLE/DROP TABLE in parser +- MVCC not wired to query path (no BEGIN/COMMIT/ROLLBACK in server) +- B-Tree indexes not integrated with LSM-Tree +- SQL schema system not connected (EdgeQL types only) +- No HTTP REST API +- Auth not wired to server --- -## Фаза 1: Релационен engine + SQL (4–6 седмици) +## Phase 0: Pipeline Integration & Parser Completion (2–3 weeks) **← IN PROGRESS** -### 1.1 SQL парсър и AST -- Имплементирай ANSI SQL подмножество (CREATE TABLE, ALTER TABLE, DROP TABLE) -- INSERT с column list: `INSERT INTO users (name, email) VALUES ('...', '...')` -- UPDATE: `UPDATE users SET name = '...' WHERE id = 1` -- SELECT с JOIN, GROUP BY, HAVING, ORDER BY, LIMIT/OFFSET -- DELETE с WHERE -- Поддръжка на `RETURNING` клауза +### 0.1 Complete DML parser (INSERT/UPDATE/DELETE) +- INSERT with column list: `INSERT INTO t (c1, c2) VALUES (v1, v2)` +- INSERT with RETURNING clause +- UPDATE with RETURNING clause +- DELETE with RETURNING clause +- Multiple VALUES rows: `VALUES (v1), (v2), ...` -### 1.2 Типова система и constraints -- `INTEGER`, `BIGINT`, `SERIAL` (auto-increment) +### 0.2 Add SQL DDL to parser +- `CREATE TABLE` with column definitions, constraints (PK, FK, UNIQUE, NOT NULL, CHECK, DEFAULT) +- `ALTER TABLE` (ADD COLUMN, DROP COLUMN, RENAME COLUMN) +- `DROP TABLE` +- Tokens: tkCreate, tkTable, tkAlter, tkColumn, tkPrimary, tkKey, tkForeign, tkReferences, tkCascade, tkUnique, tkNotNull, tkCheck, tkDefault, tkRename, tkAdd, tkDrop + +### 0.3 SQL-compatible schema system +- SQL table catalog (separate from EdgeQL type system) +- Store schema in LSM-Tree (`_schema_tables`, `_schema_columns`, `_schema_indexes`) +- Column type enforcement during INSERT/UPDATE +- Schema validation on CREATE TABLE + +### 0.4 AST → IR lowering pass +- Convert Select AST nodes to IR plans (scan → filter → project → sort → limit) +- Convert Insert AST nodes to IR plans (values) +- Convert Update/Delete AST nodes to IR plans +- Convert CTE AST nodes to IR plans +- Lower JOINs to IR join nodes + +### 0.5 Codegen → Storage execution +- Execute StorageOp tree against LSM-Tree +- sokScan: full table scan via `scanMemTable()` / SSTable reader +- sokPointRead: key-based lookup +- sokFilter: evaluate IR expressions against rows +- sokProject: column selection +- sokSort: in-memory sort +- sokLimit: slice results +- sokInsert/sokUpdate/sokDelete: write to LSM-Tree + +### 0.6 Wire server to use pipeline +- Replace `execSelect/execInsert/execDelete` with pipeline-based execution +- Server flow: lex → parse → AST→IR lower → codegen → execute StorageOp +- Keep backward-compatible wire protocol +- All 56 existing tests must still pass + +--- + +## Phase 1: Schema & Indexes (2–3 weeks) + +### 1.1 SQL type system +- `INTEGER`, `BIGINT`, `SMALLINT`, `SERIAL` (auto-increment on INSERT) - `VARCHAR(n)`, `TEXT` - `BOOLEAN` - `TIMESTAMP`, `DATE` (ISO 8601) -- `JSON`, `JSONB` (in-memory + компресия) -- `UUID` (v4) -- Constraints: `PRIMARY KEY`, `FOREIGN KEY`, `UNIQUE`, `NOT NULL`, `CHECK`, `DEFAULT` -- Foreign key каскади: `ON DELETE CASCADE/SET NULL` +- `JSON`, `JSONB` +- `UUID` (v4 generation) +- `NUMERIC(p,s)`, `DOUBLE PRECISION`, `REAL` -### 1.3 B-Tree индекси + интеграция с query planner +### 1.2 Constraints enforcement +- PRIMARY KEY: unique index + NOT NULL +- FOREIGN KEY + ON DELETE CASCADE/SET NULL/RESTRICT +- UNIQUE: unique index +- NOT NULL: check on INSERT/UPDATE +- CHECK: evaluate expression on INSERT/UPDATE +- DEFAULT: fill missing values on INSERT + +### 1.3 B-Tree index integration - `CREATE INDEX idx_name ON table(column)` - `CREATE UNIQUE INDEX` -- Покриващ индекс (covering index) за чести колони -- Query planner да избира индекс вместо full scan -- `EXPLAIN` за анализ на заявки +- B-Tree indexes created per table column +- Query planner uses B-Tree for WHERE clauses on indexed columns +- Range scans via B-Tree leaf linked list -### 1.4 Транзакции (ACID) -- `BEGIN`, `COMMIT`, `ROLLBACK` -- Isolation level: `READ COMMITTED` (първа фаза), `REPEATABLE READ` (втора) -- Deadlock detection и timeout -- MVCC интеграция с LSM-Tree (versioned reads) -- WAL за crash recovery при транзакции +### 1.4 Query planner +- Choose index scan vs full scan based on WHERE clause +- Multi-column index support +- Covering index optimization +- `EXPLAIN` output with cost estimates +- Adaptive query reoptimization (wire up `adaptive.nim`) --- -## Фаза 2: Web API & Authentication (3–4 седмици) +## Phase 2: Transactions (2–3 weeks) -### 2.1 HTTP REST API -- `POST /query` — изпълнява SQL, връща JSON -- `GET /health` — readiness/liveness probe -- `GET /metrics` — брой заявки, latency, errors (Prometheus формат) -- JSON request/response body: - ```json - { "query": "SELECT * FROM users WHERE id = 1", "params": [] } - ``` -- Batch queries: `POST /batch` — множество заявки в едно тяло +### 2.1 Wire MVCC into server pipeline +- `BEGIN`, `COMMIT`, `ROLLBACK` commands +- Server tracks per-connection Transaction state +- All reads/writes through TxnManager +- Isolation: Read Committed (Phase 2a), Repeatable Read (Phase 2b) + +### 2.2 WAL crash recovery +- Implement REDO: replay committed WAL entries into LSM-Tree +- Implement UNDO: remove uncommitted entries on recovery +- Checkpoint markers in WAL +- Point-in-time recovery + +### 2.3 Compaction +- Implement actual SSTable merge (currently simulated) +- Read multiple SSTables, merge key-value pairs, write merged SSTable +- Level-based compaction strategy +- Background compaction scheduling + +### 2.4 Deadlock detection wiring +- Wire deadlock detection into TxnManager +- Automatic deadlock timeout and victim selection +- Client notification on rollback + +--- + +## Phase 3: HTTP REST API & Authentication (2–3 weeks) + +### 3.1 HTTP server +- HTTP/1.1 server alongside TCP wire protocol (shared port or separate) +- `POST /query` — execute SQL, return JSON +- `GET /health` — readiness/liveness +- `GET /metrics` — Prometheus format - Content-Type: `application/json` -### 2.2 Authentication & Authorization -- JWT bearer token в HTTP header `Authorization` -- `CREATE USER` / `DROP USER` / `ALTER USER` -- `GRANT` / `REVOKE` за права на таблици -- Row-Level Security (RLS): `CREATE POLICY` -- Хеширане на пароли с bcrypt/argon2 -- Rate limiting per API key / IP +### 3.2 Authentication +- `CREATE USER` / `DROP USER` / `ALTER USER` SQL +- Password hashing with argon2 +- JWT token creation with HMAC-SHA256 (replace djb2 `simpleHash`) +- `Authorization: Bearer ` in HTTP headers +- Per-user namespace isolation -### 2.3 WebSocket за real-time -- `ws://host:port/live` — subscribe към таблица/ред -- `NOTIFY` / `LISTEN` аналог -- Пуш нотификации при INSERT/UPDATE/DELETE +### 3.3 Authorization +- `GRANT` / `REVOKE` for table-level privileges (SELECT, INSERT, UPDATE, DELETE) +- Row-Level Security (RLS): `CREATE POLICY` on tables +- Wire auth into both HTTP and TCP protocol paths -### 2.4 CORS и HTTP hardening -- CORS headers за browser достъп -- TLS termination (reuse `ssl.nim`) +### 3.4 Rate limiting & TLS +- Wire RateLimiter into HTTP server (token bucket per IP) +- Wire TLS/SSL using OpenSSL FFI (not mock) +- Self-signed cert generation +- Configurable TLS via `baradadb cert create` + +--- + +## Phase 4: WebSocket & Real-time (1–2 weeks) + +### 4.1 WebSocket server +- `ws://host:port/live` — subscribe to table changes +- `SUBSCRIBE table_name` WebSocket message +- Push notifications on INSERT/UPDATE/DELETE +- `NOTIFY` / `LISTEN` analogue + +### 4.2 CORS & HTTP hardening +- CORS headers for browser access - Request size limits (10MB default) -- Connection keep-alive и HTTP/2 readiness +- Connection keep-alive +- HTTP/2 readiness (ALPN negotiation) --- -## Фаза 3: ERP фичове (4–5 седмици) +## Phase 5: ERP Features (3–4 weeks) -### 3.1 Schema migrations -- `CREATE MIGRATION` / `APPLY MIGRATION` -- Версиониране на схемата в `__schema_version` таблица -- Up/down скриптове -- Dry-run режим -- CLI: `baradadb migrate status`, `baradadb migrate up`, `baradadb migrate down` +### 5.1 Schema migrations +- `CREATE MIGRATION` → `APPLY MIGRATION` +- Versioned schema in `_schema_version` table +- Up/down migration scripts +- Dry-run mode +- CLI: `baradadb migrate status|up|down` -### 3.2 Views и materialized views -- `CREATE VIEW` — read-only virtual table -- `CREATE MATERIALIZED VIEW` — кеширана snapshot + `REFRESH` -- Поддръжка в query planner +### 5.2 Views +- `CREATE VIEW` — virtual table (stored query) +- `CREATE MATERIALIZED VIEW` — cached snapshot + `REFRESH` +- View usage in query planner -### 3.3 Triggers и stored functions -- `CREATE TRIGGER` — `BEFORE`/`AFTER` INSERT/UPDATE/DELETE -- Stored functions in Nim (compiled to UDF): - ```sql - CREATE FUNCTION total_price(quantity INT, price DECIMAL) RETURNS DECIMAL - AS 'quantity * price'; - ``` -- Функции за ERP: `vat_calc`, `currency_convert`, `invoice_number_next` +### 5.3 Triggers & stored functions +- `CREATE TRIGGER` — BEFORE/AFTER on INSERT/UPDATE/DELETE +- Stored functions in Nim (compile to UDF) +- ERP helper functions: `vat_calc`, `currency_convert`, `invoice_number_next` -### 3.4 Full-text search за ERP документи -- `CREATE FULLTEXT INDEX ON invoices(content)` -- `WHERE content @@ 'търсене'` -- Bulgarian stemming (reuse `fts/multilang.nim`) +### 5.4 Full-text search for ERP documents +- `CREATE FULLTEXT INDEX ON table(column)` +- `WHERE content @@ 'search query'` +- Bulgarian stemming integration -### 3.5 Partitioning -- `CREATE TABLE orders (...) PARTITION BY RANGE (created_at)` -- Автоматичен partition pruning в query planner -- Полезно за ERP: архивиране на стари данни +### 5.5 Partitioning +- `CREATE TABLE (...) PARTITION BY RANGE (col)` +- Auto partition pruning in query planner +- Useful for ERP: archive old data by date range --- -## Фаза 4: Production readiness & DevEx (3–4 седмици) +## Phase 6: Production Readiness (2–3 weeks) -### 4.1 Backup & Restore +### 6.1 Backup & Restore - `baradadb backup --output backup.tar.gz` - `baradadb restore --input backup.tar.gz` -- Incremental backup чрез WAL archiving +- Incremental backup via WAL archiving - Point-in-time recovery (PITR) -- Scheduled backups (cron integration) -### 4.2 Docker и deployment -- `Dockerfile` — multi-stage build с Nim -- `docker-compose.yml` — single node + volume +### 6.2 Docker & deployment +- `Dockerfile` — multi-stage build with Nim +- `docker-compose.yml` — single node - `docker-compose.raft.yml` — 3-node cluster -- Helm chart за Kubernetes (statefulset + PVC) -- Environment-based config (`BARADB_PORT`, `BARADB_DATA_DIR`, `BARADB_RAFT_PEERS`) +- Environment-based config (`BARADB_PORT`, `BARADB_DATA_DIR`) -### 4.3 Monitoring и observability -- Structured logging (JSON format) -- Prometheus `/metrics` endpoint: - - `baradb_queries_total`, `baradb_query_duration_seconds` - - `baradb_connections_active`, `baradb_storage_size_bytes` - - `baradb_replication_lag_seconds` -- OpenTelemetry tracing integration -- Slow query log (threshold configurable) +### 6.3 Monitoring +- Structured JSON logging +- Prometheus `/metrics`: `baradb_queries_total`, `baradb_query_duration_s`, `baradb_connections_active`, `baradb_storage_size_bytes` +- Slow query log (configurable threshold) +- OpenTelemetry tracing -### 4.4 ORM / Client SDK -- **Nim**: `baradb` nimble пакет — fluent query builder + миграции - ```nim - let users = db.table("users") - .where("active", "=", true) - .orderBy("created_at", "DESC") - .limit(10) - .all() - ``` -- **Python**: `pip install baradb` — async/sync client -- **JavaScript/TypeScript**: `npm install baradb` — promise-based client -- **Go**: `go get github.com/baradb/go-client` -- Connection pooling във всички клиенти - -### 4.5 Admin Dashboard (Web UI) -- Лек вграден админ панел на `http://host:port/admin` -- SQL playground с резултати в таблица -- Schema browser (таблици, колони, индекси) -- Metrics charts (latency, throughput, storage) +### 6.4 Admin dashboard +- Web UI on `http://host:port/admin` +- SQL playground with results table +- Schema browser (tables, columns, indexes) +- Metrics charts - User management UI -### 4.6 Performance tuning -- Prepared statements кеш -- Query result cache (LRU, TTL-based) -- Connection pool в сървъра (max 1000 конекции) -- Auto-compaction scheduling -- Configurable cache size за page cache +### 6.5 Client SDK improvements +- Nim: transaction API, prepared statements, auth +- Python: complete result parsing, transaction API, async support +- JavaScript: actual TCP/WebSocket connection, complete result parsing +- Go: complete result parsing, transaction API +- Rust: complete result parsing, transaction API +- Connection pooling in all clients --- -## Приоритетна матрица +## Priority Matrix -| Задача | Влияние | Трудност | Приоритет | -|--------|---------|----------|-----------| -| SQL парсър + AST | Критично | Висока | P0 | -| ACID транзакции | Критично | Висока | P0 | -| HTTP REST API | Критично | Средна | P0 | -| B-Tree индекси | Високо | Средна | P1 | -| JWT Auth + RLS | Високо | Средна | P1 | -| Schema migrations | Високо | Средна | P1 | -| Docker + Compose | Средно | Ниска | P2 | -| Backup/Restore | Средно | Средна | P2 | -| WebSocket real-time | Средно | Средна | P2 | -| Admin Dashboard | Средно | Висока | P2 | -| Views + Triggers | Ниско | Средна | P3 | -| Client SDK (ORM) | Ниско | Висока | P3 | -| Partitioning | Ниско | Висока | P3 | -| Kubernetes Helm | Ниско | Средна | P3 | +| Task | Impact | Difficulty | Priority | +|------|--------|-----------|----------| +| Pipeline integration (Phase 0) | Critical | High | **P0** | +| SQL DDL parser (Phase 0) | Critical | Medium | **P0** | +| AST→IR lowering (Phase 0) | Critical | High | **P0** | +| Codegen execution (Phase 0) | Critical | High | **P0** | +| SQL schema system (Phase 1) | Critical | High | **P0** | +| B-Tree index integration (Phase 1) | High | Medium | P1 | +| Constraint enforcement (Phase 1) | High | Medium | P1 | +| MVCC wiring (Phase 2) | Critical | High | **P0** | +| WAL recovery (Phase 2) | High | Medium | P1 | +| HTTP REST API (Phase 3) | Critical | Medium | **P0** | +| JWT Auth + RLS (Phase 3) | High | Medium | P1 | +| WebSocket real-time (Phase 4) | Medium | Medium | P2 | +| Schema migrations (Phase 5) | High | Medium | P1 | +| Backup/Restore (Phase 6) | Medium | Medium | P2 | +| Docker + Compose (Phase 6) | Medium | Low | P2 | +| Admin Dashboard (Phase 6) | Medium | High | P2 | +| Views + Triggers (Phase 5) | Low | Medium | P3 | +| Partitioning (Phase 5) | Low | High | P3 | +| Client SDK (Phase 6) | Medium | High | P2 | +| Kubernetes Helm (Phase 6) | Low | Medium | P3 | --- -## Очакван резултат +## Expected Results -- **Фаза 1:** BaraDB поддържа ANSI SQL subset с ACID транзакции. Може да замени SQLite/PostgreSQL за малки проекти. -- **Фаза 2:** REST API + auth правят базата достъпна от всякакъв web stack. WebSocket добавя real-time възможности. -- **Фаза 3:** ERP-ready фичове — migrations, views, triggers, partitioning. Може да поддържа реален бизнес софтуер. -- **Фаза 4:** Production tooling — Docker, backup, monitoring, ORM, admin UI. Solo-dev може да deploy-не за 5 минути. +- **Phase 0:** Server uses full pipeline. INSERT/UPDATE/DELETE/CREATE TABLE work properly. 56 existing tests pass + new tests. +- **Phase 1:** SQL schema with constraints, B-Tree indexes, EXPLAIN. Can define tables with PKs, FKs, and indexes. +- **Phase 2:** ACID transactions with MVCC, WAL recovery, compaction. Can use BEGIN/COMMIT/ROLLBACK. +- **Phase 3:** HTTP REST API with JWT auth, user management, rate limiting. DB accessible from browser. +- **Phase 4:** Real-time WebSocket subscriptions. Notifications on data changes. +- **Phase 5:** ERP-grade features: migrations, views, triggers, partitioning, full-text search. +- **Phase 6:** Docker, backup, monitoring, admin UI. Deploy in 5 minutes. -**Крайна оценка след плана:** от 8.5/10 към 9.5/10 — готова за production web/ERP. +**Final score after plan:** 9.5/10 — production-ready for web/ERP workloads. diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index b03d9a1..3014c0e 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -2,7 +2,6 @@ import std/asyncdispatch import std/asyncnet import std/strutils -import std/re import std/os import std/endians import config @@ -10,6 +9,7 @@ import ../protocol/wire import ../query/lexer import ../query/parser import ../query/ast +import ../query/executor import ../storage/lsm type @@ -17,6 +17,7 @@ type config: BaraConfig running: bool db: LSMTree + ctx: ExecutionContext ClientConnection = ref object socket: AsyncSocket @@ -24,7 +25,8 @@ type proc newServer*(config: BaraConfig): Server = let dataDir = config.dataDir / "server" - Server(config: config, running: false, db: newLSMTree(dataDir)) + let db = newLSMTree(dataDir) + Server(config: config, running: false, db: db, ctx: newExecutionContext(db)) # ---------------------------------------------------------------------- # Wire Protocol Helpers @@ -51,85 +53,10 @@ proc parseHeader(data: string): (bool, MessageHeader) = return (true, MessageHeader(kind: kind, length: length, requestId: requestId)) # ---------------------------------------------------------------------- -# Query Execution +# Query Execution (pipeline-based) # ---------------------------------------------------------------------- -proc extractStringLiteral(node: Node): string = - if node == nil: - return "" - case node.kind - of nkStringLit: return node.strVal - of nkIdent: return node.identName - of nkIntLit: return $node.intVal - of nkBinOp: - if node.binOp == bkEq and node.binLeft != nil and node.binRight != nil: - # Accept any identifier on the left side (e.g., key = 'x', name = 'y') - if node.binLeft.kind == nkIdent: - return extractStringLiteral(node.binRight) - if node.binRight.kind == nkIdent: - return extractStringLiteral(node.binLeft) - return "" - else: return "" - -proc execSelect(db: LSMTree, astNode: Node): QueryResult = - result = QueryResult(columns: @["key", "value"], rows: @[]) - - var keyFilter = "" - if astNode.selWhere != nil and astNode.selWhere.whereExpr != nil: - let whereExpr = astNode.selWhere.whereExpr - keyFilter = extractStringLiteral(whereExpr) - - if keyFilter != "": - # Point read - let (found, val) = db.get(keyFilter) - if found: - var row: seq[WireValue] = @[] - row.add(WireValue(kind: fkString, strVal: keyFilter)) - row.add(WireValue(kind: fkBytes, bytesVal: val)) - result.rows.add(row) - result.rowCount = 1 - else: - # Full scan of memory tables - for entry in db.scanMemTable(): - if entry.deleted: - continue - var row: seq[WireValue] = @[] - row.add(WireValue(kind: fkString, strVal: entry.key)) - row.add(WireValue(kind: fkBytes, bytesVal: entry.value)) - result.rows.add(row) - result.rowCount = result.rows.len - -proc execInsert(db: LSMTree, query: string): QueryResult = - result = QueryResult() - # Manual parsing for simple INSERT: INSERT table { field := 'value' } - # We use the value as the key for simple KV semantics - let pattern = re"INSERT\s+(\w+)\s*\{\s*(\w+)\s*:=\s*'([^']+)'\s*\}" - var matches: array[3, string] - if query.match(pattern, matches): - let key = matches[2] # use the value as key - let value = matches[2] - db.put(key, cast[seq[byte]](value)) - result.affectedRows = 1 - else: - # Try simpler pattern: INSERT table { field := value } - let pattern2 = re"INSERT\s+(\w+)\s*\{\s*(\w+)\s*:=\s*(\w+)\s*\}" - var matches2: array[3, string] - if query.match(pattern2, matches2): - let key = matches2[2] - let value = matches2[2] - db.put(key, cast[seq[byte]](value)) - result.affectedRows = 1 - -proc execDelete(db: LSMTree, astNode: Node): QueryResult = - result = QueryResult() - var keyFilter = "" - if astNode.delWhere != nil and astNode.delWhere.whereExpr != nil: - keyFilter = extractStringLiteral(astNode.delWhere.whereExpr) - if keyFilter != "": - db.delete(keyFilter) - result.affectedRows = 1 - -proc executeQuery(db: LSMTree, query: string): (bool, QueryResult, string) = +proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string): (bool, QueryResult, string) = try: let tokens = tokenize(query) let astNode = parse(tokens) @@ -139,15 +66,17 @@ proc executeQuery(db: LSMTree, query: string): (bool, QueryResult, string) = let stmt = astNode.stmts[0] case stmt.kind - of nkSelect: - let qr = execSelect(db, stmt) - return (true, qr, "") - of nkInsert: - let qr = execInsert(db, query) - return (true, qr, "") - of nkDelete: - let qr = execDelete(db, stmt) - return (true, qr, "") + of nkSelect, nkInsert, nkUpdate, nkDelete, nkCreateTable, nkDropTable, + nkCreateType, nkBeginTxn, nkCommitTxn, nkRollbackTxn, nkExplainStmt: + let (success, errMsg, affectedRows) = executor.executeQuery(ctx, astNode) + if success: + var qr = QueryResult(affectedRows: affectedRows, rowCount: affectedRows) + if stmt.kind == nkSelect: + qr.columns = @["key", "value"] + qr.rows = @[] + return (true, qr, "") + else: + return (false, QueryResult(), errMsg) else: return (false, QueryResult(), "Unsupported statement type: " & $stmt.kind) except Exception as e: @@ -237,7 +166,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} let queryStr = readString(cast[seq[byte]](payload), pos) echo "[", clientId, "] Query: ", queryStr - let (success, result, errorMsg) = executeQuery(server.db, queryStr) + let (success, result, errorMsg) = executeQuery(server.db, server.ctx, queryStr) if success: if result.rows.len > 0: let dataMsg = serializeResult(result, header.requestId) diff --git a/src/barabadb/query/ast.nim b/src/barabadb/query/ast.nim index 36774b7..dd32158 100644 --- a/src/barabadb/query/ast.nim +++ b/src/barabadb/query/ast.nim @@ -11,8 +11,15 @@ type nkCreateType nkDropType nkAlterType + nkCreateTable + nkDropTable + nkAlterTable nkCreateIndex nkDropIndex + nkBeginTxn + nkCommitTxn + nkRollbackTxn + nkExplainStmt # Clauses nkFrom @@ -63,10 +70,11 @@ type # Join nkJoin - # Type definitions + # Type definitions / DDL nkPropertyDef nkLinkDef nkIndexDef + nkColumnDef nkConstraintDef # Top-level @@ -157,6 +165,41 @@ type of nkAlterType: atName*: string atOps*: seq[Node] + of nkCreateTable: + crtName*: string + crtColumns*: seq[Node] + crtConstraints*: seq[Node] + crtIfNotExists*: bool + of nkDropTable: + drtName*: string + drtIfExists*: bool + of nkAlterTable: + altName*: string + altOps*: seq[Node] + of nkColumnDef: + cdName*: string + cdType*: string + cdConstraints*: seq[Node] + of nkConstraintDef: + cstName*: string + cstType*: string + cstExpr*: Node + cstColumns*: seq[string] + cstRefTable*: string + cstRefColumns*: seq[string] + cstOnDelete*: string + cstOnUpdate*: string + cstCheck*: Node + cstDefault*: Node + of nkBeginTxn: + btxnMode*: string + of nkCommitTxn: + ctxnChain*: bool + of nkRollbackTxn: + rtxnChain*: bool + of nkExplainStmt: + expStmt*: Node + expAnalyze*: bool of nkCreateIndex: ciTarget*: string ciName*: string @@ -301,9 +344,6 @@ type idName*: string idExpr*: Node idKind*: IndexKind - of nkConstraintDef: - cdName*: string - cdExpr*: Node of nkStatementList: stmts*: seq[Node] @@ -314,5 +354,9 @@ proc newNode*(kind: NodeKind, line, col: int = 0): Node = of nkInsert: result.insFields = @[]; result.insValues = @[] of nkUpdate: result.updSet = @[] of nkDelete: discard + of nkCreateTable: result.crtColumns = @[]; result.crtConstraints = @[] + of nkAlterTable: result.altOps = @[] + of nkColumnDef: result.cdConstraints = @[] + of nkConstraintDef: result.cstColumns = @[]; result.cstRefColumns = @[] of nkStatementList: result.stmts = @[] else: discard diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim new file mode 100644 index 0000000..a7dccc3 --- /dev/null +++ b/src/barabadb/query/executor.nim @@ -0,0 +1,398 @@ +## BaraQL Executor — AST lowering, IR compilation, and execution +import std/strutils +import std/tables +import ast +import ir +import ../core/types +import ../storage/lsm + +type + ExecutionContext* = ref object + db*: LSMTree + tables*: Table[string, TableDef] # table name -> definition + + TableDef* = object + name*: string + columns*: seq[ColumnDef] + pkColumns*: seq[string] + + ColumnDef* = object + name*: string + colType*: string + isPk*: bool + isNotNull*: bool + isUnique*: bool + defaultVal*: string + + Row = Table[string, string] + +proc newExecutionContext*(db: LSMTree): ExecutionContext = + ExecutionContext(db: db, tables: initTable[string, TableDef]()) + +proc execScan(ctx: ExecutionContext, table: string): seq[Row] = + ## Full table scan via LSM-Tree memtable scan. + ## Rows are stored as: "{table}.{key}" -> value + ## For simple KV: key is the PK value, value is the serialized row + result = @[] + let prefix = table & "." + for entry in ctx.db.scanMemTable(): + if entry.deleted: + continue + if not entry.key.startsWith(prefix): + continue + let rest = entry.key[prefix.len..^1] + var row: Table[string, string] + row["$key"] = rest + row["$value"] = cast[string](entry.value) + result.add(row) + +proc execPointRead(ctx: ExecutionContext, table: string, key: string): seq[Row] = + ## Point read from LSM-Tree + let fullKey = table & "." & key + let (found, val) = ctx.db.get(fullKey) + if found: + var row: Table[string, string] + row["$key"] = key + row["$value"] = cast[string](val) + return @[row] + return @[] + +proc execInsert(ctx: ExecutionContext, table: string, fields: seq[string], values: seq[seq[string]]): int = + ## Insert rows into LSM-Tree. + ## Each row is stored as key=table.pk_value, value= + var count = 0 + for rowVals in values: + var key = "" + var valStr = "" + for i, f in fields: + if i < rowVals.len: + if key == "": + key = f & "=" & rowVals[i] + else: + valStr &= f & "=" & rowVals[i] + if i < rowVals.len - 1: + valStr &= "," + let fullKey = table & "." & key + ctx.db.put(fullKey, cast[seq[byte]](valStr)) + inc count + return count + +proc execDelete(ctx: ExecutionContext, table: string, key: string): int = + let fullKey = table & "." & key + let (found, _) = ctx.db.get(fullKey) + if found: + ctx.db.delete(fullKey) + return 1 + return 0 + +# ---------------------------------------------------------------------- +# AST → IR Lowering +# ---------------------------------------------------------------------- + +proc lowerExpr(node: Node): IRExpr = + if node == nil: + return nil + case node.kind + of nkIntLit: + result = IRExpr(kind: irekLiteral) + result.literal = IRLiteral(kind: vkInt64, int64Val: node.intVal) + of nkFloatLit: + result = IRExpr(kind: irekLiteral) + result.literal = IRLiteral(kind: vkFloat64, float64Val: node.floatVal) + of nkStringLit: + result = IRExpr(kind: irekLiteral) + result.literal = IRLiteral(kind: vkString, strVal: node.strVal) + of nkBoolLit: + result = IRExpr(kind: irekLiteral) + result.literal = IRLiteral(kind: vkBool, boolVal: node.boolVal) + of nkNullLit: + result = IRExpr(kind: irekLiteral) + result.literal = IRLiteral(kind: vkNull) + of nkIdent: + result = IRExpr(kind: irekField) + result.fieldPath = @[node.identName] + of nkPath: + result = IRExpr(kind: irekField) + result.fieldPath = node.pathParts + of nkBinOp: + result = IRExpr(kind: irekBinary) + var irOp: IROperator + case node.binOp + of bkAdd: irOp = irAdd + of bkSub: irOp = irSub + of bkMul: irOp = irMul + of bkDiv: irOp = irDiv + of bkMod: irOp = irMod + of bkEq: irOp = irEq + of bkNotEq: irOp = irNeq + of bkLt: irOp = irLt + of bkLtEq: irOp = irLte + of bkGt: irOp = irGt + of bkGtEq: irOp = irGte + of bkAnd: irOp = irAnd + of bkOr: irOp = irOr + else: irOp = irEq + result.binOp = irOp + result.binLeft = lowerExpr(node.binLeft) + result.binRight = lowerExpr(node.binRight) + of nkUnaryOp: + result = IRExpr(kind: irekUnary) + result.unOp = if node.unOp == ukNot: irNot else: irNot + result.unExpr = lowerExpr(node.unOperand) + of nkFuncCall: + result = IRExpr(kind: irekAggregate) + case node.funcName.toLower() + of "count": result.aggOp = irCount + of "sum": result.aggOp = irSum + of "avg": result.aggOp = irAvg + of "min": result.aggOp = irMin + of "max": result.aggOp = irMax + else: result = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkNull)) + result.aggArgs = @[] + for arg in node.funcArgs: + result.aggArgs.add(lowerExpr(arg)) + of nkIsExpr: + result = IRExpr(kind: irekUnary) + result.unOp = if node.isNegated: irIsNotNull else: irIsNull + result.unExpr = lowerExpr(node.isExpr) + of nkLikeExpr: + result = IRExpr(kind: irekBinary) + result.binOp = if node.likeCaseInsensitive: irILike else: irLike + result.binLeft = lowerExpr(node.likeExpr) + result.binRight = lowerExpr(node.likePattern) + of nkBetweenExpr: + result = IRExpr(kind: irekBinary) + result.binOp = irBetween + result.binLeft = lowerExpr(node.betweenExpr) + result.binRight = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkString, strVal: "")) + of nkInExpr: + result = IRExpr(kind: irekBinary) + result.binOp = irIn + result.binLeft = lowerExpr(node.inLeft) + result.binRight = lowerExpr(node.inRight) + of nkExists: + result = IRExpr(kind: irekExists) + else: + result = IRExpr(kind: irekLiteral, literal: IRLiteral(kind: vkNull)) + +proc lowerSelect(node: Node): IRPlan = + result = IRPlan(kind: irpkScan) + if node.selFrom != nil and node.selFrom.fromTable.len > 0: + result.scanTable = node.selFrom.fromTable + result.scanAlias = node.selFrom.fromAlias + + # WHERE → Filter + if node.selWhere != nil and node.selWhere.whereExpr != nil: + let filterPlan = IRPlan(kind: irpkFilter) + filterPlan.filterSource = result + filterPlan.filterCond = lowerExpr(node.selWhere.whereExpr) + result = filterPlan + + # GROUP BY + if node.selGroupBy.len > 0: + let groupPlan = IRPlan(kind: irpkGroupBy) + groupPlan.groupSource = result + groupPlan.groupKeys = @[] + for g in node.selGroupBy: + groupPlan.groupKeys.add(lowerExpr(g)) + groupPlan.groupAggs = @[] + if node.selHaving != nil: + groupPlan.groupHaving = lowerExpr(node.selHaving.havingExpr) + result = groupPlan + + # SELECT → Project + let projectPlan = IRPlan(kind: irpkProject) + projectPlan.projectSource = result + projectPlan.projectExprs = @[] + projectPlan.projectAliases = @[] + for e in node.selResult: + projectPlan.projectExprs.add(lowerExpr(e)) + if e.kind == nkIdent: + projectPlan.projectAliases.add(e.identName) + else: + projectPlan.projectAliases.add("") + result = projectPlan + + # ORDER BY → Sort + if node.selOrderBy.len > 0: + let sortPlan = IRPlan(kind: irpkSort) + sortPlan.sortSource = result + sortPlan.sortExprs = @[] + sortPlan.sortDirs = @[] + for o in node.selOrderBy: + sortPlan.sortExprs.add(lowerExpr(o.orderByExpr)) + sortPlan.sortDirs.add(o.orderByDir == sdAsc) + result = sortPlan + + # LIMIT/OFFSET + if node.selLimit != nil or node.selOffset != nil: + let limitPlan = IRPlan(kind: irpkLimit) + limitPlan.limitSource = result + limitPlan.limitCount = if node.selLimit != nil and node.selLimit.limitExpr.kind == nkIntLit: + node.selLimit.limitExpr.intVal else: 0 + limitPlan.limitOffset = if node.selOffset != nil and node.selOffset.offsetExpr.kind == nkIntLit: + node.selOffset.offsetExpr.intVal else: 0 + result = limitPlan + +proc lowerInsert(node: Node): IRPlan = + result = IRPlan(kind: irpkInsert) + result.insertTable = node.insTarget + result.insertFields = @[] + for f in node.insFields: + if f.kind == nkIdent: + result.insertFields.add(f.identName) + else: + result.insertFields.add("") + result.insertValues = @[] + for rowNode in node.insValues: + var rowVals: seq[IRExpr] = @[] + if rowNode.kind == nkArrayLit: + for v in rowNode.arrayElems: + rowVals.add(lowerExpr(v)) + else: + rowVals.add(lowerExpr(rowNode)) + result.insertValues.add(rowVals) + +# ---------------------------------------------------------------------- +# IR Plan Execution +# ---------------------------------------------------------------------- + +proc executePlan(ctx: ExecutionContext, plan: IRPlan): seq[Row] = + if plan == nil: + return @[] + + case plan.kind + of irpkScan: + result = execScan(ctx, plan.scanTable) + + of irpkFilter: + let sourceRows = executePlan(ctx, plan.filterSource) + result = sourceRows # TODO: actual filter evaluation + + of irpkProject: + let sourceRows = executePlan(ctx, plan.projectSource) + result = sourceRows + + of irpkSort: + let sourceRows = executePlan(ctx, plan.sortSource) + result = sourceRows + + of irpkLimit: + let sourceRows = executePlan(ctx, plan.limitSource) + var start = int(plan.limitOffset) + if start > sourceRows.len: start = sourceRows.len + var endIdx = start + int(plan.limitCount) + if endIdx > sourceRows.len or plan.limitCount == 0: + endIdx = sourceRows.len + result = sourceRows[start.. means link - discard p.advance() # consume -> + if p.peek().kind == tkArrow: + discard p.advance() let target = p.expect(tkIdent).value result.ctLinks.add(Node(kind: nkLinkDef, ldName: fieldTok.value, ldTarget: target, @@ -484,13 +531,221 @@ proc parseCreateType(p: var Parser): Node = discard p.match(tkSemicolon) discard p.expect(tkRbrace) +proc parseCreateTable(p: var Parser): Node = + let tok = p.expect(tkCreate) + result = Node(kind: nkCreateTable, line: tok.line, col: tok.col) + result.crtColumns = @[] + result.crtConstraints = @[] + + # Optional IF NOT EXISTS + if p.peek().kind == tkIf: + discard p.advance() + discard p.expect(tkNot) + discard p.expect(tkExists) + result.crtIfNotExists = true + + discard p.expect(tkTable) + if p.peek().kind == tkIdent and p.peek().value.toLower() == "if": + discard p.advance() # if + discard p.expect(tkNot) + discard p.expect(tkExists) + result.crtIfNotExists = true + + result.crtName = p.expect(tkIdent).value + discard p.expect(tkLParen) + + while p.peek().kind != tkRParen and p.peek().kind != tkEof: + discard p.match(tkComma) + + # Check if this is a table-level constraint + if p.peek().kind in {tkPrimary, tkForeign, tkUnique, tkCheck}: + let cst = Node(kind: nkConstraintDef) + cst.cstColumns = @[]; cst.cstRefColumns = @[] + if p.match(tkPrimary): + discard p.expect(tkKey) + cst.cstType = "pkey" + if p.peek().kind == tkLParen: + discard p.advance() + cst.cstColumns.add(p.expect(tkIdent).value) + while p.match(tkComma): + cst.cstColumns.add(p.expect(tkIdent).value) + discard p.expect(tkRParen) + elif p.match(tkForeign): + discard p.expect(tkKey) + cst.cstType = "fkey" + if p.peek().kind == tkLParen: + discard p.advance() + cst.cstColumns.add(p.expect(tkIdent).value) + while p.match(tkComma): + cst.cstColumns.add(p.expect(tkIdent).value) + discard p.expect(tkRParen) + discard p.expect(tkReferences) + cst.cstRefTable = p.expect(tkIdent).value + if p.peek().kind == tkLParen: + discard p.advance() + cst.cstRefColumns.add(p.expect(tkIdent).value) + while p.match(tkComma): + cst.cstRefColumns.add(p.expect(tkIdent).value) + discard p.expect(tkRParen) + if p.peek().kind == tkOn: + discard p.advance() + if p.peek().kind == tkDelete: + discard p.advance() + if p.peek().kind == tkCascade: + discard p.advance() + cst.cstOnDelete = "CASCADE" + elif p.peek().kind == tkSet: + discard p.advance() + discard p.match(tkNull) + cst.cstOnDelete = "SET NULL" + elif p.match(tkUnique): + cst.cstType = "unique" + if p.peek().kind == tkLParen: + discard p.advance() + cst.cstColumns.add(p.expect(tkIdent).value) + while p.match(tkComma): + cst.cstColumns.add(p.expect(tkIdent).value) + discard p.expect(tkRParen) + elif p.match(tkCheck): + cst.cstType = "check" + discard p.expect(tkLParen) + cst.cstCheck = p.parseExpr() + discard p.expect(tkRParen) + result.crtConstraints.add(cst) + continue + + # Parse column definition + let colName = p.expect(tkIdent).value + var colType = "" + if p.peek().kind == tkIdent: + colType = p.advance().value.toUpper() + if p.peek().kind == tkLParen: + discard p.advance() + let size = p.expect(tkIntLit).value + colType &= "(" & size & ")" + discard p.expect(tkRParen) + + let colDef = Node(kind: nkColumnDef, cdName: colName, cdType: colType) + colDef.cdConstraints = @[] + + # Parse column constraints + while p.peek().kind in {tkPrimary, tkNot, tkNull, tkUnique, tkCheck, tkDefault, tkReferences}: + let cst = Node(kind: nkConstraintDef) + cst.cstColumns = @[colName]; cst.cstRefColumns = @[] + if p.match(tkPrimary): + discard p.expect(tkKey) + cst.cstType = "pkey" + elif p.match(tkNot): + discard p.expect(tkNull) + cst.cstType = "notnull" + elif p.match(tkNull): + cst.cstType = "null" + elif p.match(tkUnique): + cst.cstType = "unique" + elif p.match(tkCheck): + cst.cstType = "check" + discard p.expect(tkLParen) + cst.cstCheck = p.parseExpr() + discard p.expect(tkRParen) + elif p.match(tkDefault): + cst.cstType = "default" + cst.cstDefault = p.parseExpr() + elif p.match(tkReferences): + cst.cstType = "fkey" + cst.cstRefTable = p.expect(tkIdent).value + if p.peek().kind == tkLParen: + discard p.advance() + cst.cstRefColumns.add(p.expect(tkIdent).value) + while p.match(tkComma): + cst.cstRefColumns.add(p.expect(tkIdent).value) + discard p.expect(tkRParen) + if p.peek().kind == tkOn: + discard p.advance() + if p.peek().kind == tkDelete: + discard p.advance() + if p.peek().kind == tkCascade: + discard p.advance() + cst.cstOnDelete = "CASCADE" + colDef.cdConstraints.add(cst) + result.crtColumns.add(colDef) + + discard p.expect(tkRParen) + +proc parseDropTable(p: var Parser): Node = + let tok = p.expect(tkDrop) + result = Node(kind: nkDropTable, line: tok.line, col: tok.col) + if p.peek().kind == tkTable: + discard p.advance() + if p.peek().kind == tkIdent and p.peek().value.toLower() == "if": + discard p.advance() + discard p.expect(tkExists) + result.drtIfExists = true + result.drtName = p.expect(tkIdent).value + +proc parseAlterTable(p: var Parser): Node = + let tok = p.expect(tkAlter) + discard p.expect(tkTable) + result = Node(kind: nkAlterTable, line: tok.line, col: tok.col) + result.altName = p.expect(tkIdent).value + result.altOps = @[] + discard p.match(tkAdd) # or tkRename + +proc parseBeginTxn(p: var Parser): Node = + let tok = p.expect(tkBegin) + result = Node(kind: nkBeginTxn, line: tok.line, col: tok.col) + discard p.match(tkIdent) # optional TRANSACTION / WORK + result.btxnMode = "" + +proc parseCommitTxn(p: var Parser): Node = + let tok = p.expect(tkCommit) + result = Node(kind: nkCommitTxn, line: tok.line, col: tok.col) + discard p.match(tkIdent) # optional TRANSACTION / WORK + +proc parseRollbackTxn(p: var Parser): Node = + let tok = p.expect(tkRollback) + result = Node(kind: nkRollbackTxn, line: tok.line, col: tok.col) + discard p.match(tkIdent) # optional TRANSACTION / WORK + +proc parseExplain(p: var Parser): Node = + let tok = p.expect(tkExplain) + result = Node(kind: nkExplainStmt, line: tok.line, col: tok.col) + if p.peek().kind == tkIdent and p.peek().value.toLower() == "analyze": + discard p.advance() + result.expAnalyze = true + result.expStmt = p.parseStatement() + proc parseStatement*(p: var Parser): Node = case p.peek().kind of tkWith, tkSelect: p.parseSelect() of tkInsert: p.parseInsert() of tkUpdate: p.parseUpdate() of tkDelete: p.parseDelete() - of tkCreate: p.parseCreateType() + of tkCreate: + if p.pos + 1 < p.tokens.len: + let next = p.tokens[p.pos + 1] + if next.kind == tkTable or + (next.kind == tkIdent and next.value.toLower() == "if"): + p.parseCreateTable() + else: + p.parseCreateType() + else: + p.parseCreateType() + of tkDrop: + if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkTable: + p.parseDropTable() + else: + let tok = p.advance() + Node(kind: nkNullLit, line: tok.line, col: tok.col) + of tkAlter: + if p.pos + 1 < p.tokens.len and p.tokens[p.pos + 1].kind == tkTable: + p.parseAlterTable() + else: + let tok = p.advance() + Node(kind: nkNullLit, line: tok.line, col: tok.col) + of tkBegin: p.parseBeginTxn() + of tkCommit: p.parseCommitTxn() + of tkRollback: p.parseRollbackTxn() + of tkExplain: p.parseExplain() else: let tok = p.advance() Node(kind: nkNullLit, line: tok.line, col: tok.col)