FOL: First Order Logic Execution Engine

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

7 files changed, 366 insertions, 99 deletions

diff --git a/lib/backend/execution/FOL.ml b/lib/backend/execution/FOL.ml
index 1a5770e..302285d 100644
--- a/lib/backend/execution/FOL.ml
+++ b/lib/backend/execution/FOL.ml
@@ -1,69 +1,9 @@
-(** A plan is pure data ([Plan.t]) and execution state is never
-    reified: a compiled plan is a [Seq.t], whose forcing is the
-    Volcano eruption [Next]
-
-    The only effects sit at scan leaves, and those are idempotent page
-    reads against a pinned [Multigroup] snapshot. A stream node may be
-    forced twice and observe the same tuples. If a truly destructive
-    source ever appears, wrap its stream in [Seq.once]. *)
-
-module Tuple = struct
-  module AttributeMap = BatMap.String
-
-  (* TODO: The assumption here is that every value is stringified *)
-  (* TODO: The field [type'] currently is not enforced as all the
-     available types of RNT *)
-  type attribute = {value: string; type': string}
-  type t = attribute AttributeMap.t
-
-  let empty : t = AttributeMap.empty
-  let access (name : string) (t : t) : attribute option = AttributeMap.find_opt name t
-
-  let merge (left : t) (right : t) : t =
-    let merger _ (left : attribute option) (right : attribute option) : attribute option =
-      match left, right with
-      | Some a, Some b -> if a = b then Some a else None (* conflicting binding gets droped out *)
-      | (Some _ as a), None | None, (Some _ as a) -> a
-      | None, None -> None
-    in
-    AttributeMap.merge merger left right
-
-  let project (keep : BatSet.String.t) (t : t) : t =
-    AttributeMap.filter (fun name _ -> BatSet.String.mem name keep) t
-
-  let rename (mapping : string BatMap.String.t) (t : t) : t =
-    AttributeMap.fold
-      (fun name attr acc ->
-        let name = BatMap.String.find_default name name mapping in
-        AttributeMap.add name attr acc )
-      t AttributeMap.empty
-end
-
-(** Stub of the cursor layer (CursorManager + Merkle paging). [page]
-    yields the tuples of [relation] starting at [offset]; an empty
-    result marks exhaustion. [args] carries the bound values for
-    parameterized (ephemeral) relations and is ignored for stored
-    ones. *)
-module Source = struct
-  type t =
-    {page: relation:string -> args:string BatFingerTree.t -> offset:int -> Tuple.t BatFingerTree.t}
-
-  let unimplemented : t = {page= (fun ~relation:_ ~args:_ ~offset:_ -> failwith "NOT IMPLEMENTED")}
-end
-
-(** The stateless operator tree representing one constructor per
-    relational operator, carrying only its description without any
-    associated cursors, counters and buffers. *)
 module Plan = struct
-  (** One segment of a parameterized relation path in a SCAN. [Var]
-      segments are resolved from the enclosing JOIN's outer tuple at
-      execution time; [Const] segments are ground literals fixed at
-      the plan construction. *)
-  type path_arg = Var of string | Const of string
+  type path_arg = Var of string | Const of Concepts.Value.value
 
   type t =
-    | Scan of {relation: string; args: path_arg BatFingerTree.t}
-    | Join of {left: t; right: t; attrs: BatSet.String.t}
+    | Scan of {path: Concepts.Path.t; args: path_arg BatFingerTree.t}
+    | Natural of {left: t; right: t}
     | Take of {limit: int; from: t}
     | Project of {attrs: BatSet.String.t; from: t}
     | Materialize of t
@@ -71,34 +11,95 @@ module Plan = struct
     | Union of t BatFingerTree.t
 end
 
-(** Compiles a [Plan.t] into a demand driven tuple stream. *)
-module FOL = struct
-  type stream = Tuple.t BatSeq.t
+type stream = (Concepts.Tuple.t, Concepts.Condition.condition) result BatSeq.t
+
+let singleton (x : ('a, 'e) result) : ('a, 'e) result BatSeq.t =
+ fun () -> BatSeq.Cons (x, BatSeq.empty)
 
-  (** Resolves a scan arguments template against the enclosing
-      bindings. *)
-  (* TODO: A [Var] missing from the bindings resolves to [""] to
-     mirror the C++ engine. This is a poor representation in the type
-     system. *)
-  let resolve (args : Plan.path_arg BatFingerTree.t) (bindings : Tuple.t) : string BatFingerTree.t =
-    BatFingerTree.map
-      (function
-        | Plan.Const value -> value
-        | Plan.Var name -> (match Tuple.access name bindings with Some a -> a.value | None -> "" ))
-      args
+module Error = struct
+  open Concepts.Condition
 
-  (** Pages through [relation] one fetch at a time. The offset lives
-      in the closure of each stream node, so re-forcing any node
-      re-reads the same page. This is the functional equivalent of the
-      cursor's [fetch_offset]. Only the current page should be
-      live. *)
-  let scan (src : Source.t) ~(relation : string) ~(args : string BatFingerTree.t) : stream =
-    let rec from offset () =
-      let page = src.page ~relation ~args ~offset in
-      if BatFingerTree.is_empty page then BatSeq.Nil
-      else
-        let rest = from (offset + BatFingerTree.size page) in
-        (BatFingerTree.fold_right (fun rest tuple () -> BatSeq.Cons (tuple, rest)) rest page) ()
+  let unbound_variable name =
+    condition "unbound-variable"
+      (Printf.sprintf "path variable %S is not bound in the enclosing tuple" name)
+      ("variable" |=| Concepts.Value.String name)
+end
+
+let resolve (args : Plan.path_arg BatFingerTree.t) (bindings : Concepts.Tuple.t) :
+    (Concepts.Value.value BatFingerTree.t, Concepts.Condition.condition) result =
+  let open Utilities.Result in
+  BatFingerTree.fold_left
+    (fun acc arg ->
+      let* acc = acc in
+      let* value =
+        match arg with
+        | Plan.Const value -> Ok value
+        | Plan.Var name ->
+            Option.to_result ~none:(Error.unbound_variable name)
+              (Concepts.Tuple.access name bindings)
+      in
+      Ok (BatFingerTree.snoc acc value) )
+    (Ok BatFingerTree.empty) args
+
+module Make (Handler : Managers.Handle.HANDLER) = struct
+  let scan handler ~path ~args : stream =
+    let opened =
+      let open Utilities.Result in
+      let* handle = Handler.open_ handler ~path in
+      Handler.open_cursor handle ~args
     in
-    from 0
+    match opened with
+    | Error condition -> singleton (Error condition)
+    | Ok cursor ->
+        let rec pull () =
+          match Handler.next cursor with
+          | Error condition -> BatSeq.Cons (Error condition, BatSeq.empty)
+          | Ok None -> BatSeq.Nil
+          | Ok (Some tuple) -> BatSeq.Cons (Ok tuple, pull)
+        in
+        pull
+
+  let rec compile handler : Plan.t -> Concepts.Tuple.t -> stream = function
+    | Scan {path; args} -> (
+        fun bindings ->
+          match resolve args bindings with
+          | Ok args -> scan handler ~path ~args
+          | Error condition -> singleton (Error condition) )
+    | Project {attrs; from} ->
+        let from = compile handler from in
+        fun bindings -> BatSeq.map (Result.map (Concepts.Tuple.project attrs)) (from bindings)
+    | Rename {attrs; from} ->
+        let from = compile handler from in
+        fun bindings -> BatSeq.map (Result.map (Concepts.Tuple.rename attrs)) (from bindings)
+    | Take {limit; from} ->
+        let from = compile handler from in
+        fun bindings -> BatSeq.take limit (from bindings)
+    | Union plans ->
+        let compiled = BatFingerTree.map (compile handler) plans in
+        fun bindings ->
+          BatFingerTree.fold_right
+            (fun rest c -> BatSeq.append (c bindings) rest)
+            BatSeq.empty compiled
+    | Natural {left; right} ->
+        let left = compile handler left and right = compile handler right in
+        let matches l r =
+          BatMap.String.for_all
+            (fun name value ->
+              match Concepts.Tuple.access name r with Some value' -> value = value' | None -> true )
+            l
+        in
+        fun bindings ->
+          left bindings
+          |> BatSeq.concat_map (function
+            | Error _ as error -> singleton error
+            | Ok l ->
+                right l
+                |> BatSeq.filter (function Ok r -> matches l r | Error _ -> true)
+                |> BatSeq.map (Result.map (Concepts.Tuple.merge l)) )
+    | Materialize from ->
+        let cached = lazy (BatSeq.memoize (compile handler from Concepts.Tuple.empty)) in
+        fun _ -> Lazy.force cached
+
+  (* TODO: This should produce ephemeral relations with support of the sublanguage to manage it. *)
+  let run handler plan : stream = compile handler plan Concepts.Tuple.empty
 end
diff --git a/lib/backend/managers/Cursor.ml b/lib/backend/managers/Cursor.ml
new file mode 100644
index 0000000..3295cae
--- /dev/null
+++ b/lib/backend/managers/Cursor.ml
@@ -0,0 +1,18 @@
+type descriptor =
+  | Stored of {merkle_root: string}
+  | Ephemeral of {merkle_root: string; dependencies: string BatFingerTree.t}
+
+module Make (Store : Abstract.Storage.STORAGE) = struct
+  type cursor = unit
+
+  let open_ (_txn : Store.transaction) (descriptor : descriptor)
+      ~(args : Concepts.Value.value BatFingerTree.t) : (cursor, Concepts.Condition.condition) result
+      =
+    ignore (descriptor, args);
+    failwith "NOT IMPLEMENTED: open a cursor by paging the relation's Merkle tree over Store.get"
+
+  let next (_cursor : cursor) : (Concepts.Tuple.t option, Concepts.Condition.condition) result =
+    failwith "NOT IMPLEMENTED: page the next tuple over Store.get"
+
+  let close (_cursor : cursor) : unit = ()
+end
diff --git a/lib/backend/managers/Cursor.mli b/lib/backend/managers/Cursor.mli
new file mode 100644
index 0000000..71fb15d
--- /dev/null
+++ b/lib/backend/managers/Cursor.mli
@@ -0,0 +1,16 @@
+type descriptor =
+  | Stored of {merkle_root: string}
+  | Ephemeral of {merkle_root: string; dependencies: string BatFingerTree.t}
+
+module Make (Store : Abstract.Storage.STORAGE) : sig
+  type cursor
+
+  val open_ :
+    Store.transaction ->
+    descriptor ->
+    args:Concepts.Value.value BatFingerTree.t ->
+    (cursor, Concepts.Condition.condition) result
+
+  val next : cursor -> (Concepts.Tuple.t option, Concepts.Condition.condition) result
+  val close : cursor -> unit
+end
diff --git a/lib/backend/managers/Handle.ml b/lib/backend/managers/Handle.ml
new file mode 100644
index 0000000..ed2f220
--- /dev/null
+++ b/lib/backend/managers/Handle.ml
@@ -0,0 +1,51 @@
+module type HANDLER = sig
+  type t
+  type handle
+  type cursor
+
+  val open_ : t -> path:Concepts.Path.t -> (handle, Concepts.Condition.condition) result
+
+  val open_cursor :
+    handle ->
+    args:Concepts.Value.value BatFingerTree.t ->
+    (cursor, Concepts.Condition.condition) result
+
+  val next : cursor -> (Concepts.Tuple.t option, Concepts.Condition.condition) result
+  val close : cursor -> unit
+end
+
+let not_a_relation () : Concepts.Condition.condition =
+  Concepts.Condition.condition "not-a-relation"
+    "handle does not refer to a stored or ephemeral relation" Concepts.Condition.empty
+
+module Make (Store : Abstract.Storage.STORAGE) = struct
+  module Cur = Cursor.Make (Store)
+
+  type t = {objects: Object.rnt_object_tree; txn: Store.transaction}
+  type handle = {object_: Object.registry; txn: Store.transaction}
+  type cursor = Cur.cursor
+
+  let create objects txn = {objects; txn}
+
+  let open_ t ~path =
+    let object_ = Object.find path t.objects in
+    Ok {object_; txn= t.txn}
+
+  let relation_descriptor (object_ : Object.registry) :
+      (Cursor.descriptor, Concepts.Condition.condition) result =
+    let {Object.entry; _} = object_ in
+    match entry.Object.kind with
+    | Object.Relation {merkle_root} -> Ok (Cursor.Stored {merkle_root})
+    | Object.EphemeralRelation {merkle_root; dependencies} ->
+        Ok (Cursor.Ephemeral {merkle_root; dependencies})
+    | _ -> Error (not_a_relation ())
+
+  let open_cursor handle ~args =
+    let open Utilities.Result in
+    let* descriptor = relation_descriptor handle.object_ in
+    Cur.open_ handle.txn descriptor ~args
+
+  let next = Cur.next
+  let close = Cur.close
+  let object_ (handle : handle) = handle.object_
+end
diff --git a/lib/backend/managers/Handle.mli b/lib/backend/managers/Handle.mli
new file mode 100644
index 0000000..5fc82b8
--- /dev/null
+++ b/lib/backend/managers/Handle.mli
@@ -0,0 +1,22 @@
+module type HANDLER = sig
+  type t
+  type handle
+  type cursor
+
+  val open_ : t -> path:Concepts.Path.t -> (handle, Concepts.Condition.condition) result
+
+  val open_cursor :
+    handle ->
+    args:Concepts.Value.value BatFingerTree.t ->
+    (cursor, Concepts.Condition.condition) result
+
+  val next : cursor -> (Concepts.Tuple.t option, Concepts.Condition.condition) result
+  val close : cursor -> unit
+end
+
+module Make (Store : Abstract.Storage.STORAGE) : sig
+  include HANDLER
+
+  val create : Object.rnt_object_tree -> Store.transaction -> t
+  val object_ : handle -> Object.registry
+end
diff --git a/lib/backend/storage/LMDB.ml b/lib/backend/storage/LMDB.ml
index 3478da2..f4fcea1 100644
--- a/lib/backend/storage/LMDB.ml
+++ b/lib/backend/storage/LMDB.ml
@@ -1,11 +1,170 @@
-type connection = unit
-type transaction = unit
+module C = struct
+  open Ctypes
+  open PosixTypes
+  open Foreign
 
-let connect (_: Concepts.Configuration.configuration) = failwith "TODO"
+  let mdb_result_of_int = function 0 -> Ok () | x -> Error x
+  let int_of_mdb_result = function Ok _ -> 0 | Error x -> x
+  let mdb_mode_t = mode_t
+  let mdb_result = view ~read:mdb_result_of_int ~write:int_of_mdb_result int
 
-let start (_: connection) = failwith "TODO"
-let commit (_: transaction) = failwith "TODO"
-let abort (_: transaction) = failwith "TODO"
+  type mdb_env
 
-let get (_: transaction) (_: Concepts.Hash.hash) = failwith "TODO"
-let put (_: transaction) (_: Concepts.Hash.hash) (_: Concepts.Representation.blob) = failwith "TODO"
+  let mdb_env : mdb_env structure typ = structure "mdb_env"
+
+  type mdb_dbi = Unsigned.uint
+
+  let mdb_dbi : mdb_dbi typ = uint
+
+  type mdb_txn
+
+  let mdb_txn : mdb_txn structure typ = structure "mdb_txn"
+
+  type mdb_val
+
+  let mdb_val : mdb_val structure typ = structure "mdb_val"
+  let mv_size = field mdb_val "mv_size" size_t
+  let mv_data = field mdb_val "mv_data" (ptr void)
+  let () = seal mdb_val
+
+  let with_output_pointer output_type default body =
+    let open Utilities.Result in
+    let output_ptr = allocate output_type default in
+    let* () = body output_ptr in
+    Ok !@output_ptr
+
+  let mdb_env_create = foreign "mdb_env_create" (ptr (ptr mdb_env) @-> returning mdb_result)
+
+  let mdb_env_create' () =
+    with_output_pointer (ptr mdb_env) (from_voidp mdb_env null) mdb_env_create
+
+  let mdb_env_open =
+    foreign "mdb_env_open" (ptr mdb_env @-> string @-> uint @-> mdb_mode_t @-> returning mdb_result)
+
+  let mdb_env_close = foreign "mdb_env_close" (ptr mdb_env @-> returning void)
+
+  let mdb_dbi_open =
+    foreign "mdb_dbi_open"
+      (ptr mdb_txn @-> ptr char @-> uint @-> ptr mdb_dbi @-> returning mdb_result)
+
+  let mdb_dbi_open' txn flags =
+    with_output_pointer mdb_dbi (Unsigned.UInt.of_int 0)
+      (mdb_dbi_open txn (from_voidp char null) flags)
+
+  let mdb_txn_begin =
+    foreign "mdb_txn_begin"
+      (ptr mdb_env @-> ptr mdb_txn @-> uint @-> ptr (ptr mdb_txn) @-> returning mdb_result)
+
+  let mdb_txn_begin' env parent flags = with_output_pointer
+                                          (ptr mdb_txn)
+                                          (from_voidp mdb_txn null)
+                                          (mdb_txn_begin env parent flags)
+  [@@ocamlformat "disable"]
+
+  let mdb_txn_commit = foreign "mdb_txn_commit" (ptr mdb_txn @-> returning mdb_result)
+  let mdb_txn_abort = foreign "mdb_txn_abort" (ptr mdb_txn @-> returning mdb_result)
+
+  (* TODO: make it so that we do not need to copy bytes to a separate array *)
+  let carray_of_bytes (b : bytes) =
+    let buffer = CArray.make char (Bytes.length b) in
+    Bytes.iteri (CArray.set buffer) b;
+    buffer
+
+  let bytes_of_carray arr =
+    let buffer = Bytes.make (CArray.length arr) (Char.chr 0) in
+    CArray.iteri (Bytes.set buffer) arr;
+    buffer
+
+  let mdb_val_ptr_of_bytes (b : bytes) =
+    let buf = carray_of_bytes b in
+    let s = make mdb_val in
+    setf s mv_size (Unsigned.Size_t.of_int (Bytes.length b));
+    setf s mv_data (to_voidp (CArray.start buf));
+    addr s
+
+  let bytes_of_mdb_val (s : mdb_val structure) =
+    let buf =
+      CArray.from_ptr (from_voidp char (getf s mv_data)) (Unsigned.Size_t.to_int (getf s mv_size))
+    in
+    bytes_of_carray buf
+
+  let mdb_get =
+    foreign "mdb_get"
+      (ptr mdb_txn @-> mdb_dbi @-> ptr mdb_val @-> ptr mdb_val @-> returning mdb_result)
+
+  let mdb_get' txn dbi key =
+    with_output_pointer mdb_val (make mdb_val) (mdb_get txn dbi (mdb_val_ptr_of_bytes key))
+    |> Result.map bytes_of_mdb_val
+
+  let mdb_put =
+    foreign "mdb_put"
+      (ptr mdb_txn @-> mdb_dbi @-> ptr mdb_val @-> ptr mdb_val @-> uint @-> returning mdb_result)
+
+  let mdb_put' txn dbi key data flags =
+    mdb_put txn dbi (mdb_val_ptr_of_bytes key) (mdb_val_ptr_of_bytes data) flags
+
+  let mdb_strerror = foreign "mdb_strerror" (int @-> returning string)
+
+  type mdb_env_ptr = mdb_env structure ptr
+  type mdb_txn_ptr = mdb_txn structure ptr
+
+  let null_txn = from_voidp mdb_txn null
+
+  module Errors = struct
+    let mdb_notfound = -30798
+  end
+end
+
+module Error = struct
+  open Concepts.Condition
+
+  let lmdb_error code =
+    condition "lmdb-error" (C.mdb_strerror code) ("code" |=| Concepts.Value.Integer code)
+end
+
+type connection = {env: C.mdb_env_ptr; dbi: C.mdb_dbi}
+type transaction = {tx: C.mdb_txn_ptr; dbi: C.mdb_dbi}
+
+let connect (_ : Concepts.Configuration.configuration) =
+  begin
+    let open Utilities.Result in
+    let* env = C.mdb_env_create' () in
+    (* TODO: make the path, POSIX mode a config parameter *)
+    match C.mdb_env_open env "/tmp/sakura.db" Unsigned.UInt.zero (PosixTypes.Mode.of_int 420) with
+    | Error x -> C.mdb_env_close env; Error x
+    | Ok () ->
+        let* tx = C.mdb_txn_begin' env C.null_txn Unsigned.UInt.zero in
+        let* dbi = C.mdb_dbi_open' tx Unsigned.UInt.zero in
+        let* () = C.mdb_txn_commit tx in
+        Ok {env; dbi}
+  end
+  |> Result.map_error Error.lmdb_error
+
+let start ({env; dbi} : connection) =
+  C.mdb_txn_begin' env C.null_txn Unsigned.UInt.zero
+  |> Result.map (fun tx -> {tx; dbi})
+  |> Result.map_error Error.lmdb_error
+
+let commit ({tx; _} : transaction) = C.mdb_txn_commit tx |> Result.map_error Error.lmdb_error
+let abort ({tx; _} : transaction) = C.mdb_txn_abort tx |> Result.map_error Error.lmdb_error
+
+(*
+ * FIXME: `blob_of_bytes` and `bytes_of_blob` should not exist.
+ * Rather, the FFI should know how to serialize a blob directly. This
+ * makes it so that we do not depend on the OCaml `bytes` type and
+ * opens the path for avoiding a copy on the FFI boundary later on
+ *)
+
+let get ({tx; dbi} : transaction) (h : Concepts.Hash.hash) =
+  begin match C.mdb_get' tx dbi (Concepts.Hash.bytes_of_hash h) with
+  | Ok x -> Ok (Some (Concepts.Representation.blob_of_bytes x))
+  | Error e when e = C.Errors.mdb_notfound -> Ok None
+  | Error e -> Error e
+  end
+  |> Result.map_error Error.lmdb_error
+
+let put ({tx; dbi} : transaction) (h : Concepts.Hash.hash) (b : Concepts.Representation.blob) =
+  C.mdb_put' tx dbi (Concepts.Hash.bytes_of_hash h)
+    (Concepts.Representation.bytes_of_blob b)
+    Unsigned.UInt.zero
+  |> Result.map_error Error.lmdb_error
diff --git a/lib/backend/storage/LMDB.mli b/lib/backend/storage/LMDB.mli
index dde4b6e..b4bab77 100644
--- a/lib/backend/storage/LMDB.mli
+++ b/lib/backend/storage/LMDB.mli
@@ -1 +1 @@
-include Abstract.Storage
+include Abstract.Storage.STORAGE