diff --git a/docs/bg/distributed.md b/docs/bg/distributed.md index 80e6c50..f2dba09 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`) се приема само от лидера: KV двойките се добавят в Raft лога и клиентът чака majority commit. Followers отказват записи с `not leader; leader is '…'`. Приложените записи отиват в **default** базата. DDL (напр. `CREATE TABLE`) още не се репликира — схемата трябва да се създаде на всеки възел. +Когато 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`) още не се репликира — схемата трябва да се създаде на всеки възел. ```nim import barabadb/core/raft diff --git a/docs/en/distributed.md b/docs/en/distributed.md index 167071e..b1341fb 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: 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 '…'`. Followers apply committed entries via `applyCommand` into the **default** database. 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`) 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. ```nim import barabadb/core/raft diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 63c7ebc..6869ba4 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -254,15 +254,22 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq # 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 and raftNode.state != rsLeader: - let who = if raftNode.leaderId.len > 0: raftNode.leaderId else: "none elected" - return (false, QueryResult(), "not leader; leader is '" & who & "'") + 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 & "'") let res = executor.executeQuery(ctx, astNode, params) if res.success: diff --git a/src/barabadb/graph/engine.nim b/src/barabadb/graph/engine.nim index c0af4f1..340c130 100644 --- a/src/barabadb/graph/engine.nim +++ b/src/barabadb/graph/engine.nim @@ -111,6 +111,40 @@ proc addEdgeWithId*(g: Graph, src, dst: NodeId, label: string = "", g.adjacency[src].add(AdjacencyEntry(edgeId: id, neighbor: dst, weight: weight, label: label)) g.reverseAdj[dst].add(AdjacencyEntry(edgeId: id, neighbor: src, weight: weight, label: label)) +proc hasEdgeBetween*(g: Graph, src, dst: NodeId, label: string = ""): bool = + ## True if an edge with the same endpoints and label already exists. + acquire(g.lock) + defer: release(g.lock) + for entry in g.adjacency.getOrDefault(src, @[]): + if entry.neighbor == dst and entry.label == label: + return true + return false + +proc addEdgeWithIdIfAbsent*(g: Graph, src, dst: NodeId, label: string = "", + weight: float64 = 1.0) = + ## Idempotent edge insert for raft apply / leader double-apply. + if hasEdgeBetween(g, src, dst, label): + return + addEdgeWithId(g, src, dst, label, weight) + +proc removeEdgesBetween*(g: Graph, src, dst: NodeId, label: string = "") = + ## Drop edges matching endpoints (and label if non-empty). Used by raft apply. + acquire(g.lock) + defer: release(g.lock) + if src notin g.adjacency: return + var keep: seq[AdjacencyEntry] = @[] + for entry in g.adjacency[src]: + if entry.neighbor == dst and (label.len == 0 or entry.label == label): + g.edges.del(entry.edgeId) + var newRev: seq[AdjacencyEntry] = @[] + for rev in g.reverseAdj.getOrDefault(dst, @[]): + if rev.edgeId != entry.edgeId: + newRev.add(rev) + g.reverseAdj[dst] = newRev + else: + keep.add(entry) + g.adjacency[src] = keep + proc getNode*(g: Graph, id: NodeId): GraphNode = acquire(g.lock) defer: release(g.lock) diff --git a/src/barabadb/query/exec/dml.nim b/src/barabadb/query/exec/dml.nim index 33655c8..4ffb78d 100644 --- a/src/barabadb/query/exec/dml.nim +++ b/src/barabadb/query/exec/dml.nim @@ -378,6 +378,26 @@ proc removeIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string, for ftsKey, ftsIdx in ctx.ftsIndexes: if ftsKey.startsWith(table & "."): ftsIdx.removeDocument(docId) + # In-memory graphs: drop node/edge when the backing row is removed. + for graphName, graph in ctx.graphs: + if table == graphName & "_nodes": + if "id" in oldRow: + try: + let nid = gengine.NodeId(parseUInt(valueToString(oldRow["id"]))) + gengine.removeNode(graph, nid) + except CatchableError: + discard + elif table == graphName & "_edges": + let srcStr = if "source_id" in oldRow: valueToString(oldRow["source_id"]) else: "" + let dstStr = if "dest_id" in oldRow: valueToString(oldRow["dest_id"]) else: "" + let label = if "edge_label" in oldRow: valueToString(oldRow["edge_label"]) else: "" + if srcStr.len > 0 and dstStr.len > 0: + try: + gengine.removeEdgesBetween(graph, + gengine.NodeId(parseUInt(srcStr)), + gengine.NodeId(parseUInt(dstStr)), label) + except CatchableError: + discard proc insertIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string, valStr: string) = @@ -416,9 +436,40 @@ proc insertIndexesForRow(ctx: ExecutionContext, table: string, fullKey: string, for col, val in newRow: meta[col] = valueToString(val) vengine.insert(vecIdx, docId, vec, meta) + # In-memory graphs: mirror insert/update of backing node/edge tables. + for graphName, graph in ctx.graphs: + if table == graphName & "_nodes": + if "id" notin newRow: continue + try: + let nid = gengine.NodeId(parseUInt(valueToString(newRow["id"]))) + var label = if "node_label" in newRow: valueToString(newRow["node_label"]) else: "" + var props = initTable[string, string]() + for col, val in newRow: + if col notin ["id", "node_label", "properties"]: + props[col] = valueToString(val) + # remove+add so property updates replace the in-memory node + gengine.removeNode(graph, nid) + gengine.addNodeWithId(graph, nid, label, props) + except CatchableError: + discard + elif table == graphName & "_edges": + let srcStr = if "source_id" in newRow: valueToString(newRow["source_id"]) else: "" + let dstStr = if "dest_id" in newRow: valueToString(newRow["dest_id"]) else: "" + let label = if "edge_label" in newRow: valueToString(newRow["edge_label"]) else: "" + var weight = 1.0 + if "weight" in newRow: + try: weight = parseFloat(valueToString(newRow["weight"])) + except CatchableError: discard + if srcStr.len > 0 and dstStr.len > 0: + try: + gengine.addEdgeWithIdIfAbsent(graph, + gengine.NodeId(parseUInt(srcStr)), + gengine.NodeId(parseUInt(dstStr)), label, weight) + except CatchableError: + discard proc applyReplicatedPut*(ctx: ExecutionContext, fullKey: string, value: seq[byte]) = - ## Apply a raft/replication put: LSM write + secondary B-tree/FTS/HNSW. + ## Apply a raft/replication put: LSM write + secondary B-tree/FTS/HNSW/graphs. ## Idempotent on the leader (local DML already applied the same engines). let dot = fullKey.find('.') let table = if dot > 0: fullKey[0..