From 919b7f1a81982533ff4ab33730c68313690c2b77 Mon Sep 17 00:00:00 2001 From: Marcos Magueta Date: Sun, 19 Jan 2025 21:50:01 -0300 Subject: cleanup and format --- .ocamlformat | 0 bin/main.ml | 14 +- lib/core_functions.ml | 23 +- lib/disk.ml | 487 +++++++++++++++++++++++++---------------- lib/dune | 12 +- lib/planner.ml | 50 ++--- lib/relation.ml | 163 ++++++++++---- lib/schema.ml | 38 ++-- lib/server.ml | 91 ++++---- lib/transaction.ml | 3 +- test/test_relational_engine.ml | 78 ++++--- 11 files changed, 601 insertions(+), 358 deletions(-) create mode 100644 .ocamlformat diff --git a/.ocamlformat b/.ocamlformat new file mode 100644 index 0000000..e69de29 diff --git a/bin/main.ml b/bin/main.ml index db591ff..5888b2b 100644 --- a/bin/main.ml +++ b/bin/main.ml @@ -1,9 +1,7 @@ let () = - begin - (* let response = *) - (* Relational_engine.Relation.write_and_retrieve() *) - (* in match response with *) - (* | Ok response -> print_endline response *) - (* | Error err -> print_endline err; *) - Relational_engine.Server.start() - end + (* let response = *) + (* Relational_engine.Relation.write_and_retrieve() *) + (* in match response with *) + (* | Ok response -> print_endline response *) + (* | Error err -> print_endline err; *) + Relational_engine.Server.start () diff --git a/lib/core_functions.ml b/lib/core_functions.ml index 7fdf1f1..6919f2f 100644 --- a/lib/core_functions.ml +++ b/lib/core_functions.ml @@ -5,19 +5,26 @@ module Types = Types_generated module Functions (F : Ctypes.FOREIGN) = struct (* open F *) open Ctypes_static + let os_extension = if Sys.os_type = "Unix" then let ic = Unix.open_process_in "uname" in let uname = input_line ic in let () = close_in ic in - if uname = "Darwin" then - "dylib" - else "so" + if uname = "Darwin" then "dylib" else "so" else "dll" + let libmerklecpp = - Dl.dlopen - ~flags:[RTLD_GLOBAL; RTLD_NOW] - ~filename:((Sys.getenv "LIBRELATIONAL_ENGINE_LIB_PATH") ^ "/librelational_engine." ^ os_extension) - let compute_hash = Foreign.foreign ~from:libmerklecpp "compute_hash" (ptr char @-> Ctypes.string @-> returning void) - let merkle_generate_root = Foreign.foreign ~from:libmerklecpp "merkle_generate_root" (ptr Ctypes.string @-> int @-> ptr char @-> returning void) + Dl.dlopen ~flags:[ RTLD_GLOBAL; RTLD_NOW ] + ~filename: + (Sys.getenv "LIBRELATIONAL_ENGINE_LIB_PATH" + ^ "/librelational_engine." ^ os_extension) + + let compute_hash = + Foreign.foreign ~from:libmerklecpp "compute_hash" + (ptr char @-> Ctypes.string @-> returning void) + + let merkle_generate_root = + Foreign.foreign ~from:libmerklecpp "merkle_generate_root" + (ptr Ctypes.string @-> int @-> ptr char @-> returning void) end diff --git a/lib/disk.ml b/lib/disk.ml index 96bb06a..d97a60c 100644 --- a/lib/disk.ml +++ b/lib/disk.ml @@ -1,34 +1,32 @@ module type Configuration = sig - val base_dir: string - val storage: string - val transactions: string - val assure: unit -> unit + val base_dir : string + val storage : string + val transactions : string + val assure : unit -> unit end -module DevelopmentConfiguration: Configuration = struct +module DevelopmentConfiguration : Configuration = struct let base_dir = "/tmp/relational-engine/" let storage = base_dir ^ "storage/" let transactions = base_dir ^ "transactions/" + let assure () = - if Sys.file_exists base_dir - then Sys.remove base_dir; + if Sys.file_exists base_dir then Sys.remove base_dir; Sys.mkdir base_dir 0o700; Sys.mkdir storage 0o700; - Sys.mkdir transactions 0o700; + Sys.mkdir transactions 0o700 end module type FileSystem = sig - type target = - | Storage - | Transaction - val append: target -> bytes -> (int*int64, string) result - val read: target -> int64 -> int -> (bytes, string) result + type target = Storage | Transaction + + val append : target -> bytes -> (int * int64, string) result + val read : target -> int64 -> int -> (bytes, string) result end -module FileSystem (C: Configuration): FileSystem = struct - type target = - | Storage - | Transaction +module FileSystem (C : Configuration) : FileSystem = struct + type target = Storage | Transaction + let target_to_string = function | Storage -> C.storage ^ "storage.re" | Transaction -> C.transactions ^ "transaction.re" @@ -49,15 +47,19 @@ module FileSystem (C: Configuration): FileSystem = struct Out_channel.close channel; Ok (Bytes.length content, offset_written) in - try - Out_channel.with_open_gen [Open_append; Open_binary; Open_creat] 0o666 (target_to_string target) register + try + Out_channel.with_open_gen + [ Open_append; Open_binary; Open_creat ] + 0o666 (target_to_string target) register with e -> - Error (Printf.sprintf "Error on appending to target '%s': %s" (target_to_string target) (Printexc.to_string e)) + Error + (Printf.sprintf "Error on appending to target '%s': %s" + (target_to_string target) (Printexc.to_string e)) - let read target (offset: int64) size = + let read target (offset : int64) size = print_endline ("READ_OFFSET: " ^ Int64.to_string offset); - print_endline ("READ_SIZE: " ^ (string_of_int ((Int64.to_int offset) + size))); - let register (channel: in_channel) = + print_endline ("READ_SIZE: " ^ string_of_int (Int64.to_int offset + size)); + let register (channel : in_channel) = let buffer = Bytes.create size in In_channel.seek channel offset; let _ = In_channel.really_input channel buffer 0 size in @@ -66,129 +68,195 @@ module FileSystem (C: Configuration): FileSystem = struct In_channel.close channel; Ok buffer in - try In_channel.with_open_gen [Open_binary; Open_rdonly] 0o666 (target_to_string target) register - with e -> Error (Printf.sprintf "Error on reading from target '%s': %s" (target_to_string target) (Printexc.to_string e)) + try + In_channel.with_open_gen + [ Open_binary; Open_rdonly ] + 0o666 (target_to_string target) register + with e -> + Error + (Printf.sprintf "Error on reading from target '%s': %s" + (target_to_string target) (Printexc.to_string e)) end module Executor = struct - module StringMap = Map.Make(String) - module IntMap = Map.Make(Int64) - module FS = FileSystem(DevelopmentConfiguration) + module StringMap = Map.Make (String) + module IntMap = Map.Make (Int64) + module FS = FileSystem (DevelopmentConfiguration) + module Location = struct - type t = - { offset: int64; - size: int; - hash: string } - [@@deriving show] + type t = { offset : int64; size : int; hash : string } [@@deriving show] end + module Hashes = struct - type t = { values: string list; - hash: string } - [@@deriving show] - type history = t list - [@@deriving show] + type t = { values : string list; hash : string } [@@deriving show] + type history = t list [@@deriving show] end - type commit = { state: string; files: Hashes.history StringMap.t } + + type references = string list IntMap.t StringMap.t + + type commit = { + state : string; + files : Hashes.history StringMap.t; + references : references; + } + type history = commit list type locations = Location.t StringMap.t - let _EMPTY_SHA_HASH_ = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + let _EMPTY_SHA_HASH_ = + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" (** Initializes the virtual file or opens it. *) - let init_file (files: Hashes.history StringMap.t) filename: Hashes.history StringMap.t = + let init_file (files : Hashes.history StringMap.t) filename : + Hashes.history StringMap.t = match StringMap.find_opt filename files with | None -> - let default: Hashes.t = { values = []; hash = _EMPTY_SHA_HASH_} in - let new_files = StringMap.add filename [default] files - in new_files - | Some _ -> - files + let default : Hashes.t = { values = []; hash = _EMPTY_SHA_HASH_ } in + let new_files = StringMap.add filename [ default ] files in + new_files + | Some _ -> files let compose_new_state files = let history = StringMap.to_list files in - let all_hashes = List.concat - @@ List.map (fun (_, (history:Hashes.history)) -> - List.map (fun ({hash; _}: Hashes.t) -> hash) history) history - in Interop.Merkle.merkle_generate_root all_hashes - - let write ({files;_} as commit: commit) (locations: locations) ~filename ?hash_to_replace (content: Bytes.t) = + let all_hashes = + List.concat + @@ List.map + (fun (_, (history : Hashes.history)) -> + List.map (fun ({ hash; _ } : Hashes.t) -> hash) history) + history + in + Interop.Merkle.merkle_generate_root all_hashes + + let write ({ files; references; _ } as commit : commit) + (locations : locations) ~filename ?hash_to_replace (content : Bytes.t) = let files = init_file files filename in - let computed_hash: string = Interop.Sha256.compute_hash content in + let computed_hash : string = Interop.Sha256.compute_hash content in match StringMap.find_opt computed_hash locations with - | None -> - begin match hash_to_replace with + | None -> ( + match hash_to_replace with | None -> - begin (* This instead should store a list of list to keep the history *) let files = - StringMap.update - filename + StringMap.update filename (function - | Some (file_hashes: Hashes.history) -> - let {values; _}: Hashes.t = List.hd file_hashes in - let new_state = Interop.Merkle.merkle_generate_root (computed_hash::values) in - Some ({values = computed_hash::values; hash = new_state}::file_hashes) - | None -> Some [{values = [computed_hash]; hash = computed_hash}]) + | Some (file_hashes : Hashes.history) -> + let { values; _ } : Hashes.t = List.hd file_hashes in + let new_state = + Interop.Merkle.merkle_generate_root + (computed_hash :: values) + in + Some + ({ values = computed_hash :: values; hash = new_state } + :: file_hashes) + | None -> + Some + [ { values = [ computed_hash ]; hash = computed_hash } ]) files in let open Extensions.Result in - let+ (size, offset) = FS.append Storage content in - let () = print_endline @@ "OFFSET: " ^ (Int64.to_string offset) in - let locations = StringMap.add computed_hash ({size; offset; hash = computed_hash}: Location.t) locations in - let commit = {state = compose_new_state files; files} - in Ok ((commit, locations), Some computed_hash) - end + let+ size, offset = FS.append Storage content in + let () = print_endline @@ "OFFSET: " ^ Int64.to_string offset in + let locations = + StringMap.add computed_hash + ({ size; offset; hash = computed_hash } : Location.t) + locations + in + let commit = + { state = compose_new_state files; files; references } + in + Ok ((commit, locations), Some computed_hash) | Some hash_to_replace -> - let update_fun = function - | Some (file_hashes: Hashes.history) -> - let {values; _}: Hashes.t = List.hd file_hashes in - let updated_entry = List.map (fun hash -> if hash = hash_to_replace then computed_hash else hash) values in - let new_entry: Hashes.t = {values = updated_entry; hash = Interop.Merkle.merkle_generate_root updated_entry} in - Some (new_entry::file_hashes) - | None -> Some [{values = [computed_hash]; hash = computed_hash}] - in - let files = StringMap.update filename update_fun files in - let open Extensions.Result in - let+ (size, offset) = FS.append Storage content in - let () = print_endline @@ "OFFSET: " ^ (Int64.to_string offset) in - let locations = StringMap.add computed_hash ({size; offset; hash = computed_hash}: Location.t) locations in - let commit = {state = compose_new_state files; files} - in Ok ((commit, locations), Some computed_hash) - end - | Some _ -> - begin match hash_to_replace with + let update_fun = function + | Some (file_hashes : Hashes.history) -> + let { values; _ } : Hashes.t = List.hd file_hashes in + let updated_entry = + List.map + (fun hash -> + if hash = hash_to_replace then computed_hash else hash) + values + in + let new_entry : Hashes.t = + { + values = updated_entry; + hash = Interop.Merkle.merkle_generate_root updated_entry; + } + in + Some (new_entry :: file_hashes) + | None -> + Some [ { values = [ computed_hash ]; hash = computed_hash } ] + in + let files = StringMap.update filename update_fun files in + let open Extensions.Result in + let+ size, offset = FS.append Storage content in + let () = print_endline @@ "OFFSET: " ^ Int64.to_string offset in + let locations = + StringMap.add computed_hash + ({ size; offset; hash = computed_hash } : Location.t) + locations + in + let commit = + { state = compose_new_state files; files; references } + in + Ok ((commit, locations), Some computed_hash)) + | Some _ -> ( + match hash_to_replace with | Some hash_to_replace -> - let update_fun = function - | Some (file_hashes: Hashes.history) -> - let {values; _}: Hashes.t = List.hd file_hashes in - let updated_entry = List.map (fun hash -> if hash = hash_to_replace then computed_hash else hash) values in - let new_entry: Hashes.t = {values = updated_entry; hash = Interop.Merkle.merkle_generate_root updated_entry} in - Some (new_entry::file_hashes) - | None -> Some [{values = [computed_hash]; hash = computed_hash}] - in - (* let location_update_fun = function - | Some (location: Location.t) -> - Some { location with references = references@location.references } - | None -> None (* This case is a bit weird to exist. Can we ever already add and get nothing? Refactor with the location passed on the `Some _` already *) - in *) - let files = StringMap.update filename update_fun files in - let commit = {state = compose_new_state files; files} in - (* let locations = StringMap.update computed_hash location_update_fun locations in *) - Ok ((commit, locations), Some computed_hash) - | None -> - Ok ((commit, locations), None) - end - - let read ({files; _} as _commit: commit) (locations: locations) ~filename = - let history: Hashes.history = StringMap.find filename files in - let {values = location_hashes; _}: Hashes.t = List.hd history in - let physical_locations = List.map (fun hash -> StringMap.find hash locations) location_hashes in + let update_fun = function + | Some (file_hashes : Hashes.history) -> + let { values; _ } : Hashes.t = List.hd file_hashes in + let updated_entry = + List.map + (fun hash -> + if hash = hash_to_replace then computed_hash else hash) + values + in + let new_entry : Hashes.t = + { + values = updated_entry; + hash = Interop.Merkle.merkle_generate_root updated_entry; + } + in + Some (new_entry :: file_hashes) + | None -> + Some [ { values = [ computed_hash ]; hash = computed_hash } ] + in + (* let location_update_fun = function + | Some (location: Location.t) -> + Some { location with references = references@location.references } + | None -> None (* This case is a bit weird to exist. Can we ever already add and get nothing? Refactor with the location passed on the `Some _` already *) + in *) + let files = StringMap.update filename update_fun files in + let commit = + { state = compose_new_state files; files; references } + in + (* let locations = StringMap.update computed_hash location_update_fun locations in *) + Ok ((commit, locations), Some computed_hash) + | None -> Ok ((commit, locations), None)) + + let read ({ files; _ } as _commit : commit) (locations : locations) ~filename + = + let history : Hashes.history = StringMap.find filename files in + let { values = location_hashes; _ } : Hashes.t = List.hd history in + let physical_locations = + List.map (fun hash -> StringMap.find hash locations) location_hashes + in (* Terrible! Here I read everything opening a new IO every time. We should be able to read with only one call, but changing the offset every time *) - let content = List.map (fun ({offset; size; _}: Location.t) -> match FS.read FS.Storage offset size with | Ok x -> x | Error err -> failwith err) physical_locations - in (*Bytes.concat Bytes.empty @@*) - List.rev content - let read_location ~hash (locations: locations) = - let ({offset; size; _}: Location.t) = StringMap.find hash locations - in match FS.read FS.Storage offset size with | Ok x -> x | Error err -> failwith err + let content = + List.map + (fun ({ offset; size; _ } : Location.t) -> + match FS.read FS.Storage offset size with + | Ok x -> x + | Error err -> failwith err) + physical_locations + in + (*Bytes.concat Bytes.empty @@*) + List.rev content + + let read_location ~hash (locations : locations) = + let ({ offset; size; _ } : Location.t) = StringMap.find hash locations in + match FS.read FS.Storage offset size with + | Ok x -> x + | Error err -> failwith err end module Startup = struct @@ -197,99 +265,138 @@ module Startup = struct end module Command = struct - module FS = FileSystem(DevelopmentConfiguration) + module FS = FileSystem (DevelopmentConfiguration) open Data_encoding - type command_kind = - | OPEN - | WRITE - | READ - (* | CREATE_RELATION *) + + type command_kind = OPEN | WRITE | READ + (* | CREATE_RELATION *) + let command_kind_encoding = - union [ - case ~title:"open" (Tag 0) - Data_encoding.empty - (function OPEN -> Some () | _ -> None) - (function () -> OPEN); - case ~title:"write" (Tag 1) - Data_encoding.empty - (function WRITE -> Some () | _ -> None) - (function () -> WRITE); - case ~title:"read" (Tag 2) - Data_encoding.empty - (function READ -> Some () | _ -> None) - (function () -> READ); - (* case ~title:"create_relation" (Tag 3) *) + union + [ + case ~title:"open" (Tag 0) Data_encoding.empty + (function OPEN -> Some () | _ -> None) + (function () -> OPEN); + case ~title:"write" (Tag 1) Data_encoding.empty + (function WRITE -> Some () | _ -> None) + (function () -> WRITE); + case ~title:"read" (Tag 2) Data_encoding.empty + (function READ -> Some () | _ -> None) + (function () -> READ); + (* case ~title:"create_relation" (Tag 3) *) (* Data_encoding.empty *) (* (function CREATE_RELATION -> Some () | _ -> None) *) (* (function () -> CREATE_RELATION); *) ] - - type t = - { kind: command_kind; - timestamp: float; - hash: string; - filename: string; - (* branch must be added here later *) - content: string } + + type t = { + kind : command_kind; + timestamp : float; + hash : string; + filename : string; + entity_id : int64 option; + (* branch must be added here later *) + content : string; + } + let command_encoding = - conv - (fun {kind; timestamp; hash; filename; content} -> (kind, timestamp, hash, filename, content)) - (fun (kind, timestamp, hash, filename, content) -> {kind; timestamp; hash; filename; content}) - Data_encoding.(tup5 command_kind_encoding float string string string) - + conv + (fun { kind; timestamp; hash; filename; entity_id; content } -> + (kind, timestamp, hash, filename, entity_id, content)) + (fun (kind, timestamp, hash, filename, entity_id, content) -> + { kind; timestamp; hash; filename; entity_id; content }) + Data_encoding.( + tup6 command_kind_encoding float string string (option int64) string) + let parse_command ~data = - let contract_encoding = Data_encoding.(tup4 command_kind_encoding string string string) in + let contract_encoding = + Data_encoding.( + tup5 command_kind_encoding string string (option int64) string) + in match Binary.of_bytes_opt contract_encoding data with - | Some (kind, hash, filename, content) -> Ok {kind; timestamp = Unix.time(); hash; filename; content} + | Some (kind, hash, filename, entity_id, content) -> + Ok + { kind; timestamp = Unix.time (); hash; filename; entity_id; content } | None -> Error "Failed to parse command" type return = | ComputedHash of string | Read of (int64 * bytes list) list | Nothing - [@@deriving show] - - type references = string list Executor.IntMap.t Executor.StringMap.t + [@@deriving show] (** TODO: This does not write atomically. If the system crashes while it attempts to write, it will corrupt. Solve this with a temporary file and move later. *) - let commit_and_perform (stream: Executor.history) (locations: Executor.locations) (references: references) ~entity_id command = - let open Extensions.Result in - let+ serialized_command = - Result.map_error (fun _ -> "Failed to serialize command to binary format.") - @@ Binary.to_bytes command_encoding command in - let+ _ = FS.append FS.Transaction serialized_command in - match command.kind with - | WRITE -> begin + let commit_and_perform (commit : Executor.commit) + (locations : Executor.locations) command = + let open Extensions.Result in + let+ serialized_command = + Result.map_error (fun _ -> + "Failed to serialize command to binary format.") + @@ Binary.to_bytes command_encoding command + in + let+ _ = FS.append FS.Transaction serialized_command in + match command.kind with + | WRITE -> ( let open Extensions.Result in - let+ ((commit, locations), computed_hash_handle) = Executor.write (List.hd stream) locations ~filename:command.filename @@ Bytes.of_string command.content in - let updated_stream = commit::stream in - match computed_hash_handle with - | Some computed_hash_handle -> - print_endline "WRITE A"; - let references = - let relation_name = List.hd @@ String.split_on_char '/' command.filename - in Executor.StringMap.update - relation_name - (function | Some entity -> - print_endline "WRITE SOME 1"; - Some (Executor.IntMap.update entity_id (function - | Some locations -> - print_endline "WRITE SOME 2"; - Some (computed_hash_handle::locations) - | None -> Some [computed_hash_handle]) entity) - | None -> Some (Executor.IntMap.empty |> Executor.IntMap.add entity_id [computed_hash_handle])) references in - Ok ((updated_stream, locations, references), ComputedHash computed_hash_handle) - | None -> Ok ((updated_stream, locations, references), Nothing) - end - | READ -> - let relation_name = List.hd @@ String.split_on_char '/' command.filename in + let+ (commit, locations), computed_hash_handle = + Executor.write commit locations ~filename:command.filename + @@ Bytes.of_string command.content + in + match (computed_hash_handle, command.entity_id) with + | Some computed_hash_handle, Some entity_id -> + print_endline "WRITE A"; + let references = + let relation_name = + List.hd @@ String.split_on_char '/' command.filename + in + Executor.StringMap.update relation_name + (function + | Some entity -> + print_endline "WRITE SOME 1"; + Some + (Executor.IntMap.update entity_id + (function + | Some locations -> + print_endline "WRITE SOME 2"; + Some (computed_hash_handle :: locations) + | None -> Some [ computed_hash_handle ]) + entity) + | None -> + Some + (Executor.IntMap.empty + |> Executor.IntMap.add entity_id + [ computed_hash_handle ])) + commit.references + in + Ok + ( (commit, locations, references), + ComputedHash computed_hash_handle ) + | Some _, None -> + Error "Cannot write an entity without its referential." + | None, _ -> Ok ((commit, locations, commit.references), Nothing)) + | READ -> + let relation_name = + List.hd @@ String.split_on_char '/' command.filename + in print_endline "READ"; print_endline relation_name; - let entities: string list Executor.IntMap.t = Executor.StringMap.find relation_name references in - let content = Executor.IntMap.fold (fun key hashes acc -> (key,List.map (fun location -> Executor.read_location ~hash:location locations) hashes)::acc) entities [] in + let entities : string list Executor.IntMap.t = + Executor.StringMap.find relation_name commit.references + in + let content = + Executor.IntMap.fold + (fun key hashes acc -> + ( key, + List.map + (fun location -> + Executor.read_location ~hash:location locations) + hashes ) + :: acc) + entities [] + in (* print_string "CONTENT: "; *) (* print_endline @@ Bytes.to_string content; *) - Ok ((stream, locations, references), Read content) - | _ -> Error "Unimplemented method" + Ok ((commit, locations, commit.references), Read content) + | _ -> Error "Unimplemented method" end diff --git a/lib/dune b/lib/dune index ec9fd1f..a84e9fa 100644 --- a/lib/dune +++ b/lib/dune @@ -1,7 +1,15 @@ (library (name relational_engine) - (libraries ctypes ctypes-foreign data-encoding lwt lwt.unix lwt-exit ppx_protocol_conv_xml_light) - (preprocess (pps ppx_deriving.show ppx_sexp_conv ppx_protocol_conv)) + (libraries + ctypes + ctypes-foreign + data-encoding + lwt + lwt.unix + lwt-exit + ppx_protocol_conv_xml_light) + (preprocess + (pps ppx_deriving.show ppx_sexp_conv ppx_protocol_conv)) ;; (c_library_flags -lssl -lcrypto -L %{env:LIBRESSL_LIB_PATH=}) ;; (foreign_stubs ;; (language cxx) diff --git a/lib/planner.ml b/lib/planner.ml index f0fbc59..1d4d830 100644 --- a/lib/planner.ml +++ b/lib/planner.ml @@ -1,33 +1,33 @@ module Scan = struct (* open Disk - let execute commit locations (entity: string) = - let binary = Executor.read commit locations ~filename:("schema~" ^ entity) in - let relation = - Result.map_error (fun _ -> "Failed to deserialize relation from binary format.") - @@ Data_encoding.Binary.of_bytes Schema.Protocol.relation_encoding (Bytes.concat Bytes.empty binary) in - match relation with - | Ok relation -> - begin - let attrs = List.map (fun (a: Schema.Protocol.attribute) -> relation.name ^ "/" ^ a.name) relation.attributes in - Executor.read commit locations ~filename:("schema~" ^ entity) - end - | Error err -> failwith err *) + let execute commit locations (entity: string) = + let binary = Executor.read commit locations ~filename:("schema~" ^ entity) in + let relation = + Result.map_error (fun _ -> "Failed to deserialize relation from binary format.") + @@ Data_encoding.Binary.of_bytes Schema.Protocol.relation_encoding (Bytes.concat Bytes.empty binary) in + match relation with + | Ok relation -> + begin + let attrs = List.map (fun (a: Schema.Protocol.attribute) -> relation.name ^ "/" ^ a.name) relation.attributes in + Executor.read commit locations ~filename:("schema~" ^ entity) + end + | Error err -> failwith err *) end module Join = struct (* 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 *) - (* Fix this when we persist the indexes *) - let left_computed_hashes: (int * string) list = List.mapi (fun index elem -> (index, Interop.Sha256.compute_hash elem)) left_attributes in - let right_computed_hashes: (int * string) list = List.mapi (fun index elem -> (index, Interop.Sha256.compute_hash elem)) right_attributes in - let references_locations_right = List.filter_map (fun (index, rhash) -> Option.map (fun (elem: Executor.Location.t) -> index, rhash, elem.references) - @@ Executor.StringMap.find_opt rhash locations) right_computed_hashes in - let indexes = - 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 *) -end + 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 *) + (* Fix this when we persist the indexes *) + let left_computed_hashes: (int * string) list = List.mapi (fun index elem -> (index, Interop.Sha256.compute_hash elem)) left_attributes in + let right_computed_hashes: (int * string) list = List.mapi (fun index elem -> (index, Interop.Sha256.compute_hash elem)) right_attributes in + let references_locations_right = List.filter_map (fun (index, rhash) -> Option.map (fun (elem: Executor.Location.t) -> index, rhash, elem.references) + @@ Executor.StringMap.find_opt rhash locations) right_computed_hashes in + let indexes = + 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 *) -module Filter = struct end + +module Filter = struct end diff --git a/lib/relation.ml b/lib/relation.ml index d6a1b70..d27f2e2 100644 --- a/lib/relation.ml +++ b/lib/relation.ml @@ -1,14 +1,19 @@ module Protocol = struct open Protocol_conv_xml open Sexplib0.Sexp_conv - type tuple = - string list [@@deriving sexp, protocol ~driver:(module Xml_light)] - type relation = - { attribute_name: string; - attribute_type: string; - tuples: tuple } - [@@deriving sexp, protocol ~driver:(module Xml_light)] - type facts = relation list [@@deriving sexp, protocol ~driver:(module Xml_light)] + + type tuple = string list + [@@deriving sexp, protocol ~driver:(module Xml_light)] + + type relation = { + attribute_name : string; + attribute_type : string; + tuples : tuple; + } + [@@deriving sexp, protocol ~driver:(module Xml_light)] + + type facts = relation list + [@@deriving sexp, protocol ~driver:(module Xml_light)] end (* @@ -49,43 +54,119 @@ let test_write_and_retrieve() = [@@warning "-8-27"] (* Suppress pattern match incomplete warnings *) *) -let write_and_retrieve() = +let write_and_retrieve () = let open Disk in - let stream: Executor.history = [{state = Executor._EMPTY_SHA_HASH_; files = Executor.StringMap.empty}] in - let locations: Executor.locations = Executor.StringMap.empty in - let references = Executor.StringMap.empty in + let commit : Executor.commit = + { + state = Executor._EMPTY_SHA_HASH_; + files = Executor.StringMap.empty; + references = Executor.StringMap.empty; + } + 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 entity_id = 0L 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 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 command_write1: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Daisy"; filename = "user/first-name"} in - let+ ((stream, locations, references), Command.ComputedHash handle11) = Command.commit_and_perform stream locations references ~entity_id command_write1 in - let command_write2: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Blossom"; filename = "user/last-name"} in - let+ ((stream, locations, references), Command.ComputedHash handle12) = Command.commit_and_perform stream locations references ~entity_id command_write2 in + let command_write1 : Command.t = + { + kind = Command.WRITE; + timestamp = 10.0; + hash = ""; + content = "Daisy"; + entity_id = Some 0L; + filename = "user/first-name"; + } + in + let+ (commit, locations, references), Command.ComputedHash handle11 = + Command.commit_and_perform commit locations command_write1 + in + let command_write2 : Command.t = + { + kind = Command.WRITE; + timestamp = 10.0; + hash = ""; + content = "Blossom"; + entity_id = Some 0L; + filename = "user/last-name"; + } + in + let+ (commit, locations, references), Command.ComputedHash handle12 = + Command.commit_and_perform commit locations command_write2 + in let entity_id = 1L in - let command_write3: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Minnie"; filename = "user/first-name"} in - let+ ((stream, locations, references), Command.ComputedHash handle21) = Command.commit_and_perform stream locations references ~entity_id command_write3 in - let command_write4: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Mouse"; filename = "user/last-name"} in - let+ ((stream, locations, references), Command.ComputedHash handle22) = Command.commit_and_perform stream locations references ~entity_id command_write4 in + let command_write3 : Command.t = + { + kind = Command.WRITE; + timestamp = 10.0; + hash = ""; + content = "Minnie"; + entity_id = Some 1L; + filename = "user/first-name"; + } + in + let+ (commit, locations, references), Command.ComputedHash handle21 = + Command.commit_and_perform commit locations command_write3 + in + let command_write4 : Command.t = + { + kind = Command.WRITE; + timestamp = 10.0; + hash = ""; + content = "Mouse"; + entity_id = Some 1L; + filename = "user/last-name"; + } + in + let+ (commit, locations, references), Command.ComputedHash handle22 = + Command.commit_and_perform commit locations command_write4 + in let entity_id = 2L in - let command_write5: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Billie"; filename = "user/first-name"} in - let+ ((stream, locations, references), Command.ComputedHash handle31) = Command.commit_and_perform stream locations references ~entity_id command_write5 in - let command_write6: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Beans"; filename = "user/last-name"} in - let+ ((stream, locations, references), Command.ComputedHash handle32) = Command.commit_and_perform stream locations references ~entity_id command_write6 in - let command_read: Command.t = {kind = Command.READ; timestamp = 0.0; hash = ""; content = ""; filename = "user"} in - let+ (_, (Command.Read content as response)) = Command.commit_and_perform (List.tl stream) locations references ~entity_id command_read in + let command_write5 : Command.t = + { + kind = Command.WRITE; + timestamp = 10.0; + hash = ""; + content = "Billie"; + entity_id = Some 2L; + filename = "user/first-name"; + } + in + let+ (commit, locations, references), Command.ComputedHash handle31 = + Command.commit_and_perform commit locations command_write5 + in + let command_write6 : Command.t = + { + kind = Command.WRITE; + timestamp = 10.0; + hash = ""; + content = "Beans"; + entity_id = Some 2L; + filename = "user/last-name"; + } + in + let+ (commit, locations, references), Command.ComputedHash handle32 = + Command.commit_and_perform commit locations command_write6 + in + let command_read : Command.t = + { + kind = Command.READ; + timestamp = 0.0; + hash = ""; + content = ""; + entity_id = None; + filename = "user"; + } + in + let+ _, (Command.Read content as response) = + Command.commit_and_perform commit locations command_read + in print_endline (Command.show_return response); - Ok (stream, locations) - [@@warning "-8-27-26"] (* Suppress pattern match incomplete warnings *) + Ok (commit, locations) +[@@warning "-8-27-26"] +(* Suppress pattern match incomplete warnings *) diff --git a/lib/schema.ml b/lib/schema.ml index b041803..90f8ee1 100644 --- a/lib/schema.ml +++ b/lib/schema.ml @@ -1,33 +1,33 @@ module Protocol = struct open Data_encoding - type attribute = - { name: string; - attr_type: string } - type relation = - { name: string; - attributes: attribute list } + + type attribute = { name : string; attr_type : string } + type relation = { name : string; attributes : attribute list } + let attribute_encoding = - conv - (fun {name; attr_type} -> (name, attr_type)) - (fun (name, attr_type) -> {name; attr_type}) + conv + (fun { name; attr_type } -> (name, attr_type)) + (fun (name, attr_type) -> { name; attr_type }) Data_encoding.(tup2 string string) - + let parse_attribute ~data = let contract_encoding = Data_encoding.(tup2 string string) in match Binary.of_bytes_opt contract_encoding data with - | Some (name, attr_type) -> Ok {name; attr_type} + | Some (name, attr_type) -> Ok { name; attr_type } | None -> Error "Failed to parse attribute" - + let relation_encoding = - conv - (fun {name; attributes} -> (name, attributes)) - (fun (name, attributes) -> {name; attributes}) + conv + (fun { name; attributes } -> (name, attributes)) + (fun (name, attributes) -> { name; attributes }) Data_encoding.(tup2 string (list attribute_encoding)) - + let parse_relation ~data = - let contract_encoding = Data_encoding.(tup2 string (list attribute_encoding)) in + let contract_encoding = + Data_encoding.(tup2 string (list attribute_encoding)) + in match Binary.of_bytes_opt contract_encoding data with - | Some (name, attributes) -> Ok {name; attributes} + | Some (name, attributes) -> Ok { name; attributes } | None -> Error "Failed to parse relation" end @@ -54,4 +54,4 @@ let test_schema_write_and_retrieve() = @@ Binary.of_bytes Protocol.relation_encoding (Bytes.concat Bytes.empty content) in Ok (stream, locations, user_relation) [@@warning "-8"] (* Suppress pattern match incomplete warnings *) -*) \ No newline at end of file +*) diff --git a/lib/server.ml b/lib/server.ml index cc233db..3fab513 100644 --- a/lib/server.ml +++ b/lib/server.ml @@ -1,60 +1,77 @@ open Lwt open Unix -let host = Unix.inet_addr_loopback +let host = Unix.inet_addr_loopback let port = 7524 -let create_socket (): Lwt_unix.file_descr t = +let create_socket () : Lwt_unix.file_descr t = let sock = Lwt_unix.socket PF_INET SOCK_STREAM 0 in Lwt_unix.set_close_on_exec sock; Lwt_unix.setsockopt sock Unix.SO_REUSEADDR true; let () = Lwt_unix.set_blocking sock false in - Lwt_io.write_line Lwt_io.stdout "🔍 Relational Engine Server Started" >>= fun () -> - Lwt_unix.bind sock @@ ADDR_INET(host, port) >>= (fun () -> Lwt_unix.listen sock 10; return sock) + Lwt_io.write_line Lwt_io.stdout "🔍 Relational Engine Server Started" + >>= fun () -> + Lwt_unix.bind sock @@ ADDR_INET (host, port) >>= fun () -> + Lwt_unix.listen sock 10; + return sock let client_read sock maxlen = let str = Bytes.create maxlen in let rec _read sock acc = - Lwt_unix.read sock str 0 maxlen - >>= fun recvlen -> Lwt_io.write_line Lwt_io.stdout (Printf.sprintf "[LOG] Received %d bytes from client" recvlen) - >>= fun () -> let acc = (acc ^ (Bytes.to_string (Bytes.sub str 0 recvlen))) in - if recvlen <= maxlen - then Lwt.return acc - else _read sock acc in - _read sock String.empty - >>= fun _ -> + Lwt_unix.read sock str 0 maxlen >>= fun recvlen -> + Lwt_io.write_line Lwt_io.stdout + (Printf.sprintf "[LOG] Received %d bytes from client" recvlen) + >>= fun () -> + let acc = acc ^ Bytes.to_string (Bytes.sub str 0 recvlen) in + if recvlen <= maxlen then Lwt.return acc else _read sock acc + in + _read sock String.empty >>= fun _ -> match Relation.write_and_retrieve () with - | Ok (last_commit::_, locations) -> - begin - (* 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: Relation.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 (Relation.Protocol.facts_to_xml_light relation_result) - end - | Ok _ -> return "Unstable" + | 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 : Relation.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 (Relation.Protocol.facts_to_xml_light relation_result) | Error err -> return err - [@@warning "-27"] +[@@warning "-27"] let rec socket_read sock = (* (Int32.of_int (String.length results))) *) - Lwt_unix.accept sock - >>= fun (client, _) -> client_read client 512 + Lwt_unix.accept sock >>= fun (client, _) -> + client_read client 512 (* >>= fun results -> Lwt_unix.send client (Data_encoding.Binary.to_bytes_exn Data_encoding.Encoding.int32 5l) 0 1 [] *) - >>= fun results -> Lwt_unix.send client (Bytes.of_string results) 0 (String.length results) [] - >>= fun _ -> Lwt_unix.wait_write client - >>= fun () -> Lwt_unix.close client - >>= fun () -> socket_read sock + >>= + fun results -> + Lwt_unix.send client (Bytes.of_string results) 0 (String.length results) [] + >>= fun _ -> + Lwt_unix.wait_write client >>= fun () -> + Lwt_unix.close client >>= fun () -> socket_read sock -let start() = +let start () = Stdlib.exit - @@ Lwt_main.run begin - Lwt_exit.wrap_and_exit (create_socket ()) >>= fun sock -> - let clean_callback = Lwt_exit.register_clean_up_callback ~loc:__LOC__ (fun _ -> Lwt_io.write_line Lwt_io.stdout "💥 Shutting down" >>= fun () -> Lwt_unix.close sock) in + @@ Lwt_main.run + ( Lwt_exit.wrap_and_exit (create_socket ()) >>= fun sock -> + let clean_callback = + Lwt_exit.register_clean_up_callback ~loc:__LOC__ (fun _ -> + Lwt_io.write_line Lwt_io.stdout "💥 Shutting down" >>= fun () -> + Lwt_unix.close sock) + in Lwt_exit.wrap_and_exit (socket_read sock) >>= fun () -> let () = Lwt_exit.unregister_clean_up_callback clean_callback in - Lwt_exit.exit_and_wait 0 - end + Lwt_exit.exit_and_wait 0 ) diff --git a/lib/transaction.ml b/lib/transaction.ml index 7ae98ab..d5f7672 100644 --- a/lib/transaction.ml +++ b/lib/transaction.ml @@ -1,2 +1 @@ -let transaction () = - () +let transaction () = () diff --git a/test/test_relational_engine.ml b/test/test_relational_engine.ml index 0c025fe..3092acc 100644 --- a/test/test_relational_engine.ml +++ b/test/test_relational_engine.ml @@ -1,38 +1,64 @@ - let%test _ = let open Relational_engine.Disk.Executor in - let stream: history = [{state = _EMPTY_SHA_HASH_; files = StringMap.empty}] in - let locations: locations = StringMap.empty in - let ((new_commit, _), _) = Result.get_ok @@ write (List.hd stream) locations ~filename:"trash" @@ Bytes.of_string "abc" in - let expected_state_hash = "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837830" in + let stream : history = + [ + { + state = _EMPTY_SHA_HASH_; + files = StringMap.empty; + references = StringMap.empty; + }; + ] + in + let locations : locations = StringMap.empty in + let (new_commit, _), _ = + Result.get_ok + @@ write (List.hd stream) locations ~filename:"trash" + @@ Bytes.of_string "abc" + in + let expected_state_hash = + "bca39cc55b19274133ce57d0bea1ff794589f025d4bcf670c44d5d2811837830" + in new_commit.state = expected_state_hash let%test _ = let open Relational_engine.Disk.Executor in - let stream: history = [{state = _EMPTY_SHA_HASH_; files = StringMap.empty}] in - let locations: locations = StringMap.empty in - let ((new_commit, new_locations), _) = Result.get_ok @@ write (List.hd stream) locations ~filename:"trash" @@ Bytes.of_string "abc" in + let stream : history = + [ + { + state = _EMPTY_SHA_HASH_; + files = StringMap.empty; + references = StringMap.empty; + }; + ] + in + let locations : locations = StringMap.empty in + let (new_commit, new_locations), _ = + Result.get_ok + @@ write (List.hd stream) locations ~filename:"trash" + @@ Bytes.of_string "abc" + in let result_test = let open Relational_engine.Extensions.Option in let+ new_file_content = StringMap.find_opt "trash" new_commit.files in - let { values; _ }: Hashes.t = List.hd new_file_content in + let { values; _ } : Hashes.t = List.hd new_file_content in StringMap.find_opt (List.hd values) new_locations - in match result_test with - | Some location -> - print_endline @@ Location.show location; - location.size = 3 && location.offset = 0L - | None -> false + in + match result_test with + | Some location -> + print_endline @@ Location.show location; + location.size = 3 && location.offset = 0L + | None -> false (* let%test _ = - let open Relational_engine.Planner in - match Relational_engine.Relation.write_and_retrieve () with - | Ok (last_commit::_, locations) -> - begin - let (first_names, last_names) = (Scan.execute last_commit locations "user/first-name", Scan.execute last_commit locations "user/last-name") in - let tuples = Join.execute first_names last_names locations in - let tuples = List.map (fun (a, b) -> Bytes.to_string a, Bytes.to_string b) tuples in - List.iter (fun (a, b) -> print_endline (Printf.sprintf "(%s, %s)" a b)) tuples; - true - end - | _ -> - false *) + let open Relational_engine.Planner in + match Relational_engine.Relation.write_and_retrieve () with + | Ok (last_commit::_, locations) -> + begin + let (first_names, last_names) = (Scan.execute last_commit locations "user/first-name", Scan.execute last_commit locations "user/last-name") in + let tuples = Join.execute first_names last_names locations in + let tuples = List.map (fun (a, b) -> Bytes.to_string a, Bytes.to_string b) tuples in + List.iter (fun (a, b) -> print_endline (Printf.sprintf "(%s, %s)" a b)) tuples; + true + end + | _ -> + false *) -- cgit v1.2.3