RNT interface

Co-authored-by: Estevan Castilho <tevo@riajuu.net>

2 files changed, 55 insertions, 70 deletions

diff --git a/lib/drl/executor.ml b/lib/drl/executor.ml
index 41bdde5..20b94b9 100644
--- a/lib/drl/executor.ml
+++ b/lib/drl/executor.ml
@@ -1,75 +1,66 @@
-module Make (Storage : Management.Physical.S) = struct
-  module Alg = Algebra.Make (Storage)
-  module Ops = Manipulation.Make (Storage)
-
+module Make (NT : Nt.S) = struct
   type error =
     | ParseError of string
     | RelationNotFound of string
-    | AlgebraError of Algebra.error
+    | NtError of Nt.error
+    | UnsupportedOperator of string
+    | UnqualifiedName of string
 
   let sexp_of_error e =
     let open Sexplib.Sexp in
     match e with
     | ParseError s -> List [ Atom "parse-error"; Atom s ]
     | RelationNotFound s -> List [ Atom "relation-not-found"; Atom s ]
-    | AlgebraError (Algebra.StorageError s) ->
-        List [ Atom "storage-error"; Atom s ]
-    | AlgebraError (Algebra.GeneratorError s) ->
-        List [ Atom "generator-error"; Atom s ]
-
-  let wrap = Result.map_error (fun e -> AlgebraError e)
-  let ast_value_to_abstract = Ast.value_to_abstract
+    | NtError e -> List [ Atom "nt-error"; Atom (Nt.string_of_error e) ]
+    | UnsupportedOperator s -> List [ Atom "unsupported-operator"; Atom s ]
+    | UnqualifiedName s -> List [ Atom "unqualified-name"; Atom s ]
 
-  let select_semijoin storage source filter =
-    let common =
-      List.filter_map
-        (fun (n, _) ->
-          if List.exists (fun (m, _) -> m = n) filter#schema then
-            Some n
-          else None)
-        source#schema
-    in
-    let source_attrs = List.map fst source#schema in
-    Result.bind
-      (wrap (Alg.equijoin storage common source filter))
-      (fun joined -> wrap (Alg.project storage source_attrs joined))
+  let ( let* ) = Result.bind
 
-  let rec execute (storage : Storage.t) (db : Management.Multigroup.multigroup)
-      (q : Ast.query) : (Relation.ephemeral, error) Result.t =
-    let ( >>= ) = Result.bind in
+  (* 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, error) result =
     match q with
-    | Ast.Base name -> (
-        match Ops.get_relation db name with
-        | None -> Error (RelationNotFound name)
-        | Some rel -> Ok rel)
-    | Ast.Const pairs ->
-        Ok
-          (Alg.const_relation
-             (List.map (fun (k, v) -> (k, ast_value_to_abstract v)) pairs))
-    | Ast.Select (filter_q, source_q) ->
-        execute storage db filter_q >>= fun filter ->
-        execute storage db source_q >>= fun source ->
-        select_semijoin storage source filter
-    | Ast.Join (attrs, q1, q2) ->
-        execute storage db q1 >>= fun r1 ->
-        execute storage db q2 >>= fun r2 ->
-        wrap (Alg.equijoin storage attrs r1 r2)
-    | Ast.Project (attrs, q) ->
-        execute storage db q >>= fun rel -> wrap (Alg.project storage attrs rel)
-    | Ast.Rename (renames, q) ->
-        execute storage db q >>= fun rel ->
-        wrap (Alg.rename storage renames rel)
-    | Ast.Cartesian (q1, q2) ->
-        execute storage db q1 >>= fun r1 ->
-        execute storage db q2 >>= fun r2 -> wrap (Alg.equijoin storage [] r1 r2)
-    | Ast.Union (q1, q2) ->
-        execute storage db q1 >>= fun r1 ->
-        execute storage db q2 >>= fun r2 -> wrap (Alg.union storage r1 r2)
-    | Ast.Diff (q1, q2) ->
-        execute storage db q1 >>= fun r1 ->
-        execute storage db q2 >>= fun r2 -> wrap (Alg.diff storage r1 r2)
+    | Ast.Base name ->
+        (match Qualified_name.try_parse name with
+         | Error s -> Error (UnqualifiedName s)
+         | Ok fqn  -> Ok (Nt.Scan { path = resolve fqn; args = [] }))
+    | 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.Take (n, q) ->
-        execute storage db q >>= fun rel -> wrap (Alg.take storage n rel)
-end
+        let* p = compile resolve q in
+        Ok (Nt.Take { limit = n; source = p })
+    | Ast.Const _ ->
+        Error (UnsupportedOperator
+          "Const: literal relations not yet reachable by VM; use Base")
+    | _ ->
+        Error (UnsupportedOperator
+          "Select/Project/Rename/Union/Diff/Cartesian require VM operators not yet implemented")
 
-module Memory = Make (Management.Physical.Memory)
+  let page_limit = 16
+
+  let execute (ctx : Sublanguage_context.t) (q : Ast.query) :
+      (Sublanguage_types.result, error) result =
+    let* plan = compile ctx.resolve q in
+    let rel_name = "query_result" in
+    let* stream =
+      NT.execute_query plan ~rel_name
+      |> Result.map_error (fun e -> NtError e)
+    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
diff --git a/lib/drl/sublanguage.ml b/lib/drl/sublanguage.ml
index f47129f..ccc0222 100644
--- a/lib/drl/sublanguage.ml
+++ b/lib/drl/sublanguage.ml
@@ -1,8 +1,7 @@
-module Make (Storage : Management.Physical.S) = struct
-  module Exec = Executor.Make (Storage)
+module Make (NT : Nt.S) = struct
+  module Exec = Executor.Make (NT)
 
   type configuration = unit
-  type storage = Storage.t
   type ast = Ast.query
   type error = Exec.error
 
@@ -14,12 +13,7 @@ module Make (Storage : Management.Physical.S) = struct
     | Ok r -> Ok r
     | Error (Parser.ParseError s) -> Error (Exec.ParseError s)
 
-  let execute storage db ast =
-    match Exec.execute storage db ast with
-    | Ok rel -> Ok (Sublanguage_types.Query (rel :> Relation.relation))
-    | Error e -> Error e
+  let execute ctx ast = Exec.execute ctx ast
 
   let sexp_of_error = Exec.sexp_of_error
 end
-
-module Memory = Make (Management.Physical.Memory)