From 6703ca2b29375a8471a0c6693dbe6117c2bc9a60 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 30 Jul 2026 15:56:48 +0300 Subject: [PATCH] feat(persist): FTS indexes survive restart (schema key + restore replay) --- src/barabadb/query/exec/context.nim | 7 +++++++ src/barabadb/query/exec/schema.nim | 1 + src/barabadb/query/executor.nim | 24 ++++++++++++++++++++++ tests/test_schema_persist.nim | 31 +++++++++++++++++++++++++++++ 4 files changed, 63 insertions(+) diff --git a/src/barabadb/query/exec/context.nim b/src/barabadb/query/exec/context.nim index 9668871..c5d08c3 100644 --- a/src/barabadb/query/exec/context.nim +++ b/src/barabadb/query/exec/context.nim @@ -15,6 +15,12 @@ import ../../vector/engine as vengine import types import schema +## Wired by executor.nim at module load. Breaks the context <-> executor +## module cycle: newExecutionContext cannot call executor code directly, so +## the engine-restore pass (FTS/HNSW/graph replay from persisted schema keys) +## is injected here and invoked nil-safely below. +var restoreEnginesHook*: proc(ctx: ExecutionContext) + # ---------------------------------------------------------------------- # Context management # ---------------------------------------------------------------------- @@ -39,6 +45,7 @@ proc newExecutionContext*(db: LSMTree, registry: DatabaseRegistry = nil): Execut result.sharedLock = SharedLock() initLock(result.sharedLock.lock) restoreSchema(result) + if restoreEnginesHook != nil: restoreEnginesHook(result) # ---------------------------------------------------------------------- # AST to SQL serializer (for VIEW DDL persistence) diff --git a/src/barabadb/query/exec/schema.nim b/src/barabadb/query/exec/schema.nim index 40c7092..6a3a9fa 100644 --- a/src/barabadb/query/exec/schema.nim +++ b/src/barabadb/query/exec/schema.nim @@ -16,6 +16,7 @@ const SchemaTriggerPrefix* = "_schema:triggers:" SchemaUserPrefix* = "_schema:users:" SchemaPolicyPrefix* = "_schema:policies:" + SchemaFtsIndexPrefix* = "_schema:ftsidx:" ## 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 dec6241..ab19df7 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -22,6 +22,7 @@ import ../storage/btree import ../storage/wal import ../core/mvcc import ../core/tracing +import ../core/logging import ../client/fileops import ../fts/engine as fts import ../core/registry @@ -1297,6 +1298,10 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu if text.len > 0: ftsIdx.addDocument(docId, text) ctx.ftsIndexes[colKey] = ftsIdx + # Persist reconstructed DDL so restoreEngines can rebuild the index + # from table data after a restart (replay re-writes the same key). + let ftsDdl = "CREATE INDEX " & idxName & " ON " & stmt.ciTarget & " (" & stmt.ciColumns.join(", ") & ") USING FTS" + ctx.db.put(SchemaFtsIndexPrefix & colKey, cast[seq[byte]](ftsDdl)) return okResult(msg="CREATE INDEX " & idxName & " on " & stmt.ciTarget & " USING FTS") if stmt.ciKind == ikHNSW: @@ -1563,6 +1568,24 @@ proc executeMigrationSql(ctx: ExecutionContext, sql: string): ExecResult = return executeQueryImpl(ctx, astNode) return okResult(msg="Empty migration body") +proc restoreEngines*(ctx: ExecutionContext) = + ## Rebuild ephemeral engines (FTS indexes) from persisted schema keys after + ## restoreSchema. Invoked via context.restoreEnginesHook at the end of + ## newExecutionContext. 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): continue + let ddl = cast[string](value) + if ddl.len == 0: continue + ddls.add(ddl) + for ddl in ddls: + try: + let res = executeQueryImpl(ctx, qpar.parse(qlex.tokenize(ddl))) + if not res.success: + warn("restoreEngines: replay failed for DDL '" & ddl & "': " & res.message) + except CatchableError as e: + warn("restoreEngines: replay raised for DDL '" & ddl & "': " & e.msg) + # ---------------------------------------------------------------------- # Hook wiring — breaks the module cycle between executor and the exec/* # submodules: eval.nim calls back into the engine for subqueries, hybrid @@ -1572,6 +1595,7 @@ proc executeMigrationSql(ctx: ExecutionContext, sql: string): ExecResult = eval.executePlanHook = plan_exec.executePlan eval.execScanHook = scan.execScan eval.executeQueryHook = executeQuery +context.restoreEnginesHook = restoreEngines # triggers.nim back-edge: fireTriggers executes trigger action statements # via the private dispatcher, so the lambda closes over executeQueryImpl. diff --git a/tests/test_schema_persist.nim b/tests/test_schema_persist.nim index 178dff0..e718c2a 100644 --- a/tests/test_schema_persist.nim +++ b/tests/test_schema_persist.nim @@ -6,6 +6,7 @@ import std/tables import barabadb/storage/lsm import barabadb/query/executor import barabadb/query/parser +import barabadb/fts/engine proc execSql(ctx: ExecutionContext, sql: string): ExecResult = let node = parse(sql) @@ -106,6 +107,36 @@ suite "Schema persistence": check sel.rows.len == 21 db2.close() + test "FTS index survives reopen": + let dir = "/tmp/baradb_schema_persist_fts" + removeDir(dir) + block: + var db = newLSMTree(dir) + var ctx = newExecutionContext(db) + check execSql(ctx, "CREATE TABLE docs (id INTEGER PRIMARY KEY, content TEXT)").success + check execSql(ctx, "INSERT INTO docs (id, content) VALUES (1, 'quick brown fox')").success + check execSql(ctx, "CREATE INDEX docs_fts ON docs (content) USING FTS").success + check ctx.ftsIndexes.hasKey("docs.content") + db.close() + # Reopen fresh context (simulates process restart) + block: + var db2 = newLSMTree(dir) + var ctx2 = newExecutionContext(db2) + # Index must be rebuilt from the persisted schema key — today it is + # silently missing, so FTS queries return empty results after reopen. + check ctx2.ftsIndexes.hasKey("docs.content") + if ctx2.ftsIndexes.hasKey("docs.content"): + check ctx2.ftsIndexes["docs.content"].search("quick", limit = 10).len >= 1 + let r = execSql(ctx2, "SELECT id FROM docs WHERE content @@ 'quick'") + check r.success + check r.rows.len == 1 + # index keeps updating after reopen + check execSql(ctx2, "INSERT INTO docs (id, content) VALUES (2, 'quick red fox')").success + if ctx2.ftsIndexes.hasKey("docs.content"): + check ctx2.ftsIndexes["docs.content"].search("red", limit = 10).len >= 1 + db2.close() + removeDir(dir) + test "Stable schema key format": check tableSchemaKey("users") == "_schema:tables:users" check serializeTableDdl(TableDef(