feat: PITR RECOVER TO TIMESTAMP, OpenTelemetry OTLP export, FTS infrastructure

- RECOVER TO TIMESTAMP '...' — parser + executor, replays WAL entries
- core/tracing.nim — OTLP/HTTP export via exportOtlp()
- FTS: ftsIndexes table in ExecutionContext, engine import ready
- 281 tests, 0 failures

src/barabadb/core/tracing.nim

@@ -3,6 +3,8 @@ import std/json
 import std/times
 import std/monotimes
 import std/tables
+import std/httpclient
+import std/strutils
 
 type
   SpanStatus* = enum
@@ -82,3 +84,44 @@ proc enable*(tracer: Tracer) =
 
 proc disable*(tracer: Tracer) =
   tracer.enabled = false
+
+proc exportOtlp*(tracer: Tracer, endpoint: string = "http://localhost:4318/v1/traces"): bool =
+  ## Export spans via OTLP/HTTP (JSON format).
+  ## Returns true on success.
+  if tracer.spans.len == 0: return true
+  var otlpSpans = newJArray()
+  for span in tracer.spans:
+    var attrs = newJArray()
+    for k, v in span.attributes:
+      attrs.add(%*{"key": k, "value": {"stringValue": v}})
+    otlpSpans.add(%*{
+      "traceId": span.traceId,
+      "spanId": span.spanId,
+      "parentSpanId": span.parentSpanId,
+      "name": span.name,
+      "kind": "SPAN_KIND_INTERNAL",
+      "startTimeUnixNano": $span.startTime,
+      "endTimeUnixNano": $span.endTime,
+      "status": {"code": if span.status == ssOk: "STATUS_CODE_OK" else: "STATUS_CODE_ERROR"},
+      "attributes": attrs,
+    })
+  let body = %*{
+    "resourceSpans": [{
+      "resource": {"attributes": [
+        {"key": "service.name", "value": {"stringValue": "baradadb"}}
+      ]},
+      "scopeSpans": [{
+        "scope": {"name": "baradadb-tracer", "version": "0.1.0"},
+        "spans": otlpSpans
+      }]
+    }]
+  }
+  try:
+    let client = newHttpClient()
+    client.headers["Content-Type"] = "application/json"
+    discard client.postContent(endpoint, body = $body)
+    client.close()
+    tracer.spans = @[]
+    return true
+  except:
+    return false

src/barabadb/query/ast.nim

@@ -20,6 +20,7 @@ type
     nkCommitTxn
     nkRollbackTxn
     nkExplainStmt
+    nkRecoverToTimestamp
     nkCreateView
     nkDropView
     nkCreateTrigger
@@ -233,6 +234,8 @@ type
     of nkExplainStmt:
       expStmt*: Node
       expAnalyze*: bool
+    of nkRecoverToTimestamp:
+      recoverTimestamp*: string
     of nkCreateView:
       cvName*: string
       cvQuery*: Node

src/barabadb/query/executor.nim

@@ -17,8 +17,10 @@ import ../core/types
 import ../protocol/wire
 import ../storage/lsm
 import ../storage/btree
+import ../storage/wal
 import ../core/mvcc
 import ../core/tracing
+import ../fts/engine as fts
 
 type
   IndexEntry* = ref object
@@ -57,6 +59,7 @@ type
     btrees*: Table[string, BTreeIndex[string, IndexEntry]]
     views*: Table[string, Node]  # view name -> SELECT AST
     cteTables*: Table[string, seq[Row]]  # CTE name -> rows
+    ftsIndexes*: Table[string, fts.InvertedIndex]  # table.col -> FTS index
     txnManager*: TxnManager
     pendingTxn*: Transaction
     onChange*: proc(ev: ChangeEvent) {.closure.}
@@ -133,6 +136,7 @@ proc newExecutionContext*(db: LSMTree): ExecutionContext =
                    btrees: initTable[string, BTreeIndex[string, IndexEntry]](),
                    views: initTable[string, Node](),
                    cteTables: initTable[string, seq[Row]](),
+                   ftsIndexes: initTable[string, fts.InvertedIndex](),
                    users: initTable[string, UserDef](),
                    policies: initTable[string, seq[PolicyDef]](),
                    currentUser: "", currentRole: "",
@@ -299,6 +303,7 @@ proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext =
   ExecutionContext(db: ctx.db, tables: ctx.tables,
                    btrees: ctx.btrees, views: ctx.views,
                    cteTables: initTable[string, seq[Row]](),
+                   ftsIndexes: ctx.ftsIndexes,
                    users: ctx.users, policies: ctx.policies,
                    txnManager: ctx.txnManager,
                    currentUser: ctx.currentUser, currentRole: ctx.currentRole,
@@ -2115,6 +2120,20 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
       return okResult(msg="ALTER TABLE " & stmt.altName & " executed")
     return errResult("Table '" & stmt.altName & "' does not exist")
 
+  of nkRecoverToTimestamp:
+    let walPath = ctx.db.dir & "/wal.log"
+    let entries = readEntries(walPath)
+    var applied = 0
+    for entry in entries:
+      if entry.kind == wekPut:
+        ctx.db.put(cast[string](entry.key), entry.value)
+        inc applied
+      elif entry.kind == wekDelete:
+        ctx.db.delete(cast[string](entry.key))
+        inc applied
+    ctx.restoreSchema()
+    return okResult(msg="RECOVERED " & $applied & " entries from WAL")
+
   of nkCreateView:
     ctx.views[stmt.cvName] = stmt.cvQuery
     let viewKey = "_schema:views:" & stmt.cvName

src/barabadb/query/lexer.nim

@@ -109,6 +109,8 @@ type
     tkPolicy
     tkEnable
     tkDisable
+    tkRecover
+    tkTimestamp
     tkFor
     tkUsing
     tkGrant
@@ -280,6 +282,8 @@ const keywords*: Table[string, TokenKind] = {
   "disable": tkDisable,
   "for": tkFor,
   "using": tkUsing,
+  "recover": tkRecover,
+  "timestamp": tkTimestamp,
   "grant": tkGrant,
   "revoke": tkRevoke,
   "count": tkCount,

src/barabadb/query/parser.nim

@@ -1181,6 +1181,13 @@ proc parseStatement*(p: var Parser): Node =
   of tkCommit: p.parseCommitTxn()
   of tkRollback: p.parseRollbackTxn()
   of tkExplain: p.parseExplain()
+  of tkRecover:
+    let tok = p.advance()
+    discard p.expect(tkTo)
+    discard p.expect(tkTimestamp)
+    let tsLit = p.expect(tkStringLit)
+    Node(kind: nkRecoverToTimestamp, recoverTimestamp: tsLit.value,
+         line: tok.line, col: tok.col)
   else:
     let tok = p.advance()
     Node(kind: nkNullLit, line: tok.line, col: tok.col)