blob: eb1f4b6d121415fc2345218afcc5736f8d461b04 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
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
|
