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.

4 files changed, 129 insertions, 0 deletions

diff --git a/lib/scl/ast.ml b/lib/scl/ast.ml
new file mode 100644
index 0000000..043f544
--- /dev/null
+++ b/lib/scl/ast.ml
@@ -0,0 +1,7 @@
+open Sexplib.Std
+
+type statement =
+  | Begin of { query : Drl.Ast.query; limit : int option }
+  | Fetch of { cursor : string; limit : int option }
+  | Close of { cursor : string }
+[@@deriving sexp]
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)
diff --git a/lib/scl/parser.ml b/lib/scl/parser.ml
new file mode 100644
index 0000000..3e9c4ed
--- /dev/null
+++ b/lib/scl/parser.ml
@@ -0,0 +1,19 @@
+type error = ParseError of string
+
+let validate = function
+  | (Ast.Begin { limit = Some limit; _ } | Ast.Fetch { limit = Some limit; _ })
+    when limit <= 0 ->
+      Error (ParseError "The provision of `limit` is expected to be positive.")
+  | stmt -> Ok stmt
+
+let of_sexp sexp =
+  match Ast.statement_of_sexp sexp with
+  | exception exn -> Error (ParseError (Printexc.to_string exn))
+  | stmt -> validate stmt
+
+let of_string s =
+  match Sexplib.Sexp.of_string s |> Ast.statement_of_sexp with
+  | exception exn -> Error (ParseError (Printexc.to_string exn))
+  | stmt -> validate stmt
+
+let to_string stmt = Ast.sexp_of_statement stmt |> Sexplib.Sexp.to_string_hum
diff --git a/lib/scl/sublanguage.ml b/lib/scl/sublanguage.ml
new file mode 100644
index 0000000..7098ced
--- /dev/null
+++ b/lib/scl/sublanguage.ml
@@ -0,0 +1,26 @@
+module Make (Storage : Management.Physical.S) = struct
+  module Exec = Executor.Make(Storage)
+
+  type storage = Storage.t
+  type ast     = Ast.statement
+  type error   = Exec.error
+
+  let name = "scl"
+
+  let parse_sexp sexp =
+    match Parser.of_sexp sexp with
+    | Ok r -> Ok r
+    | Error (Parser.ParseError s) -> Error (Exec.ParseError s)
+
+  let execute storage db ast =
+    Exec.execute storage db ast
+    |> Result.map (function
+      | Executor.Batch { cursor_id; rows; has_more } ->
+        Sublanguage.Cursor { cursor_id; rows; has_more }
+      | Executor.Closed db ->
+        Sublanguage.Transition (db, "cursor closed"))
+
+  let sexp_of_error = Exec.sexp_of_error
+end
+
+module Memory = Make(Management.Physical.Memory)