blob: fffea19dd7e87e5d57082c23fe1e76382c190183 (
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
|
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.
A qualified base reference ([<mg>:<rel>]) resolves through [ctx.resolve]
to its stored-relation path; an unqualified name resolves to the
session-scoped ephemeral binding created by DML [Define]. *)
let rec compile
(ctx : Sublanguage_context.t)
(q : Ast.query) : (Nt.plan_node, Condition.t) result =
match q with
| Ast.Base name ->
(match Qualified_name.try_parse name with
| Error _ ->
Ok (Nt.Scan
{ path = Sublanguage_context.ephemeral_path ctx name;
args = [] })
| Ok fqn -> Ok (Nt.Scan { path = ctx.resolve fqn; args = [] }))
| Ast.Join (on_attrs, q1, q2) ->
let* p1 = compile ctx q1 in
let* p2 = compile ctx q2 in
Ok (Nt.Join { left = p1; right = p2; on_attrs })
| Ast.Take (n, q) ->
let* p = compile ctx q in
Ok (Nt.Take { limit = n; source = p })
| Ast.Project (attrs, q) ->
let* p = compile ctx q in
Ok (Nt.Project { attrs; source = p })
| Ast.Const _ ->
Error (Error.unsupported_operator
"Const: literal relations not yet reachable by VM; use Base")
| _ ->
Error (Error.unsupported_operator
"Select/Rename/Union/Diff/Cartesian require VM operators not yet implemented")
let page_limit = 16
let execute (ctx : Sublanguage_context.t) (q : Ast.query) :
(Sublanguage_types.result, Condition.t) result =
let* plan = compile ctx 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
|
