path: root/lib/scl/executor.ml

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)