From 3a32e46f647314aea380449b31384ad121f31e4e Mon Sep 17 00:00:00 2001 From: Marcos Date: Sun, 19 Jan 2025 02:00:15 -0300 Subject: unite relation attributes by entity map --- lib/disk.ml | 59 ++++++++++++++++++++++++++++++--------------------------- lib/planner.ml | 9 ++++----- lib/relation.ml | 56 ++++++++++++++++++++++++++++++++---------------------- lib/schema.ml | 10 ++++++---- lib/server.ml | 8 +++++--- 5 files changed, 79 insertions(+), 63 deletions(-) (limited to 'lib') diff --git a/lib/disk.ml b/lib/disk.ml index ea863a5..85b37d7 100644 --- a/lib/disk.ml +++ b/lib/disk.ml @@ -77,8 +77,7 @@ module Executor = struct type t = { offset: int64; size: int; - hash: string; - references: string list } + hash: string } [@@deriving show] end module Hashes = struct @@ -110,7 +109,7 @@ module Executor = struct 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 references (content: Bytes.t) = + let write ({files;_} 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 match StringMap.find_opt computed_hash locations with @@ -133,7 +132,7 @@ module Executor = struct 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; references = references}: Location.t) locations 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 @@ -150,7 +149,7 @@ module Executor = struct 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; references = references}: Location.t) locations 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 @@ -165,15 +164,15 @@ module Executor = struct Some (new_entry::file_hashes) | None -> Some [{values = [computed_hash]; hash = computed_hash}] in - let location_update_fun = function + (* 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 + 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) + (* let locations = StringMap.update computed_hash location_update_fun locations in *) + Ok ((commit, locations), Some computed_hash) | None -> Ok ((commit, locations), None) end @@ -186,6 +185,9 @@ module Executor = struct 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 @@ -226,29 +228,29 @@ module Command = struct timestamp: float; hash: string; filename: string; - references: string list; (* branch must be added here later *) content: string } let command_encoding = conv - (fun {kind; timestamp; hash; filename; references; content} -> (kind, timestamp, hash, filename, references, content)) - (fun (kind, timestamp, hash, filename, references, content) -> {kind; timestamp; hash; filename; references; content}) - Data_encoding.(tup6 command_kind_encoding float string string (list string) string) + (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) let parse_command ~data = - let contract_encoding = Data_encoding.(tup5 command_kind_encoding string string (list string) string) in + let contract_encoding = Data_encoding.(tup4 command_kind_encoding string string string) in match Binary.of_bytes_opt contract_encoding data with - | Some (kind, hash, filename, references, content) -> Ok {kind; timestamp = Unix.time(); hash; filename; references; content} + | Some (kind, hash, filename, content) -> Ok {kind; timestamp = Unix.time(); hash; filename; content} | None -> Error "Failed to parse command" type return = | ComputedHash of string - | Read of bytes list + | Read of (string * bytes list) list | Nothing + [@@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) command = + let commit_and_perform (stream: Executor.history) (locations: Executor.locations) references command = let open Extensions.Result in let+ serialized_command = Result.map_error (fun _ -> "Failed to serialize command to binary format.") @@ -256,17 +258,18 @@ module Command = struct let+ _ = FS.append FS.Transaction serialized_command in match command.kind with | WRITE -> begin - let open Extensions.Result in - let+ ((commit, locations), computed_hash_handle) = Executor.write (List.hd stream) locations ~filename:command.filename command.references @@ Bytes.of_string command.content in - let updated_stream = commit::stream in - match computed_hash_handle with - | Some computed_hash_handle -> Ok ((updated_stream, locations), ComputedHash computed_hash_handle) - | None -> Ok ((updated_stream, locations), Nothing) - end + 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 -> Ok ((updated_stream, locations), ComputedHash computed_hash_handle) + | None -> Ok ((updated_stream, locations), Nothing) + end | READ -> - let content = Executor.read (List.hd stream) locations ~filename:command.filename in - (* print_string "CONTENT: "; *) - (* print_endline @@ Bytes.to_string content; *) - Ok ((stream, locations), Read content) + let entities: string list Executor.StringMap.t = Executor.StringMap.find command.filename references in + let content = Executor.StringMap.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), Read content) | _ -> Error "Unimplemented method" end diff --git a/lib/planner.ml b/lib/planner.ml index b88090d..f0fbc59 100644 --- a/lib/planner.ml +++ b/lib/planner.ml @@ -1,5 +1,5 @@ module Scan = struct - open Disk + (* open Disk let execute commit locations (entity: string) = let binary = Executor.read commit locations ~filename:("schema~" ^ entity) in let relation = @@ -11,11 +11,11 @@ module Scan = struct 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 + | Error err -> failwith err *) end module Join = struct - open Disk + (* 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 *) @@ -26,8 +26,7 @@ module Join = struct 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 - + in List.map (fun (lindex, rindex, _, _) -> List.nth left_attributes lindex, List.nth right_attributes rindex) indexes *) end module Filter = struct diff --git a/lib/relation.ml b/lib/relation.ml index ed644dd..2f0c2a5 100644 --- a/lib/relation.ml +++ b/lib/relation.ml @@ -11,32 +11,33 @@ module Protocol = struct type facts = relation list [@@deriving sexp, protocol ~driver:(module Xml_light)] end +(* let test_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 open Extensions.Result in let command_write1: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Daisy"; filename = "user/first-name"; references = []} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Daisy"; filename = "user/first-name"} in let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write1 in let command_write2: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Blossom"; filename = "user/last-name"; references = [handle]} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Blossom"; filename = "user/last-name"} in let command_write3: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Minnie"; filename = "user/first-name"; references = []} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Minnie"; filename = "user/first-name"} in let+ ((stream, locations), Command.ComputedHash _handle) = Command.commit_and_perform stream locations command_write2 in let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write3 in let command_write4: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Mouse"; filename = "user/last-name"; references = [handle]} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Mouse"; filename = "user/last-name"} in let+ ((stream, locations), Command.ComputedHash _handle) = Command.commit_and_perform stream locations command_write4 in let command_write5: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Billie"; filename = "user/first-name"; references = []} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Billie"; filename = "user/first-name"} in let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write5 in let command_write6: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Beans"; filename = "user/last-name"; references = [handle]} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Beans"; filename = "user/last-name"} in let+ ((stream, locations), Command.ComputedHash _handle) = Command.commit_and_perform stream locations command_write6 in - let command_read1: Command.t = {kind = Command.READ; timestamp = 0.0; hash = ""; content = ""; filename = "user/first-name"; references = []} in + let command_read1: Command.t = {kind = Command.READ; timestamp = 0.0; hash = ""; content = ""; filename = "user/first-name"} in let+ (_, Command.Read first_name) = Command.commit_and_perform (List.tl stream) locations command_read1 in - let command_read2: Command.t = {kind = Command.READ; timestamp = 0.0; hash = ""; content = ""; filename = "user/last-name"; references = []} in + let command_read2: Command.t = {kind = Command.READ; timestamp = 0.0; hash = ""; content = ""; filename = "user/last-name"} in let+ (_, Command.Read last_name) = Command.commit_and_perform (List.tl stream) locations command_read2 in (* List.iter (fun a -> print_endline a) (List.map Bytes.to_string first_name); *) (* List.iter (fun a -> print_endline a) (List.map Bytes.to_string last_name); *) @@ -45,8 +46,8 @@ let test_write_and_retrieve() = {attribute_name = "user/last-name"; attribute_type = "string"; tuples = List.map Bytes.to_string last_name}] in Ok (Xml.to_string (Protocol.facts_to_xml_light relation_result)) (* (Sexp.to_string (Protocol.sexp_of_t )) *) - [@@warning "-8"] (* Suppress pattern match incomplete warnings *) - + [@@warning "-8-27"] (* Suppress pattern match incomplete warnings *) +*) let write_and_retrieve() = let open Disk in @@ -54,22 +55,31 @@ let write_and_retrieve() = let locations: Executor.locations = Executor.StringMap.empty in let open Extensions.Result in let command_write1: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Daisy"; filename = "user/first-name"; references = []} in - let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write1 in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Daisy"; filename = "user/first-name"} in + let+ ((stream, locations), Command.ComputedHash handle11) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write1 in let command_write2: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Blossom"; filename = "user/last-name"; references = [handle]} in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Blossom"; filename = "user/last-name"} in + let+ ((stream, locations), Command.ComputedHash handle12) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write2 in let command_write3: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Minnie"; filename = "user/first-name"; references = []} in - let+ ((stream, locations), Command.ComputedHash _handle) = Command.commit_and_perform stream locations command_write2 in - let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write3 in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Minnie"; filename = "user/first-name"} in + let+ ((stream, locations), Command.ComputedHash handle21) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write3 in let command_write4: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Mouse"; filename = "user/last-name"; references = [handle]} in - let+ ((stream, locations), Command.ComputedHash _handle) = Command.commit_and_perform stream locations command_write4 in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Mouse"; filename = "user/last-name"} in + let+ ((stream, locations), Command.ComputedHash handle22) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write4 in let command_write5: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Billie"; filename = "user/first-name"; references = []} in - let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write5 in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Billie"; filename = "user/first-name"} in + let+ ((stream, locations), Command.ComputedHash handle31) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write5 in let command_write6: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Beans"; filename = "user/last-name"; references = [handle]} in - let+ ((stream, locations), Command.ComputedHash _handle) = Command.commit_and_perform stream locations command_write6 in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = "Beans"; filename = "user/last-name"} in + let+ ((stream, locations), Command.ComputedHash handle32) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write6 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 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 command_read in + print_endline (Command.show_return response); Ok (stream, locations) - [@@warning "-8"] (* Suppress pattern match incomplete warnings *) + [@@warning "-8-27"] (* Suppress pattern match incomplete warnings *) diff --git a/lib/schema.ml b/lib/schema.ml index 81632b4..b041803 100644 --- a/lib/schema.ml +++ b/lib/schema.ml @@ -31,6 +31,7 @@ module Protocol = struct | None -> Error "Failed to parse relation" end +(* let test_schema_write_and_retrieve() = let open Disk in let open Data_encoding in @@ -43,13 +44,14 @@ let test_schema_write_and_retrieve() = @@ Binary.to_bytes Protocol.relation_encoding relation in let command_write: Command.t = - {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = Bytes.to_string user_relation; filename = "user"; references = []} in - let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations command_write in + {kind = Command.WRITE; timestamp = 10.0; hash = ""; content = Bytes.to_string user_relation; filename = "user"} in + let+ ((stream, locations), Command.ComputedHash handle) = Command.commit_and_perform stream locations Executor.StringMap.empty command_write in let command_read: Command.t = - {kind = Command.READ; timestamp = 10.0; hash = handle; content = ""; filename = "schema~user"; references = []} in - let+ ((stream, locations), Command.Read content) = Command.commit_and_perform stream locations command_read in + {kind = Command.READ; timestamp = 10.0; hash = handle; content = ""; filename = "schema~user"} in + let+ ((stream, locations), Command.Read content) = Command.commit_and_perform stream locations Executor.StringMap.empty command_read in let+ user_relation = Result.map_error (fun _ -> "Failed to deserialize relation from binary format.") @@ 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 3a6d1fa..cc233db 100644 --- a/lib/server.ml +++ b/lib/server.ml @@ -26,9 +26,10 @@ let client_read sock maxlen = 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 tuples in + (* 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 @@ -36,6 +37,7 @@ let client_read sock maxlen = end | Ok _ -> return "Unstable" | Error err -> return err + [@@warning "-27"] let rec socket_read sock = (* (Int32.of_int (String.length results))) *) -- cgit v1.2.3