diff --git a/docs/bg/distributed.md b/docs/bg/distributed.md index f2dba09..50f857b 100644 --- a/docs/bg/distributed.md +++ b/docs/bg/distributed.md @@ -17,7 +17,7 @@ Leader election и log репликация през TCP. Включване: | `BARADB_RAFT_PEERS` | Списък `id@host:port` (вкл. себе си) | | `BARADB_RAFT_WRITE_TIMEOUT_MS` | Макс. изчакване за majority commit при SQL записи (по подразбиране 5000) | -Когато Raft е активен, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` и транзакционен `COMMIT`) се приема само от лидера на **`default`** базата: KV двойките се добавят в Raft лога и клиентът чака majority commit. Followers отказват записи с `not leader; leader is '…'`. Записи към друга database name се отказват (`raft writes only supported on the 'default' database`). Приложените записи обновяват LSM + secondary B-tree/FTS/HNSW индекси и in-memory графи на всеки възел. DDL (напр. `CREATE TABLE`) още не се репликира — схемата трябва да се създаде на всеки възел. +Когато Raft е активен, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` и транзакционен `COMMIT`) и schema DDL (`CREATE`/`DROP`/`ALTER` table, index, view, graph, …) се приемат само от лидера на **`default`** базата. DML отива като put/delete; DDL — като `ddl` запис с оригиналния SQL, преизпълнен на всеки възел при apply. Followers отказват и двете с `not leader; leader is '…'`. Записи към друга database name се отказват. `CREATE`/`DROP DATABASE` не се репликират през raft (multi-DB е извън v1). Приложен DML обновява и secondary B-tree/FTS/HNSW индекси и in-memory графи. ```nim import barabadb/core/raft diff --git a/docs/en/distributed.md b/docs/en/distributed.md index b1341fb..bc0b865 100644 --- a/docs/en/distributed.md +++ b/docs/en/distributed.md @@ -17,7 +17,7 @@ Leader election and log replication over TCP. Enable with: | `BARADB_RAFT_PEERS` | Comma-separated `id@host:port` (include self) | | `BARADB_RAFT_WRITE_TIMEOUT_MS` | Max wait for majority commit on SQL writes (default 5000) | -When Raft is enabled, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` and transactional `COMMIT`) is accepted only on the leader of the **`default`** database: each write's KV pairs are appended to the Raft log and the client waits until the entry is majority-committed. Followers reject writes with `not leader; leader is '…'`. Writes against any other database name are rejected (`raft writes only supported on the 'default' database`). Committed entries update LSM plus secondary B-tree/FTS/HNSW indexes and in-memory graphs on every node. DDL (e.g. `CREATE TABLE`) is not replicated yet — apply schema on every node. +When Raft is enabled, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` and transactional `COMMIT`) and schema DDL (`CREATE`/`DROP`/`ALTER` table, index, view, graph, …) are accepted only on the leader of the **`default`** database. DML ships as put/delete log entries; DDL ships as a `ddl` entry with the original SQL and is re-executed on every node at apply. Followers reject both with `not leader; leader is '…'`. Writes against any other database name are rejected (`raft writes only supported on the 'default' database`). `CREATE`/`DROP DATABASE` are not raft-replicated (multi-DB is out of scope for v1). Committed DML also updates secondary B-tree/FTS/HNSW indexes and in-memory graphs. ```nim import barabadb/core/raft diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 6869ba4..da318d4 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -207,6 +207,14 @@ proc valueToWire(val: string, colType: string): WireValue = return WireValue(kind: fkJson, jsonVal: val) return WireValue(kind: fkString, strVal: val) +proc waitRaftCommit(node: RaftNode, lastIdx: uint64, timeoutMs: int): Future[(bool, string)] {.async.} = + let deadline = getMonoTime() + initDuration(milliseconds = timeoutMs) + while node.commitIndex < lastIdx and getMonoTime() < deadline: + await sleepAsync(10) + if node.commitIndex < lastIdx: + return (false, "raft commit timeout") + return (true, "") + proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])], timeoutMs: int): Future[(bool, string)] {.async.} = ## C3b leader write path: append each written KV pair to the Raft log and @@ -226,12 +234,17 @@ proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])], if entry.index == 0: return (false, "lost leadership during raft append") lastIdx = entry.index - let deadline = getMonoTime() + initDuration(milliseconds = timeoutMs) - while node.commitIndex < lastIdx and getMonoTime() < deadline: - await sleepAsync(10) - if node.commitIndex < lastIdx: - return (false, "raft commit timeout") - return (true, "") + return await waitRaftCommit(node, lastIdx, timeoutMs) + +proc appendDdlToRaft*(node: RaftNode, sql: string, + timeoutMs: int): Future[(bool, string)] {.async.} = + ## C3c schema path: append one "ddl" log entry with the original SQL text. + ## Followers re-execute it via applyCommand (executor, no raft recursion). + ## MUST be called outside the storage gate (same as appendWriteToRaft). + let entry = node.appendLog("ddl", cast[seq[byte]](sql)) + if entry.index == 0: + return (false, "lost leadership during raft append") + return await waitRaftCommit(node, entry.index, timeoutMs) proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[], replication: ReplicationManager = nil, @@ -244,6 +257,7 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq var qr = QueryResult() var msg = "" var kvPairs: seq[(string, seq[byte])] = @[] + var needsRaftDdl = false withStorageGate: try: let tokens = tokenize(query) @@ -252,24 +266,23 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq if astNode.stmts.len == 0: return (true, QueryResult(), "") - # C3b: writes go through the Raft log — only the leader may accept them. - # Inspect every statement so "SELECT 1; INSERT ..." cannot bypass the gate. - # Raft state machine is wired only to the default database (v1). - if raftNode != nil: - var hasWrite = false - for stmt in astNode.stmts: - if isWrite(stmt): - hasWrite = true - break - if hasWrite: - let dbName = if ctx.currentDatabase.len > 0: ctx.currentDatabase else: "default" - if dbName != "default": - return (false, QueryResult(), - "raft writes only supported on the 'default' database; current is '" & - dbName & "'") - if raftNode.state != rsLeader: - let who = if raftNode.leaderId.len > 0: raftNode.leaderId else: "none elected" - return (false, QueryResult(), "not leader; leader is '" & who & "'") + # C3b/C3c: DML + schema DDL go through the Raft log — only the leader + # of the default database may accept them. Inspect every statement so + # "SELECT 1; INSERT/CREATE ..." cannot bypass the gate. + var hasWrite = false + needsRaftDdl = false + for stmt in astNode.stmts: + if isWrite(stmt): hasWrite = true + if isRaftDdl(stmt): needsRaftDdl = true + if raftNode != nil and (hasWrite or needsRaftDdl): + let dbName = if ctx.currentDatabase.len > 0: ctx.currentDatabase else: "default" + if dbName != "default": + return (false, QueryResult(), + "raft writes only supported on the 'default' database; current is '" & + dbName & "'") + if raftNode.state != rsLeader: + let who = if raftNode.leaderId.len > 0: raftNode.leaderId else: "none elected" + return (false, QueryResult(), "not leader; leader is '" & who & "'") let res = executor.executeQuery(ctx, astNode, params) if res.success: @@ -322,12 +335,19 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq return (false, QueryResult(), res.message) except Exception as e: return (false, QueryResult(), e.msg) - # C3b: leader appends writes to the Raft log and waits for majority commit - # (outside the storage gate — see appendWriteToRaft). - if ok and raftNode != nil and kvPairs.len > 0: - let (raftOk, raftErr) = await appendWriteToRaft(raftNode, kvPairs, raftWriteTimeoutMs) - if not raftOk: - return (false, QueryResult(), raftErr) + # Raft log append + majority wait (outside the storage gate). + # DDL batches ship the original SQL once (re-executed on apply). Pure DML + # ships KV pairs. Mixed DDL+DML in one query uses the DDL path only so the + # whole batch is re-run in order on followers. + if ok and raftNode != nil: + if needsRaftDdl: + let (raftOk, raftErr) = await appendDdlToRaft(raftNode, query, raftWriteTimeoutMs) + if not raftOk: + return (false, QueryResult(), raftErr) + elif kvPairs.len > 0: + let (raftOk, raftErr) = await appendWriteToRaft(raftNode, kvPairs, raftWriteTimeoutMs) + if not raftOk: + return (false, QueryResult(), raftErr) return (ok, qr, msg) # ---------------------------------------------------------------------- diff --git a/src/barabadb/query/exec/dml.nim b/src/barabadb/query/exec/dml.nim index 4ffb78d..c465c3e 100644 --- a/src/barabadb/query/exec/dml.nim +++ b/src/barabadb/query/exec/dml.nim @@ -488,3 +488,11 @@ proc applyReplicatedDelete*(ctx: ExecutionContext, fullKey: string) = if found and table.len > 0: removeIndexesForRow(ctx, table, fullKey, cast[string](existing)) ctx.db.delete(fullKey) + +proc isBenignRaftReplayError*(msg: string): bool = + ## Leader re-applies committed DDL/DML after local execution; followers may + ## also see IF EXISTS / race re-applies. Treat common idempotent failures as OK. + let m = msg.toLower() + "already exists" in m or "does not exist" in m or + "duplicate" in m or "unique" in m or + "unknown table" in m or "no such table" in m diff --git a/src/barabadb/query/exec/params.nim b/src/barabadb/query/exec/params.nim index 7dc64b5..73c5896 100644 --- a/src/barabadb/query/exec/params.nim +++ b/src/barabadb/query/exec/params.nim @@ -179,6 +179,17 @@ proc isDDL*(stmt: Node): bool = else: result = false +proc isRaftDdl*(stmt: Node): bool = + ## Schema changes that go through the Raft log when clustering is on. + ## CREATE/DROP DATABASE are excluded — multi-DB is out of scope for v1 raft + ## (state machine is wired only to the default database). + if not isDDL(stmt): return false + case stmt.kind + of nkCreateDatabase, nkDropDatabase: + result = false + else: + result = true + proc isWrite*(stmt: Node): bool = ## True for statements that mutate stored data. `nkCommitTxn` is included ## because COMMIT emits the transaction's buffered kvPairs. diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index ce5cbc5..43dca17 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -317,8 +317,23 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu row[colName] = w.binRight.strVal rows.add(row) return okResult(rows, coveredCols) - # Fetch actual row data from LSM - let rows = execPointRead(ctx, stmt.selFrom.fromTable, colName & "=" & w.binRight.strVal) + # Fetch full rows via the LSM keys stored in the index — never + # reconstruct the primary key from the filter column (secondary + # indexes are not the PK). + var rows: seq[Row] = @[] + for entry in entries: + if entry.lsmKey.len == 0: continue + let (found, val) = ctx.db.get(entry.lsmKey) + if found: + var row = parseRowDataToValueRow(cast[string](val)) + let prefix = stmt.selFrom.fromTable & "." + if entry.lsmKey.startsWith(prefix): + let rest = entry.lsmKey[prefix.len..^1] + row["$key"] = rest + let eqPos = rest.find('=') + if eqPos >= 0: + row[rest[0..