blob: 9e48021ef2458e5cac7b5cd1ef978299726ac754 (
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
|
module Make (NT : Nt.S) = struct
module Error = struct
open Condition
let parse_error error = condition "parse-error" error empty
let relation_not_found name = condition "relation-not-found" "Relation not found" ("name" |=| (of_string name))
let unsupported_operator msg = condition "unsupported-operator" msg empty
end
let ( let* ) = Result.bind
(* Compile a DRL AST query to a Tarski VM plan node tree.
Every base relation reference must be fully qualified ([<mg>:<rel>]);
[resolve] maps an FQN to its absolute RNT path. *)
let rec compile
(resolve : ?branch:string -> Qualified_name.t -> string)
(q : Ast.query) : (Nt.plan_node, Condition.t) result =
match q with
| Ast.Base name ->
let* fqn = Qualified_name.try_parse name in
Ok (Nt.Scan { path = resolve fqn; args = [] })
| Ast.Const _ ->
Error (Error.unsupported_operator
"Const: literal relations not yet reachable by VM; use Base")
| Ast.Select _ ->
Error (Error.unsupported_operator "Select unsupported")
| Ast.Join (on_attrs, q1, q2) ->
let* p1 = compile resolve q1 in
let* p2 = compile resolve q2 in
Ok (Nt.Join { left = p1; right = p2; on_attrs })
| Ast.Cartesian (q1, q2) ->
let* p1 = compile resolve q1 in
let* p2 = compile resolve q2 in
Ok (Nt.Join { left = p1; right = p2; on_attrs = [] })
| Ast.Project (attrs, query) ->
let* source = compile resolve query in
Ok (Nt.Project { attrs; source })
| Ast.Rename _ -> Error (Error.unsupported_operator "Rename unsupported")
| Ast.Union _ -> Error (Error.unsupported_operator "Union unsupported")
| Ast.Diff _ -> Error (Error.unsupported_operator "Diff unsupported")
| Ast.Take (n, q) ->
let* p = compile resolve q in
Ok (Nt.Take { limit = n; source = p })
let page_limit = 16
let execute (ctx : Sublanguage_context.t) (q : Ast.query) :
(Sublanguage_types.result, Condition.t) result =
let* plan = compile ctx.resolve q in
let rel_name = "query_result" in
let* stream = NT.execute_query plan ~rel_name
in
let rec drain acc count =
if count >= page_limit then (List.rev acc, true)
else
match NT.stream_next stream with
| Error _ -> (List.rev acc, false)
| Ok None -> (List.rev acc, false)
| Ok (Some t) -> drain (t :: acc) (count + 1)
in
let (rows, has_more) = drain [] 0 in
ignore (NT.stream_close stream);
Ok (Sublanguage_types.Cursor { cursor_id = "0"; rows; has_more })
end
|
