From 013cbf09e4ea9871311bf52b6873c43fc9be5441 Mon Sep 17 00:00:00 2001
From: Marcos Magueta <maguetamarcos@gmail.com>
Date: Tue, 24 Mar 2026 17:51:03 -0300
Subject: Session with stream control sublanguage (#55)

SCL adds cursor-based streaming to Sakura so clients can consume DRL query results incrementally in bounded batches rather than all at once. Three statements: Begin, Fetch, and Close. `open`, `advance`, and `release` cursors backed by a session registry that suspends the generator between calls. SCL is just an optional consumption layer above the existing algebra.
---
 lib/scl/executor.ml | 77 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 77 insertions(+)
 create mode 100644 lib/scl/executor.ml

(limited to 'lib/scl/executor.ml')

diff --git a/lib/scl/executor.ml b/lib/scl/executor.ml
new file mode 100644
index 0000000..6cb17b8
--- /dev/null
+++ b/lib/scl/executor.ml
@@ -0,0 +1,77 @@
+let default_batch = 50
+
+type cursor_batch = {
+  cursor_id : string;
+  rows : Tuple.materialized list;
+  has_more : bool;
+}
+
+type exec_result =
+  | Batch of cursor_batch
+  | Closed of Management.Database.t
+
+let sessions : Session.t option ref = ref None
+
+let set_sessions s = sessions := Some s
+
+module Make (Storage : Management.Physical.S) = struct
+  module DrlExec = Drl.Executor.Make(Storage)
+  module Alg = Algebra.Make(Storage)
+
+  type error =
+    | ParseError  of string
+    | QueryError  of DrlExec.error
+    | CursorError of string
+
+  let sexp_of_error e =
+    let open Sexplib.Sexp in
+    match e with
+    | ParseError s  -> List [Atom "parse-error";  Atom s]
+    | QueryError e  -> DrlExec.sexp_of_error e
+    | CursorError s -> List [Atom "cursor-error"; Atom s]
+
+  let get_sessions () =
+    match !sessions with
+    | Some s -> Ok s
+    | None   -> Error (CursorError "The session registry has not been initialised; ensure `set_sessions` is called at startup.")
+
+  let ( let* ) = Result.bind
+
+  let execute
+      (storage : Storage.t)
+      (db : Management.Database.t)
+      (stmt : Ast.statement)
+    : (exec_result, error) result =
+    match stmt with
+    | Ast.Begin { query; limit } ->
+      let* reg = get_sessions () in
+      let* rel =
+        DrlExec.execute storage db query
+        |> Result.map_error (fun e -> QueryError e)
+      in
+      let gen = Alg.to_generator storage rel in
+      let query_str = Drl.Parser.to_string query in
+      let cur = Session.register reg ~db ~query:query_str ~generator:gen in
+      let batch = Option.value ~default:default_batch limit in
+      let* (rows, has_more) =
+        Session.fetch reg ~id:cur.Session.id ~limit:batch
+        |> Result.map_error (fun s -> CursorError s)
+      in
+      Ok (Batch { cursor_id = cur.Session.id; rows; has_more })
+
+    | Ast.Fetch { cursor; limit } ->
+      let* reg = get_sessions () in
+      let batch = Option.value ~default:default_batch limit in
+      let* (rows, has_more) =
+        Session.fetch reg ~id:cursor ~limit:batch
+        |> Result.map_error (fun s -> CursorError s)
+      in
+      Ok (Batch { cursor_id = cursor; rows; has_more })
+
+    | Ast.Close { cursor } ->
+      let* reg = get_sessions () in
+      Session.close reg ~id:cursor;
+      Ok (Closed db)
+end
+
+module Memory = Make(Management.Physical.Memory)
-- 
cgit v1.2.3