diff --git a/src/barabadb/query/exec/schema.nim b/src/barabadb/query/exec/schema.nim
index b87a4d5..ba7dc7c 100644
--- a/src/barabadb/query/exec/schema.nim
+++ b/src/barabadb/query/exec/schema.nim
@@ -19,6 +19,7 @@ const
   SchemaFtsIndexPrefix* = "_schema:ftsidx:"
   SchemaVecIndexPrefix* = "_schema:vecidx:"
   SchemaGraphsPrefix* = "_schema:graphs:"
+  SchemaBtreeIndexPrefix* = "_schema:btreeidx:"
   ## Legacy CREATE TABLE keys (pre-fix) used a migrations: counter suffix
   SchemaLegacyCreatePrefix* = "_schema:migrations:"
 
diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim
index 303dcbb..a8c1f19 100644
--- a/src/barabadb/query/executor.nim
+++ b/src/barabadb/query/executor.nim
@@ -878,11 +878,12 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
         dataKeys.add(key)
     for key in dataKeys:
       ctx.db.delete(key)
-    # Remove persisted FTS/HNSW index schema keys for this table
+    # Remove persisted FTS/HNSW/B-tree index schema keys for this table
     var engineKeys: seq[string] = @[]
     for (key, _) in ctx.db.scanAll():
       if key.startsWith(SchemaFtsIndexPrefix & prefix) or
-         key.startsWith(SchemaVecIndexPrefix & prefix):
+         key.startsWith(SchemaVecIndexPrefix & prefix) or
+         key.startsWith(SchemaBtreeIndexPrefix & prefix):
         engineKeys.add(key)
     for key in engineKeys:
       ctx.db.delete(key)
@@ -1387,6 +1388,15 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
       if idxVal.len > 0 and not isNull(idxVal):
         let lsmKey = if "$key" in row: stmt.ciTarget & "." & valueToString(row["$key"]) else: ""
         ctx.btrees[colKey].insert(idxVal, IndexEntry(lsmKey: lsmKey, rowValue: ""))
+    # Persist reconstructed DDL so restoreEngines can rebuild the index
+    # from table data after a restart (replay re-writes the same key).
+    # Unnamed indexes: persist the nameless form (see FTS branch above).
+    # The CREATE INDEX AST does not track UNIQUE, so it is not preserved.
+    let btreeDdl = if stmt.ciName.len > 0:
+        "CREATE INDEX " & idxName & " ON " & stmt.ciTarget & " (" & stmt.ciColumns.join(", ") & ")"
+      else:
+        "CREATE INDEX ON " & stmt.ciTarget & " (" & stmt.ciColumns.join(", ") & ")"
+    ctx.db.put(SchemaBtreeIndexPrefix & colKey, cast[seq[byte]](btreeDdl))
     return okResult(msg="CREATE INDEX " & idxName & " on " & stmt.ciTarget)
 
   of nkDropIndex:
@@ -1400,8 +1410,16 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
         targetKey = key
         found = true
         break
+      # A custom index name only appears in the persisted DDL — match it
+      # against the stored "CREATE INDEX <name> ON" text as well.
+      let (hasDdl, ddl) = ctx.db.get(SchemaBtreeIndexPrefix & key)
+      if hasDdl and cast[string](ddl).startsWith("CREATE INDEX " & stmt.diName & " ON "):
+        targetKey = key
+        found = true
+        break
     if found:
       ctx.btrees.del(targetKey)
+      ctx.db.delete(SchemaBtreeIndexPrefix & targetKey)
       return okResult(msg="DROP INDEX " & stmt.diName)
     # FTS/HNSW engine indexes: in-memory maps are keyed by table.col, and a
     # custom index name only appears in the persisted DDL — match it against
@@ -1633,14 +1651,15 @@ proc executeMigrationSql(ctx: ExecutionContext, sql: string): ExecResult =
   return okResult(msg="Empty migration body")
 
 proc restoreEngines*(ctx: ExecutionContext) =
-  ## Rebuild ephemeral engines (FTS/HNSW indexes, graphs) from persisted
+  ## Rebuild ephemeral engines (B-tree/FTS/HNSW indexes, graphs) from persisted
   ## schema keys after restoreSchema. Invoked via context.restoreEnginesHook
   ## at the end of newExecutionContext. Index replay re-persists the same
   ## key, so it is idempotent.
   var ddls: seq[string] = @[]
   for (key, value) in ctx.db.scanAll():
     if not key.startsWith(SchemaFtsIndexPrefix) and
-       not key.startsWith(SchemaVecIndexPrefix): continue
+       not key.startsWith(SchemaVecIndexPrefix) and
+       not key.startsWith(SchemaBtreeIndexPrefix): continue
     let ddl = cast[string](value)
     if ddl.len == 0: continue
     ddls.add(ddl)
diff --git a/tests/test_schema_persist.nim b/tests/test_schema_persist.nim
index 00e10fd..375900c 100644
--- a/tests/test_schema_persist.nim
+++ b/tests/test_schema_persist.nim
@@ -4,6 +4,7 @@ import std/os
 import std/strutils
 import std/tables
 import barabadb/storage/lsm
+import barabadb/storage/btree
 import barabadb/query/executor
 import barabadb/query/parser
 import barabadb/fts/engine
@@ -363,6 +364,83 @@ suite "Schema persistence":
       db2.close()
     removeDir(dir)
 
+  test "B-tree index survives reopen":
+    let dir = "/tmp/baradb_schema_persist_btree"
+    removeDir(dir)
+    block:
+      var db = newLSMTree(dir)
+      var ctx = newExecutionContext(db)
+      check execSql(ctx, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)").success
+      check execSql(ctx, "INSERT INTO users (id, name, age) VALUES (1, 'Alice', 30)").success
+      check execSql(ctx, "INSERT INTO users (id, name, age) VALUES (2, 'Bob', 25)").success
+      check execSql(ctx, "INSERT INTO users (id, name, age) VALUES (3, 'Carol', 40)").success
+      check execSql(ctx, "CREATE INDEX users_age ON users (age)").success
+      check "users.age" in ctx.btrees
+      db.close()
+    # Reopen fresh context (simulates process restart)
+    block:
+      var db2 = newLSMTree(dir)
+      var ctx2 = newExecutionContext(db2)
+      check "users.age" in ctx2.btrees
+      let r = execSql(ctx2, "SELECT id FROM users WHERE age = 25")
+      check r.success
+      check r.rows.len == 1
+      # index keeps updating after reopen
+      check execSql(ctx2, "INSERT INTO users (id, name, age) VALUES (4, 'Dan', 55)").success
+      if "users.age" in ctx2.btrees:
+        check ctx2.btrees["users.age"].get("55").len >= 1
+      db2.close()
+    removeDir(dir)
+
+  test "Unnamed B-tree index survives reopen":
+    let dir = "/tmp/baradb_schema_persist_btree_noname"
+    removeDir(dir)
+    block:
+      var db = newLSMTree(dir)
+      var ctx = newExecutionContext(db)
+      check execSql(ctx, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)").success
+      check execSql(ctx, "INSERT INTO users (id, name, age) VALUES (1, 'Alice', 30)").success
+      # No index name — idxName defaults to colKey (users.age) at execution
+      check execSql(ctx, "CREATE INDEX ON users (age)").success
+      check "users.age" in ctx.btrees
+      db.close()
+    # Reopen fresh context (simulates process restart)
+    block:
+      var db2 = newLSMTree(dir)
+      var ctx2 = newExecutionContext(db2)
+      # Persisted DDL must be replayable — the dotted fallback name is not.
+      check "users.age" in ctx2.btrees
+      let r = execSql(ctx2, "SELECT id FROM users WHERE age = 30")
+      check r.success
+      check r.rows.len == 1
+      db2.close()
+    removeDir(dir)
+
+  test "DROP INDEX removes B-tree index and its schema key":
+    let dir = "/tmp/baradb_schema_persist_dropbtree"
+    removeDir(dir)
+    block:
+      var db = newLSMTree(dir)
+      var ctx = newExecutionContext(db)
+      check execSql(ctx, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)").success
+      check execSql(ctx, "INSERT INTO users (id, name, age) VALUES (1, 'Alice', 30)").success
+      check execSql(ctx, "CREATE INDEX users_age ON users (age)").success
+      check "users.age" in ctx.btrees
+      let (foundBefore, _) = db.get(SchemaBtreeIndexPrefix & "users.age")
+      check foundBefore
+      let d = execSql(ctx, "DROP INDEX users_age")
+      check d.success
+      check "users.age" notin ctx.btrees
+      let (found, _) = db.get(SchemaBtreeIndexPrefix & "users.age")
+      check not found
+      db.close()
+    block:
+      var db2 = newLSMTree(dir)
+      var ctx2 = newExecutionContext(db2)
+      check "users.age" notin ctx2.btrees  # no ghost rebuild
+      db2.close()
+    removeDir(dir)
+
   test "Stable schema key format":
     check tableSchemaKey("users") == "_schema:tables:users"
     check serializeTableDdl(TableDef(