implement join and fix serialization

7 files changed, 271 insertions, 43 deletions

diff --git a/client.el b/client.el
index 8656306..91f833f 100644
--- a/client.el
+++ b/client.el
@@ -8,7 +8,7 @@
     (message "Received: %s" response)    
     (delete-process process)))
 
-(setq relational-engine-address "/tmp/relational_engine.socket")
+(setq relational-engine-address "localhost")
 (setq relational-engine-port 7524)
 
 (defun relational-client (content)
@@ -18,7 +18,13 @@
     (set-process-filter connection 'relational-engine-handler)
     (process-send-string connection content)))
 
-(setq msg "Hello Relational Engine! :)")
+(setq msg "(SequentialRead (relation_name address))")
 
 (relational-client msg)
-(print *mailbox*)
+
+(with-current-buffer "xml"
+  (erase-buffer)
+  (goto-char (point-min))
+  (insert *mailbox*)
+  ;; (xml-format-buffer)
+  )
diff --git a/lib/disk.ml b/lib/disk.ml
index 6996b2d..fbb6219 100644
--- a/lib/disk.ml
+++ b/lib/disk.ml
@@ -130,6 +130,7 @@ module Executor = struct
   end
 
   open Sexplib0.Sexp_conv
+  open Protocol_conv_xml
 
   type relational_type =
     | Text
@@ -139,7 +140,7 @@ module Executor = struct
     (* | DiscriminatedUnion of (string * relational_type option) list * string *)
       (* [("MemberA", None); ("MemberB", Some Integer32)], "NameOfDU" *)
     | Relation of string
-  [@@deriving show, sexp]
+  [@@deriving show, sexp, protocol ~driver:(module Xml_light)]
 
   type relational_literal =
     | LText of string
@@ -147,7 +148,7 @@ module Executor = struct
     | LInteger64 of int64
     | LBoolean of bool
     | LRelation of int64*string
-  [@@deriving show]
+  [@@deriving show, sexp, protocol ~driver:(module Xml_light)]
 
   type schema = (string * relational_type) list StringMap.t
   (* entity * type * attribute_name *)
@@ -446,12 +447,15 @@ module Command = struct
   }
 
   open Sexplib0.Sexp_conv
+  open Protocol_conv_xml
 
   type t =
     | SequentialRead of { relation_name : string }
     | SpecifyRelation of { relation_name: string;
                            attributes: (string*Executor.relational_type) list }
-  [@@deriving sexp]
+    | NaturalJoin of {left_relation: string;
+                      right_relation: string}
+  [@@deriving sexp, protocol ~driver:(module Xml_light)]
   
   let command_encoding =
     conv
@@ -474,7 +478,7 @@ module Command = struct
     | ComputedHash of string
     | Read of (int64 * (string*Executor.relational_literal) list) list
     | Nothing
-  [@@deriving show]
+  [@@deriving show, sexp, protocol ~driver:(module Xml_light)]
 
   (** Writes the commit and location to the disk, but cleans it before.
       Needless to say this is absolutely just a dirty trick to test persistency of the structures.
@@ -536,6 +540,8 @@ module Command = struct
   
   let perform (commit : Executor.commit) (locations : Executor.locations) (command : t) =
     match command with
+    | NaturalJoin {left_relation = _; right_relation = _} ->
+       Ok ((commit, locations), Nothing)
     | SpecifyRelation {relation_name; attributes} ->
        (* If it already exists under the relation name key, it will be replaced *)
        Ok (({ commit with
diff --git a/lib/dune b/lib/dune
index a84e9fa..8a0ae55 100644
--- a/lib/dune
+++ b/lib/dune
@@ -7,7 +7,8 @@
   lwt
   lwt.unix
   lwt-exit
-  ppx_protocol_conv_xml_light)
+  ppx_protocol_conv_xml_light
+  sexplib)
  (preprocess
   (pps ppx_deriving.show ppx_sexp_conv ppx_protocol_conv))
  ;; (c_library_flags -lssl -lcrypto -L %{env:LIBRESSL_LIB_PATH=})
diff --git a/lib/planner.ml b/lib/planner.ml
index 0470092..db48bca 100644
--- a/lib/planner.ml
+++ b/lib/planner.ml
@@ -8,8 +8,8 @@ end
 module Scan = struct
   open Disk
 
-  let execute commit locations (entity : string) =
-    Executor.read commit locations ~filename:("schema~" ^ entity) |> function
+  let execute (commit: Executor.commit) locations (entity : string) =
+    (*Executor.read commit locations ~filename:("schema~" ^ entity) |> function
     | Some binary -> (
         let relation =
           Result.map_error (fun _ ->
@@ -28,9 +28,41 @@ module Scan = struct
             Executor.read commit locations ~filename:("schema~" ^ entity)
         | Error err -> failwith err)
     | None -> failwith ""
+    *)
+    let entities : (string * Executor.relational_type * string) list Executor.IntMap.t =
+      Executor.StringMap.find_opt entity commit.references
+      |> function
+        | Some entities -> entities
+        | None -> Executor.IntMap.empty
+    in
+
+    let content: (int64 * (string*Executor.relational_literal) list) list =
+      Executor.IntMap.fold
+        (fun key elems acc ->
+          (key,
+           List.map
+             (fun (location, type', attribute_name) ->
+               (attribute_name, Executor.read_location ~hash:location locations type'))
+             elems)
+          :: acc)
+        entities []
+    in content
+    
 end
 
 module Join = struct
+  open Disk
+  module NestedLoop = struct
+    let execute
+          (left: (int64 * (string*Executor.relational_literal) list) list)
+          (right: (int64 * (string*Executor.relational_literal) list) list)
+          (join_clause: ((int64 * (string*Executor.relational_literal) list) *
+                         (int64 * (string*Executor.relational_literal) list)) -> bool) =
+      let result = ref [] in
+      List.iter (fun left_tuple -> List.iter (fun right_tuple -> if join_clause(left_tuple, right_tuple)
+                                                                 then result := (left_tuple,right_tuple)::(!result)) right) left;
+      !result
+  end
   (* open Disk
      let execute (left_attributes: bytes list) (right_attributes: bytes list) locations =
        (* Redundant operations that are already in the locations, but this is easier to implement now *)
@@ -43,7 +75,27 @@ module Join = struct
          List.filter_map (fun (lindex, lhash) -> Option.map (fun (rindex, rhash, _) -> lindex, rindex, lhash, rhash)
                                                  @@ List.find_opt (fun (_rindex, _rhash, references) -> List.exists ((=) lhash) references) references_locations_right) left_computed_hashes
        in List.map (fun (lindex, rindex, _, _) -> List.nth left_attributes lindex, List.nth right_attributes rindex) indexes *)
-
+  open Disk
+  let natural commit locations left_relation right_relation =
+    let left_scan = Scan.execute commit locations left_relation in
+    let right_scan = Scan.execute commit locations right_relation in
+    let common_attributes =
+      let left_attributes = commit.schema
+                            |> Executor.StringMap.find left_relation in
+      let right_attributes = commit.schema
+                             |> Executor.StringMap.find right_relation in
+      List.filter_map (fun left_attr -> if List.mem left_attr right_attributes
+                                        then Some (fst left_attr)
+                                        else None) left_attributes
+    in
+    let join_clause = fun ((_entity_id_1, left_tuple), (_entity_id_2,right_tuple)): bool ->
+      (* Check only the key name for now, not subtype, so literals must be equal for it to work *)
+      let left_tuple = Executor.StringMap.filter (fun key _ -> List.mem key common_attributes)
+                       @@ Executor.StringMap.of_list left_tuple in
+      let right_tuple = Executor.StringMap.filter (fun key _ -> List.mem key common_attributes)
+                        @@ Executor.StringMap.of_list right_tuple in
+      left_tuple = right_tuple
+    in NestedLoop.execute left_scan right_scan join_clause
 end
 
 module Filter = struct end
diff --git a/lib/protocol.ml b/lib/protocol.ml
index 5fe29d1..c67fb9a 100644
--- a/lib/protocol.ml
+++ b/lib/protocol.ml
@@ -1,9 +1,10 @@
 open Protocol_conv_xml
 open Sexplib0.Sexp_conv
+(* open Sexplib *)
 
 type tuple = string list [@@deriving sexp, protocol ~driver:(module Xml_light)]
 
-type command = Disk.Command.t [@@deriving sexp]
+type command = Disk.Command.t [@@deriving sexp, protocol ~driver:(module Xml_light)]
 
 type relation = {
   attribute_name : string;
@@ -11,6 +12,7 @@ type relation = {
   tuples : tuple;
 }
 [@@deriving sexp, protocol ~driver:(module Xml_light)]
-
-type facts = relation list
+(* (int64 * (string*Executor.relational_literal) list) list *)
+(* type facts = relation list *)
+type facts = (int64 * (string*Disk.Executor.relational_literal) list) list
 [@@deriving sexp, protocol ~driver:(module Xml_light)]
diff --git a/lib/relation.ml b/lib/relation.ml
index 8dad75d..a7b86c1 100644
--- a/lib/relation.ml
+++ b/lib/relation.ml
@@ -10,6 +10,179 @@ let print_references (references: (string * Disk.Executor.relational_type * stri
       print_newline ())
     references
 
+let prepare () =
+  let open Disk in
+  let open Executor in
+  let schema =
+    Executor.StringMap.empty
+    |> Executor.StringMap.add "user"
+         [
+           ("address", Relation "address");
+           ("last_name", Text);
+           ("first_name", Text);
+         ]
+    |> Executor.StringMap.add "address"
+         [ ("street", Text); ("number", Integer32) ]
+  in
+  let commit : Executor.commit =
+    {
+      state = Executor._EMPTY_SHA_HASH_;
+      files = Executor.StringMap.empty;
+      references = Executor.StringMap.empty;
+      schema;
+    }
+  in
+  let locations : Executor.locations = Executor.StringMap.empty in
+  let relation_name = "user" in
+  (* let entity_id, _ = Executor.StringMap.find relation_name references |> Executor.IntMap.max_binding in *)
+  (* let references =
+     Executor.StringMap.empty
+     |> Executor.StringMap.add "user" (Executor.StringMap.empty
+                                       |> Executor.StringMap.add "entity_0" [handle11; handle12]
+                                       |> Executor.StringMap.add "entity_1" [handle21; handle22]
+                                       |> Executor.StringMap.add "entity_2" [handle31; handle32]) in *)
+  let open Extensions.Result in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "Daisy";
+      entity_id = Some 0L;
+      attribute = "user/first-name";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  print_references commit.references;
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "Duck";
+      entity_id = Some 0L;
+      attribute = "user/last-name";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  print_references commit.references;
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "Saint James St.";
+      entity_id = Some 0L;
+      attribute = "address/street";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  print_references commit.references;
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LInteger32 41l;
+      entity_id = Some 0L;
+      attribute = "address/number";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  print_references commit.references;
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LRelation (0L, "address");
+      entity_id = Some 0L;
+      attribute = "user/address";
+      type' = Executor.Relation "address";
+    }
+    |> Command.transact commit locations
+  in
+  print_references commit.references;
+  let entity_id = 1L in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "Minnie";
+      entity_id = Some 1L;
+      attribute = "user/first-name";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "Mouse";
+      entity_id = Some 1L;
+      attribute = "user/last-name";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LRelation (0L, "address");
+      entity_id = Some 1L;
+      attribute = "user/address";
+      type' = Executor.Relation "address";
+    }
+    |> Command.transact commit locations
+  in
+  let entity_id = 2L in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "Scrooge";
+      entity_id = Some 2L;
+      attribute = "user/first-name";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "McDuck";
+      entity_id = Some 2L;
+      attribute = "user/last-name";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LText "President Julian St.";
+      entity_id = Some 1L;
+      attribute = "address/street";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  print_references commit.references;
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LInteger32 768l;
+      entity_id = Some 1L;
+      attribute = "address/number";
+      type' = Executor.Text;
+    }
+    |> Command.transact commit locations
+  in
+  let+ (commit, locations), _ =
+    {
+      timestamp = 10.0;
+      content = Executor.LRelation (1L, "address");
+      entity_id = Some 2L;
+      attribute = "user/address";
+      type' = Executor.Relation "address";
+    }
+    |> Command.transact commit locations
+  in
+  Ok (commit, locations)
+  [@@warning "-8-27-26"]
+
 let write_and_retrieve () =
   let open Disk in
   let open Executor in
@@ -184,6 +357,7 @@ let write_and_retrieve () =
     Command.perform commit locations (SequentialRead { relation_name = "user" })
   in
   print_endline (Command.show_return response);
-  Ok (commit, locations)
+  (* Ok (commit, locations) *)
+  Ok response
 [@@warning "-8-27-26"]
 (* Suppress pattern match incomplete warnings *)
diff --git a/lib/server.ml b/lib/server.ml
index 4af94fb..bc7dac2 100644
--- a/lib/server.ml
+++ b/lib/server.ml
@@ -28,34 +28,21 @@ let client_read sock maxlen =
   _read sock String.empty >>= fun x ->
   print_string "RECEIVED: ";
   print_endline x;
-  let command = Protocol.command_of_sexp (Sexplib0.Sexp_conv.sexp_of_string x) in
-  match
-    (* Relation.write_and_retrieve () *)
-    command
-  with
-  | Disk.Command.SequentialRead {relation}
-  (* | Ok (last_commit, locations) -> *)
-      (* let (first_names, last_names) = (Planner.Scan.execute last_commit locations "user/first-name", Planner.Scan.execute last_commit locations "user/last-name") in *)
-      (* let tuples = Planner.Join.execute first_names last_names locations in *)
-      let fname, lname = List.split [] in
-      (* tuples in *)
-      let relation_result : Protocol.facts =
-        [
-          {
-            attribute_name = "user/first-name";
-            attribute_type = "string";
-            tuples = List.map Bytes.to_string fname;
-          };
-          {
-            attribute_name = "user/last-name";
-            attribute_type = "string";
-            tuples = List.map Bytes.to_string lname;
-          };
-        ]
-      in
-      return @@ Xml.to_string (Protocol.facts_to_xml_light relation_result)
-  | Error err -> return err
-[@@warning "-27"]
+  let command = Protocol.command_of_sexp (Sexplib.Sexp.of_string x) in
+  (* let command = Protocol.command_of_xml_light_exn (Protocol_conv_xml.Xml_light.of_string x) in *)
+  let Ok (commit, locations) = Relation.prepare () in
+  let Ok (_, (Disk.Command.Read response)) =
+    Disk.Command.perform commit locations command
+  in return @@ Sexplib.Sexp.to_string (Protocol.sexp_of_facts response)
+  (* in return @@ Xml.to_string (Protocol.facts_to_xml_light response) *)
+  (* match *)
+  (* Relation.write_and_retrieve () *)
+  (* with *)
+  (* | Ok (Disk.Command.Read response) -> *)
+      (* return @@ Xml.to_string (Protocol.facts_to_xml_light response) *)
+      (* return @@ Sexplib0.Sexp_conv.string_of_sexp (Protocol.sexp_of_facts response) *)
+  (* | Error err -> return err *)
+[@@warning "-27-8-26"]
 
 let rec socket_read sock =
   (* (Int32.of_int (String.length results))) *)