aboutsummaryrefslogtreecommitdiffstats
path: root/lib
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--lib/scl/ast.ml7
-rw-r--r--lib/scl/executor.ml77
-rw-r--r--lib/scl/parser.ml19
-rw-r--r--lib/scl/sublanguage.ml26
-rw-r--r--lib/session.ml82
-rw-r--r--lib/sublanguage.ml7
6 files changed, 218 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)
diff --git a/lib/session.ml b/lib/session.ml
new file mode 100644
index 0000000..272a419
--- /dev/null
+++ b/lib/session.ml
@@ -0,0 +1,82 @@
+(** Cursor registry for streaming tuple consumption.
+
+ Each cursor wraps a [Generator.t] produced by a DRL query and
+ tracks the current position so that successive [fetch] calls
+ return the next batch of rows without replaying the query. *)
+
+type cursor = {
+ id : string;
+ mutable generator : Generator.t;
+ mutable position : int;
+ db_snapshot : Management.Database.t;
+ query_sexp : string;
+ created_at : float;
+}
+
+type t = {
+ tbl : (string, cursor) Hashtbl.t;
+ mutable counter : int;
+}
+
+let create () : t =
+ { tbl = Hashtbl.create 16; counter = 0 }
+
+let register (reg : t) ~(db : Management.Database.t) ~(query : string)
+ ~(generator : Generator.t) : cursor =
+ let raw = string_of_int reg.counter ^ query ^ db.Management.Database.hash in
+ let id = Conventions.Hash.hash_text raw in
+ reg.counter <- reg.counter + 1;
+ let cur = {
+ id;
+ generator;
+ position = 0;
+ db_snapshot = db;
+ query_sexp = query;
+ created_at = Unix.gettimeofday ();
+ } in
+ Hashtbl.replace reg.tbl id cur;
+ cur
+
+let fetch (reg : t) ~(id : string) ~(limit : int)
+ : (Tuple.materialized list * bool, string) result =
+ match Hashtbl.find_opt reg.tbl id with
+ | None -> Error ("The cursor with identifier `" ^ id ^ "` was not found in the session registry.")
+ | Some cur ->
+ let rec go gen pos acc count =
+ if count >= limit then (List.rev acc, gen, pos, true)
+ else
+ match gen (Some pos) with
+ | Generator.Done -> (List.rev acc, gen, pos, false)
+ (* Generator errors are treated as exhaustion: the partial batch is
+ returned with has_more = false and the cursor is auto-removed.
+ The caller receives no indication that the stream terminated
+ abnormally. When fetch gains a richer return type this arm
+ should surface the error message instead. *)
+ | Generator.Error _ -> (List.rev acc, gen, pos, false)
+ | Generator.Value (t, next) ->
+ begin match t with
+ | Tuple.Materialized m ->
+ go next (pos + 1) (m :: acc) (count + 1)
+ | Tuple.NonMaterialized _ ->
+ go next (pos + 1) acc count
+ end
+ in
+ let (rows, gen', pos', has_more) =
+ go cur.generator cur.position [] 0
+ in
+ cur.generator <- gen';
+ cur.position <- pos';
+ if not has_more then Hashtbl.remove reg.tbl id;
+ Ok (rows, has_more)
+
+let close (reg : t) ~(id : string) : unit =
+ Hashtbl.remove reg.tbl id
+
+let gc (reg : t) ~(max_age : float) : unit =
+ let now = Unix.gettimeofday () in
+ let to_remove =
+ Hashtbl.fold (fun k cur acc ->
+ if now -. cur.created_at >= max_age then k :: acc else acc)
+ reg.tbl []
+ in
+ List.iter (Hashtbl.remove reg.tbl) to_remove
diff --git a/lib/sublanguage.ml b/lib/sublanguage.ml
index 3df91f1..d92358e 100644
--- a/lib/sublanguage.ml
+++ b/lib/sublanguage.ml
@@ -14,9 +14,16 @@
(Turing-complete; breaks the termination guarantee of DRL)
- ACL -> a functor restriction: narrows the admissible natural
transformations based on authorization context *)
+type cursor_result = {
+ cursor_id : string;
+ rows : Tuple.materialized list;
+ has_more : bool;
+}
+
type result =
| Query of Relation.t
| Transition of Management.Database.t * string
+ | Cursor of cursor_result
module type S = sig
type storage
generated by cgit v1.2.3 (git 2.25.1) at 2026-11-04 12:05:01 +0000