feat(raft): replicate schema DDL through the raft log

- isRaftDdl + leader-only gate for CREATE/DROP/ALTER (not DATABASE)
- appendDdlToRaft ships original SQL; applyCommand re-executes via
  applyReplicatedDdl (idempotent on leader double-apply)
- Mixed DDL+DML batches use the DDL path so order is preserved
- Fix secondary-index point lookup to use entry.lsmKey (not filter col)
- E2E: CREATE only on leader, schema + index SELECT on follower

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

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

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)
 
 # ----------------------------------------------------------------------

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

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.

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..<eqPos]] = rest[eqPos+1..^1]
+                    rows.add(row)
                 let tbl = ctx.getTableDef(stmt.selFrom.fromTable)
                 var cols: seq[string] = @[]
                 for c in tbl.columns: cols.add(c.name)
@@ -1749,6 +1764,22 @@ proc restoreEngines*(ctx: ExecutionContext) =
     except CatchableError as e:
       warn("restoreEngines: graph rebuild failed for '" & name & "': " & e.msg)
 
+proc applyReplicatedDdl*(ctx: ExecutionContext, sql: string) {.gcsafe.} =
+  ## Raft "ddl" log entry: re-execute the original SQL on this node's context.
+  ## Called from applyCommand (no server/raft layer — must not re-append).
+  ## Leader double-apply failures (already exists / does not exist) are ignored
+  ## so the state machine keeps advancing. {.cast(gcsafe).} is required because
+  ## executeQuery touches the registry factory (same pattern as other engine
+  ## callbacks under the storage gate on the single-threaded apply path).
+  {.cast(gcsafe).}:
+    try:
+      let tokens = qlex.tokenize(sql)
+      let astNode = qpar.parse(tokens)
+      if astNode.stmts.len == 0: return
+      discard executeQuery(ctx, astNode)
+    except CatchableError:
+      discard
+
 # ----------------------------------------------------------------------
 # Hook wiring — breaks the module cycle between executor and the exec/*
 # submodules: eval.nim calls back into the engine for subqueries, hybrid

src/baradadb.nim

@@ -345,7 +345,7 @@ proc main() =
     raftNode.peerAddrs = config.raftPeerAddrs
     tcpServer.raftNode = raftNode  # C3b: executeQuery rejects writes on followers
     # Wire state machine: committed entries update LSM + secondary indexes
-    # (B-tree/FTS/HNSW) on every node via applyReplicatedPut/Delete.
+    # (B-tree/FTS/HNSW/graphs) and re-execute schema DDL on every node.
     let defaultDbInfo = getDatabaseInfo(registry, "default")
     raftNode.applyCommand = proc(cmd: string, data: seq[byte]) {.gcsafe.} =
       withStorageGate:
@@ -356,6 +356,8 @@ proc main() =
             applyReplicatedPut(ctx, parts[0], cast[seq[byte]](parts[1]))
         elif cmd == "delete":
           applyReplicatedDelete(ctx, cast[string](data))
+        elif cmd == "ddl":
+          applyReplicatedDdl(ctx, cast[string](data))
 
     # Wire RAFT ↔ DistTxn
     wireRaftDistTxn(raftNode, tcpServer)

tests/bugfix_test.nim

@@ -408,3 +408,11 @@ suite "Raft write classification":
     for s in ast.stmts:
       if isWrite(s): anyWrite = true
     check anyWrite
+
+  test "isRaftDdl covers schema but not CREATE DATABASE":
+    check isRaftDdl(parse("CREATE TABLE t (id INT)").stmts[0])
+    check isRaftDdl(parse("CREATE INDEX i ON t (id)").stmts[0])
+    check isRaftDdl(parse("DROP TABLE t").stmts[0])
+    check not isRaftDdl(parse("CREATE DATABASE other").stmts[0])
+    check not isRaftDdl(parse("INSERT INTO t (id) VALUES (1)").stmts[0])
+    check not isRaftDdl(parse("SELECT 1").stmts[0])

tests/raft_writes_e2e_test.nim

@@ -211,21 +211,62 @@ proc runWritesScenario() =
       return
     echo "leader elected: ", nodes[leaderIdx].id, " (term ", leaderTerm, ")"
 
-    # Schema: CREATE TABLE / INDEX are not raft writes (no kvPairs), and
-    # _schema keys are not replicated — create them locally on every node.
-    for i in 0 ..< nodes.len:
-      let db = openClient(nodes[i].clientPort)
+    let followerIdx = (if leaderIdx == 0: 1 else: 0)
+
+    # Schema: CREATE TABLE / INDEX go through the raft "ddl" log (C3c).
+    # Create only on the leader; followers must learn schema via apply.
+    block:
+      let db = openClient(nodes[leaderIdx].clientPort)
       try:
         db.exec(sql"CREATE TABLE rw_test (id INT PRIMARY KEY, name STRING)")
         db.exec(sql"CREATE INDEX idx_rw_name ON rw_test (name)")
       except CatchableError as e:
-        echo "CREATE TABLE/INDEX failed on ", nodes[i].id, ": ", e.msg
+        echo "leader CREATE TABLE/INDEX failed: ", e.msg
         dumpAll(nodes)
         fail()
         return
       db.close()
+    echo "leader schema committed via raft ddl"
 
-    let followerIdx = (if leaderIdx == 0: 1 else: 0)
+    # Follower rejection of DDL (not just DML).
+    block:
+      let db = openClient(nodes[followerIdx].clientPort)
+      var rejected = false
+      try:
+        db.exec(sql"CREATE TABLE should_fail (id INT)")
+      except CatchableError as e:
+        rejected = "not leader" in e.msg
+        if not rejected:
+          echo "follower CREATE failed without 'not leader': ", e.msg
+      db.close()
+      if not rejected:
+        echo "follower CREATE was not rejected with 'not leader'"
+        dumpAll(nodes)
+        fail()
+        return
+    echo "follower CREATE rejected with 'not leader'"
+
+    # Wait until the follower has applied CREATE TABLE (SELECT no longer
+    # errors with unknown table). Deadline 5s.
+    block:
+      let db = openClient(nodes[followerIdx].clientPort)
+      defer: db.close()
+      let start = getTime()
+      var ready = false
+      while getTime() - start < initDuration(seconds = 5):
+        try:
+          discard db.getAllRows(sql"SELECT * FROM rw_test")
+          ready = true
+          break
+        except CatchableError:
+          sleep(100)
+      if not ready:
+        echo "follower ", nodes[followerIdx].id,
+             " never applied CREATE TABLE within 5s"
+        dumpAll(nodes)
+        fail()
+        return
+    echo "schema replicated to follower ", nodes[followerIdx].id
 
     # Leader write: INSERT goes through the raft log and waits for majority
     # commit before responding (Task 2) — expect success.

tests/test_all.nim

@@ -2706,6 +2706,24 @@ suite "Raft SQL Write Path":
     check res.keyValuePairs.len == 1
     check res.keyValuePairs[0][1].len == 0
 
+  test "appendDdlToRaft fails when node is not leader":
+    var n = newRaftNode("n1", @["n2"], raftPort = 29113)
+    let (ok, err) = waitFor appendDdlToRaft(n,
+      "CREATE TABLE t (id INT)", timeoutMs = 200)
+    check not ok
+    check "lost leadership" in err
+
+  test "applyReplicatedDdl creates table on empty context":
+    var testDir = getTempDir() / "baradb_raft_ddl_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
+    createDir(testDir)
+    var db = newLSMTree(testDir)
+    var ctx = qexec.newExecutionContext(db)
+    applyReplicatedDdl(ctx, "CREATE TABLE ddl_t (id INT PRIMARY KEY, name STRING)")
+    check "ddl_t" in ctx.tables
+    # Idempotent re-apply (leader double-apply)
+    applyReplicatedDdl(ctx, "CREATE TABLE ddl_t (id INT PRIMARY KEY, name STRING)")
+    check "ddl_t" in ctx.tables
+
   test "appendWriteToRaft fails when node is not leader":
     var n = newRaftNode("n1", @["n2"], raftPort = 29111)
     # Still a follower — appendLog returns index 0.