From fdc63e730f91a4605ed4ff96a908e1f5e822d1bb Mon Sep 17 00:00:00 2001 From: Marcos Magueta Date: Mon, 20 Apr 2026 21:55:07 -0300 Subject: New Server & Manipulation refactor Co-authored-by: Estevan Castilho --- lib/algebra.ml | 2 +- lib/conventions.ml | 3 +- lib/dbms_language.ml | 37 ----- lib/dcl/executor.ml | 147 +++++++++--------- lib/ddl/executor.ml | 8 +- lib/dml/executor.ml | 14 +- lib/envelope.ml | 16 +- lib/error.ml | 34 +++++ lib/icl/executor.ml | 8 +- lib/listener.ml | 188 +++++++++++++++++++++++ lib/management/branch.ml | 4 +- lib/management/merge.ml | 1 + lib/manipulation.ml | 388 +++++++++++++++++++++++------------------------ lib/manipulation.mli | 137 ----------------- lib/prelude/standard.ml | 15 +- lib/relation.ml | 2 +- lib/transport.ml | 41 +++++ 17 files changed, 565 insertions(+), 480 deletions(-) delete mode 100644 lib/dbms_language.ml create mode 100644 lib/error.ml create mode 100644 lib/listener.ml delete mode 100644 lib/manipulation.mli create mode 100644 lib/transport.ml (limited to 'lib') diff --git a/lib/algebra.ml b/lib/algebra.ml index 30ed1a6..d8e766f 100644 --- a/lib/algebra.ml +++ b/lib/algebra.ml @@ -44,7 +44,7 @@ module Make (Storage : Management.Physical.S) = struct in Relation.make ~hash:None ~name ~schema ~tree:None ~constraints ~cardinality ~generator:(Some gen) - ~membership_criteria:(fun _ -> true) + ~membership_criteria:(fun _ _ -> true) ~provenance:Relation.Provenance.Undefined ~lineage:(Relation.Lineage.Base "derived") diff --git a/lib/conventions.ml b/lib/conventions.ml index de04712..93dd579 100644 --- a/lib/conventions.ml +++ b/lib/conventions.ml @@ -25,7 +25,8 @@ end module AbstractValue = struct type t = Obj.t - let hash (elem : t) = Sha256.to_hex (Sha256.string (Bytes.to_string (Marshal.to_bytes elem []))) + let hash (elem : t) = + Sha256.to_hex (Sha256.string (Bytes.to_string (Marshal.to_bytes elem []))) let sexp_of_t (v : t) = let open Sexplib.Sexp in diff --git a/lib/dbms_language.ml b/lib/dbms_language.ml deleted file mode 100644 index 47e5a99..0000000 --- a/lib/dbms_language.ml +++ /dev/null @@ -1,37 +0,0 @@ -(** Composable dispatch over sublanguages. - - All commands must use the tagged envelope format: (tag payload) where [tag] - is the sublanguage name (drl, dml, ddl, icl, dcl, ...). Adding a new - sublanguage is just extending the list passed to [create]. *) - -module Make (Storage : Management.Physical.S with type error = string) = struct - module type SubS = Sublanguage.S with type storage = Storage.t - - type dispatch_error = NoMatch of Sexplib.Sexp.t - type dispatch = (module SubS) list - - let create (langs : (module SubS) list) : dispatch = langs - - let execute (langs : dispatch) (storage : Storage.t) - (db : Management.Database.t) (input : string) : - (Sublanguage.result, dispatch_error) Result.t = - let no_match s = - Error (NoMatch Sexplib.Sexp.(List [ Atom "parse-error"; Atom s ])) - in - match Envelope.parse input with - | Error msg -> no_match msg - | Ok { tag; payload } -> ( - match List.find_opt (fun (module L : SubS) -> L.name = tag) langs with - | None -> no_match ("unknown sublanguage: " ^ tag) - | Some (module L) -> ( - match L.parse_sexp payload with - | Error e -> Error (NoMatch (L.sexp_of_error e)) - | Ok ast -> ( - match L.execute storage db ast with - | Ok result -> Ok result - | Error e -> Error (NoMatch (L.sexp_of_error e))))) - - let sexp_of_dispatch_error (NoMatch sexp) = sexp -end - -module Memory = Make (Management.Physical.Memory) diff --git a/lib/dcl/executor.ml b/lib/dcl/executor.ml index 8c2fbdc..a651067 100644 --- a/lib/dcl/executor.ml +++ b/lib/dcl/executor.ml @@ -4,16 +4,24 @@ module Make (Storage : Management.Physical.S with type error = string) = struct module MergeOps = Management.Merge.Make (Storage) (Ops) type error = - | ParseError of string - | BranchError of string - | MergeError of string + | ParseError of string + | BranchError of Error.t + | BranchNotFound of string + | NoDatabaseAtHash of Conventions.Hash.t + | StorageError of string + | MergeError of Error.t + | Conflict of string let sexp_of_error e = let open Sexplib.Sexp in match e with - | ParseError s -> List [ Atom "parse-error"; Atom s ] - | BranchError s -> List [ Atom "branch-error"; Atom s ] - | MergeError s -> List [ Atom "merge-error"; Atom s ] + | ParseError s -> List [Atom "parse-error"; Atom s] + | BranchError s -> List [Atom "branch-error"; Error.sexp_of_error s] + | BranchNotFound s -> List [Atom "branch-not-found"; Atom s] + | NoDatabaseAtHash s -> List [Atom "no-database-at-hash"; Atom s] + | StorageError s -> List [Atom "storage-error"; Atom s] + | MergeError s -> List [Atom "merge-error"; Error.sexp_of_error s] + | Conflict s -> List [Atom "conflict"; Atom s] let convert_strategy : Ast.merge_strategy -> Management.Merge.strategy = function @@ -24,71 +32,68 @@ module Make (Storage : Management.Physical.S with type error = string) = struct let execute (storage : Storage.t) (db : Management.Database.t) (stmt : Ast.statement) : (Management.Database.t * string, error) result = match stmt with - | Ast.CreateBranch { name; hash } -> ( - let tip = - match hash with Some h -> h | None -> db.Management.Database.hash - in - match Branch.create storage ~name ~tip with - | Error e -> Error (BranchError e) - | Ok () -> Ok (db, "Branch " ^ name ^ " created")) - | Ast.Checkout branch_name -> ( - match Branch.get_tip storage branch_name with - | Error e -> Error (BranchError e) - | Ok None -> Error (BranchError ("Branch not found: " ^ branch_name)) - | Ok (Some tip) -> ( - match Branch.checkout storage branch_name with - | Error e -> Error (BranchError e) - | Ok () -> ( - match Ops.load_database storage tip with - | Error e -> - Error (BranchError (Manipulation.string_of_error e)) - | Ok None -> Error (BranchError ("No database at hash: " ^ tip)) - | Ok (Some loaded_db) -> Ok (loaded_db, "HEAD:" ^ branch_name))) - ) - | Ast.GetHead -> ( - match Branch.get_head storage with - | Error e -> Error (BranchError e) - | Ok None -> Ok (db, "HEAD is unset") - | Ok (Some name) -> Ok (db, "HEAD:" ^ name)) - | Ast.GetBranchTip name -> ( - match Branch.get_tip storage name with - | Error e -> Error (BranchError e) - | Ok None -> Error (BranchError ("Branch not found: " ^ name)) - | Ok (Some hash) -> Ok (db, "branch:" ^ name ^ "=" ^ hash)) - | Ast.UpdateBranchTip { name; hash } -> ( - match Branch.update_tip storage ~name ~tip:hash with - | Error e -> Error (BranchError e) - | Ok () -> Ok (db, "Branch " ^ name ^ " updated")) - | Ast.Merge { left; right; strategy } -> ( - match (Branch.get_tip storage left, Branch.get_tip storage right) with - | Error e, _ | _, Error e -> Error (BranchError e) - | Ok None, _ -> Error (BranchError ("Branch not found: " ^ left)) - | _, Ok None -> Error (BranchError ("Branch not found: " ^ right)) - | Ok (Some left_tip), Ok (Some right_tip) -> ( - let strat = convert_strategy strategy in - match - MergeOps.merge ~storage ~strategy:strat ~left_tip ~right_tip - with - | Error e -> Error (MergeError (Manipulation.string_of_error e)) - | Ok (Management.Merge.Failed conflicts) -> - let msgs = - List.map - (function - | Management.Merge.TupleConflict { relation; hash } -> - "tuple conflict in " ^ relation ^ " (" ^ hash ^ ")" - | Management.Merge.SchemaConflict r -> - "schema conflict in " ^ r - | Management.Merge.ConstraintConflict r -> - "constraint conflict in " ^ r) - conflicts - in - Error (MergeError ("Merge failed: " ^ String.concat "; " msgs)) - | Ok (Management.Merge.Clean merged_db) -> - let _ = - Branch.update_tip storage ~name:left - ~tip:merged_db.Management.Database.hash - in - Ok (merged_db, "Merged:" ^ right ^ "→" ^ left))) + | Ast.CreateBranch { name; hash } -> + let tip = match hash with + | Some h -> h + | None -> db.Management.Database.hash + in + (match Branch.create storage ~name ~tip with + | Error e -> Error (StorageError e) + | Ok () -> Ok (db, "Branch " ^ name ^ " created")) + + | Ast.Checkout branch_name -> + (match Branch.get_tip storage branch_name with + | Error e -> Error (StorageError e) + | Ok None -> Error (BranchNotFound branch_name) + | Ok (Some tip) -> + (match Branch.checkout storage branch_name with + | Error e -> Error (StorageError e) + | Ok () -> + (match Ops.load_database storage tip with + | Error e -> Error (BranchError e) + | Ok None -> Error (NoDatabaseAtHash tip) + | Ok (Some loaded_db) -> + Ok (loaded_db, "HEAD:" ^ branch_name)))) + + | Ast.GetHead -> + (match Branch.get_head storage with + | Error e -> Error (StorageError e) + | Ok None -> Ok (db, "HEAD is unset") + | Ok (Some name) -> Ok (db, "HEAD:" ^ name)) + + | Ast.GetBranchTip name -> + (match Branch.get_tip storage name with + | Error e -> Error (StorageError e) + | Ok None -> Error (BranchNotFound name) + | Ok (Some hash) -> Ok (db, "branch:" ^ name ^ "=" ^ hash)) + + | Ast.UpdateBranchTip { name; hash } -> + (match Branch.update_tip storage ~name ~tip:hash with + | Error e -> Error (StorageError e) + | Ok () -> Ok (db, "Branch " ^ name ^ " updated")) + + | Ast.Merge { left; right; strategy } -> + (match Branch.get_tip storage left, Branch.get_tip storage right with + | Error e, _ | _, Error e -> Error (StorageError e) + | Ok None, _ -> Error (BranchNotFound left) + | _, Ok None -> Error (BranchNotFound right) + | Ok (Some left_tip), Ok (Some right_tip) -> + let strat = convert_strategy strategy in + (match MergeOps.merge ~storage ~strategy:strat ~left_tip ~right_tip with + | Error e -> Error (MergeError e) + | Ok (Management.Merge.Failed conflicts) -> + let msgs = List.map (function + | Management.Merge.TupleConflict { relation; hash } -> + "tuple conflict in " ^ relation ^ " (" ^ hash ^ ")" + | Management.Merge.SchemaConflict r -> "schema conflict in " ^ r + | Management.Merge.ConstraintConflict r -> "constraint conflict in " ^ r + ) conflicts in + Error (Conflict ("Merge failed: " ^ String.concat "; " msgs)) + | Ok (Management.Merge.Clean merged_db) -> + let _ = Branch.update_tip storage ~name:left + ~tip:merged_db.Management.Database.hash in + Ok (merged_db, + "Merged:" ^ right ^ "→" ^ left))) end module Memory = Make (Management.Physical.Memory) diff --git a/lib/ddl/executor.ml b/lib/ddl/executor.ml index d374ffe..9df878d 100644 --- a/lib/ddl/executor.ml +++ b/lib/ddl/executor.ml @@ -3,15 +3,15 @@ module Make (Storage : Management.Physical.S) = struct type error = | ParseError of string - | ManipulationError of Manipulation.error + | ManipulationError of Error.t | RelationNotFound of string let sexp_of_error e = let open Sexplib.Sexp in match e with - | ParseError s -> List [ Atom "parse-error"; Atom s ] - | ManipulationError e -> Manipulation.sexp_of_error e - | RelationNotFound s -> List [ Atom "relation-not-found"; Atom s ] + | ParseError s -> List [Atom "parse-error"; Atom s] + | ManipulationError e -> Error.sexp_of_error e + | RelationNotFound s -> List [Atom "relation-not-found"; Atom s] let ( let* ) = Result.bind let wrap_manip r = Result.map_error (fun e -> ManipulationError e) r diff --git a/lib/dml/executor.ml b/lib/dml/executor.ml index 08f1a26..85f9791 100644 --- a/lib/dml/executor.ml +++ b/lib/dml/executor.ml @@ -5,20 +5,18 @@ module Make (Storage : Management.Physical.S) = struct type error = | ParseError of string - | ManipulationError of Manipulation.error + | ManipulationError of Error.t | RelationNotFound of string | AlgebraError of Algebra.error let sexp_of_error e = let open Sexplib.Sexp in match e with - | ParseError s -> List [ Atom "parse-error"; Atom s ] - | ManipulationError e -> Manipulation.sexp_of_error e - | RelationNotFound s -> List [ Atom "relation-not-found"; Atom s ] - | AlgebraError (Algebra.StorageError s) -> - List [ Atom "storage-error"; Atom s ] - | AlgebraError (Algebra.GeneratorError s) -> - List [ Atom "generator-error"; Atom s ] + | ParseError s -> List [Atom "parse-error"; Atom s] + | ManipulationError e -> Error.sexp_of_error e + | RelationNotFound s -> List [Atom "relation-not-found"; Atom s] + | AlgebraError (Algebra.StorageError s) -> List [Atom "storage-error"; Atom s] + | AlgebraError (Algebra.GeneratorError s) -> List [Atom "generator-error"; Atom s] let ( let* ) = Result.bind let wrap_manip r = Result.map_error (fun e -> ManipulationError e) r diff --git a/lib/envelope.ml b/lib/envelope.ml index bed8bc0..4b8bc46 100644 --- a/lib/envelope.ml +++ b/lib/envelope.ml @@ -6,12 +6,12 @@ type tagged = { tag : string; payload : Sexplib.Sexp.t } -let parse (input : string) : (tagged, string) Result.t = - match Sexplib.Sexp.of_string input with +let parse (input : Sexplib.Sexp.t) : (tagged, Error.t) Result.t = + match input with | Sexplib.Sexp.List (Sexplib.Sexp.Atom tag :: rest) -> - let payload = - match rest with [ single ] -> single | many -> Sexplib.Sexp.List many - in - Ok { tag; payload } - | _ -> Error "expected (tag payload...)" - | exception exn -> Error (Printexc.to_string exn) + let payload = match rest with + | [single] -> single + | many -> Sexplib.Sexp.List many + in + Ok { tag; payload } + | s -> Error (Error.MalformedExpression s) diff --git a/lib/error.ml b/lib/error.ml new file mode 100644 index 0000000..edaa85b --- /dev/null +++ b/lib/error.ml @@ -0,0 +1,34 @@ + +(** Error types for manipulation operations *) +type t = + | RelationNotFound of string + | RelationAlreadyExists of string + | TupleNotFound of Conventions.Hash.t + | DuplicateTuple of Conventions.Hash.t + | ConstraintViolation of string + | StorageError of string + | UnrecognizedSublanguage of string + | MalformedExpression of Sexplib.Sexp.t + (* TODO: think of a better way to manage and serialize sublanguage errors *) + | SublanguageError of Sexplib.Sexp.t + | Conflict of { old_db : Management.Database.t; new_db : Management.Database.t} + | SyntaxError of string + +let sexp_of_error e = + let open Sexplib.Sexp in + let error e ps = List (Atom e :: ps) in + let (<+>) key value = List [Atom key; value] in + match e with + (* TODO: think of a subset of common attributes that every error should have (message, etc)? *) + | RelationNotFound s -> error "relation-not-found" ["relation" <+> (Atom s)] + | RelationAlreadyExists s -> error "relation-already-exists" ["relation" <+> (Atom s)] + | TupleNotFound h -> error "tuple-not-found" ["hash" <+> (Atom h)] + | DuplicateTuple h -> error "duplicate-tuple" ["hash" <+> (Atom h)] + | ConstraintViolation s -> error "constraint-violation" ["message" <+> (Atom s)] + | StorageError s -> error "storage-error" ["message" <+> (Atom s)] + | UnrecognizedSublanguage s -> error "unrecognized-sublanguage" ["tag" <+> (Atom s)] + | MalformedExpression s -> error "malformed-expression" ["expression" <+> s] + | SublanguageError s -> error "sublanguage-error" ["error" <+> s] + | SyntaxError s -> error "syntax-error" ["message" <+> (Atom s)] + | Conflict { old_db; new_db } -> error "conflict" ["old-hash" <+> (Atom old_db.hash); + "new-hash" <+> (Atom new_db.hash)] diff --git a/lib/icl/executor.ml b/lib/icl/executor.ml index 07d0978..6238906 100644 --- a/lib/icl/executor.ml +++ b/lib/icl/executor.ml @@ -3,15 +3,15 @@ module Make (Storage : Management.Physical.S) = struct type error = | ParseError of string - | ManipulationError of Manipulation.error + | ManipulationError of Error.t | ConversionError of string let sexp_of_error e = let open Sexplib.Sexp in match e with - | ParseError s -> List [ Atom "parse-error"; Atom s ] - | ManipulationError e -> Manipulation.sexp_of_error e - | ConversionError s -> List [ Atom "conversion-error"; Atom s ] + | ParseError s -> List [Atom "parse-error"; Atom s] + | ManipulationError e -> Error.sexp_of_error e + | ConversionError s -> List [Atom "conversion-error"; Atom s] let wrap_manip = Result.map_error (fun e -> ManipulationError e) diff --git a/lib/listener.ml b/lib/listener.ml new file mode 100644 index 0000000..d7951aa --- /dev/null +++ b/lib/listener.ml @@ -0,0 +1,188 @@ +module type LISTENER = functor (T : Transport.TRANSPORT)(S : Management.Physical.S with type error = string) -> sig + val run : T.t -> Management.Database.t -> S.t -> unit +end + +module Make : LISTENER = + functor (T : Transport.TRANSPORT)(S : Management.Physical.S with type error = string) -> struct + module type SubS = Sublanguage.S with type storage = S.t + module BranchOps = Management.Branch.Make (S) + module AlgebraOps = Algebra.Make (S) + + let read_command input = + try + Ok (Sexplib.Sexp.input_sexp input) + with Sexplib.Sexp.Parse_error { err_msg ; _} -> + Error (Error.SyntaxError err_msg) + + let sublanguages = (* should this go here? *) + List.fold_right + (fun (module Language : SubS) -> Utilities.StringMap.add Language.name (module Language : SubS)) + [ + (module Drl.Sublanguage.Make(S) : SubS); + (module Ddl.Sublanguage.Make(S) : SubS); + (module Dml.Sublanguage.Make(S) : SubS); + (module Icl.Sublanguage.Make(S) : SubS); + (module Dcl.Sublanguage.Make(S) : SubS); + (module Scl.Sublanguage.Make(S) : SubS); + ] + Utilities.StringMap.empty + + let fmap f m = Result.bind m f + + let find_language tag = + Utilities.StringMap.find_opt tag sublanguages + |> Option.to_result ~none:(Error.UnrecognizedSublanguage tag) + + let execute storage db expr (module Language : SubS) = + Language.parse_sexp expr + |> fmap (Language.execute storage db) + |> Result.map_error (fun e -> Error.SublanguageError (Language.sexp_of_error e)) + + let execute_command storage db = function + | Sexplib.Sexp.(List [Atom tag; expr]) -> + find_language tag + |> fmap (execute storage db expr) + | s -> Error (Error.MalformedExpression s) + + let advance_head_branch storage new_hash = + match BranchOps.get_head storage with + | Ok (Some branch_name) -> + ignore (BranchOps.update_tip storage ~name:branch_name ~tip:new_hash) + | _ -> () + + let perform storage db_head old_db result = + match result with + | Sublanguage.Transition (new_db, _) -> + if Atomic.compare_and_set db_head old_db new_db + then (advance_head_branch storage new_db.hash; Ok result) + else Error (Error.Conflict { old_db; new_db }) + | _ -> Ok result + + let get_branch storage = + match BranchOps.get_head storage with Ok (Some name) -> name | _ -> "--" + + let current_limit = 16 (* TODO: make the limit per-connection? *) + + let materialize_generator gen limit = + let rec go gen pos acc count = + if count >= limit then (List.rev acc, true) + else + match gen (Some pos) with + | Generator.Done -> (List.rev acc, false) + | Generator.Error _ -> (List.rev acc, false) + | Generator.Value (t, next) -> ( + let mat = + match t with + | Tuple.Materialized m -> Some m + | Tuple.NonMaterialized _ -> None + in + match mat with + | None -> go next (pos + 1) acc count + | Some m -> go next (pos + 1) (m :: acc) (count + 1)) + in + go gen 0 [] 0 + + let materialize_relation storage (rel : Relation.t) limit = + let gen = AlgebraOps.to_generator storage rel in + materialize_generator gen limit + + let tuple_to_sexp (t : Tuple.materialized) = + let open Sexplib.Sexp in + List (Tuple.AttributeMap.bindings t.Tuple.attributes + |> List.map (fun (k, attr) -> List [ Atom k; Conventions.AbstractValue.sexp_of_t attr.Attribute.value ])) + + (* TODO: does our network protocol make sense? *) + let serialize storage (db : Management.Database.t) = + let open Sexplib.Sexp in + function + | Error e -> List [ Atom "error"; (Error.sexp_of_error e) ] + (* TODO: forsake the global cursor register in favour of per-connection bookkeeping *) + | Ok (Sublanguage.Cursor { cursor_id; rows; has_more }) -> + let row_sexps = List.map tuple_to_sexp rows in + let rows_sexp = List row_sexps in + List [ Atom "cursor"; + List [ Atom "id"; Atom cursor_id ]; + List [ Atom "rows"; rows_sexp ]; + List [ Atom "row_count"; Atom (string_of_int (List.length row_sexps)) ]; + List [ Atom "has_more"; Atom (string_of_bool has_more) ]; + List [ Atom "db_hash"; Atom db.hash ]; + List [ Atom "db_name"; Atom db.name ]; + List [ Atom "branch"; Atom (get_branch storage) ] ] + | Ok (Sublanguage.Query rel) -> + let tuples, truncated = materialize_relation storage rel current_limit in + let schema_sexp = List (List.map (fun (a, d) -> List [ Atom a; Atom d ]) rel.Relation.schema) in + let rows_sexp = List (List.map tuple_to_sexp tuples) in + List [ Atom "relation"; + List [ Atom "name"; Atom rel.Relation.name ]; + List [ Atom "schema"; schema_sexp ]; + List [ Atom "rows"; rows_sexp ]; + List [ Atom "row_count"; Atom (string_of_int (List.length tuples)) ]; + List [ Atom "truncated"; Atom (string_of_bool truncated) ]; + List [ Atom "db_hash"; Atom db.hash ]; + List [ Atom "db_name"; Atom db.name ]; + List [ Atom "branch"; Atom (get_branch storage) ] ] + | Ok (Sublanguage.Transition (new_db, message)) -> + List [ Atom "ok"; + List [ Atom "message"; Atom message ]; + List [ Atom "db_hash"; Atom new_db.hash ]; + List [ Atom "db_name"; Atom new_db.name ]; + (* TODO: `serialize' shouldn't need access to the storage layer *) + List [ Atom "branch"; Atom (get_branch storage) ] ] + + let print_with_time str = (* TODO: a proper logger *) + let now = Unix.gettimeofday () in + let tm = Unix.localtime now in + let orange = "\027[38;5;208m" in + let reset = "\027[0m" in + let formatted_time = + Printf.sprintf "%04d-%02d-%02d %02d:%02d:%02d" (tm.Unix.tm_year + 1900) + (tm.Unix.tm_mon + 1) tm.Unix.tm_mday tm.Unix.tm_hour tm.Unix.tm_min + tm.Unix.tm_sec + in + print_endline @@ Printf.sprintf "%s[%s]%s %s" orange formatted_time reset str + + let output_response output sexp = + let response = Sexplib.Sexp.to_string sexp in + output_string output response; + (* Some programs, like our current iteration of + * relational-explorer, expect server responses to always end + * with a newline. Although that is technically incorrect, + * acommodating for it doesn't bring us much harm; thus: + *) + output_string output "\n"; + flush output; + print_with_time response + + let handle_client connection db_head (storage : S.t) = + let input = T.input connection in + let output = T.output connection in + try + while true do + let db = Atomic.get db_head in + read_command input + |> fmap (execute_command storage db) + |> fmap (perform storage db_head db) + |> serialize storage db + |> output_response output + |> ignore + done + with + End_of_file -> () + (* TODO: use the logger instead *) + | e -> Printf.eprintf "Error handling connection: %s" (Printexc.to_string e) + + let spawn_handler db_head storage connection = + Stdlib.Domain.spawn (fun () -> handle_client connection db_head storage) + + let run transport db storage = + (* TODO: consider how to handle multiple databases within a + * single server (a catalog perhaps?) + *) + let db_head = Atomic.make db in + T.listen transport; + while true do + T.accept transport + |> spawn_handler db_head storage + |> ignore + done + end diff --git a/lib/management/branch.ml b/lib/management/branch.ml index ea15311..1d218b1 100644 --- a/lib/management/branch.ml +++ b/lib/management/branch.ml @@ -100,7 +100,7 @@ module Make (Storage : Physical.S with type error = string) = struct ~tree:None ~constraints:None ~cardinality:Conventions.Cardinality.ConstrainedFinite ~generator:(Some generator) - ~membership_criteria:(fun _ -> true) + ~membership_criteria:(fun _ _ -> true) ~provenance:(Relation.Provenance.Base Prelude.Catalog.branch_rel_name) ~lineage:(Relation.Lineage.Base Prelude.Catalog.branch_rel_name) @@ -127,7 +127,7 @@ module Make (Storage : Physical.S with type error = string) = struct ~tree:None ~constraints:None ~cardinality:Conventions.Cardinality.ConstrainedFinite ~generator:(Some generator) - ~membership_criteria:(fun _ -> true) + ~membership_criteria:(fun _ _ -> true) ~provenance:(Relation.Provenance.Base Prelude.Catalog.head_rel_name) ~lineage:(Relation.Lineage.Base Prelude.Catalog.head_rel_name) end diff --git a/lib/management/merge.ml b/lib/management/merge.ml index d30545a..a8860fe 100644 --- a/lib/management/merge.ml +++ b/lib/management/merge.ml @@ -10,6 +10,7 @@ type conflict = type strategy = PreferLeft | PreferRight | RevertToAncestor type merge_result = Clean of Database.t | Failed of conflict list +(* TODO: Remove this later. We already should have the interface on the manipulation *) (** Module type for the subset of Manipulation.Make we need. *) module type MANIPULATION = sig type storage diff --git a/lib/manipulation.ml b/lib/manipulation.ml index b44cc53..91cd35d 100644 --- a/lib/manipulation.ml +++ b/lib/manipulation.ml @@ -1,6 +1,6 @@ (** Database manipulation operations. - This module implements the core CRUD operations for the relational engine: + This module implements the core operations for the relational engine: - Database creation - Relation creation and deletion - Tuple insertion and deletion @@ -14,68 +14,69 @@ module RelationMap = Management.Database.RelationMap -(** Error types for manipulation operations *) -type error = - | RelationNotFound of string - | RelationAlreadyExists of string - | TupleNotFound of Conventions.Hash.t - | DuplicateTuple of Conventions.Hash.t - | ConstraintViolation of string - | StorageError of string - -let string_of_error = function - | RelationNotFound s -> "RelationNotFound: " ^ s - | RelationAlreadyExists s -> "RelationAlreadyExists: " ^ s - | TupleNotFound h -> "TupleNotFound: " ^ h - | DuplicateTuple h -> "DuplicateTuple: " ^ h - | ConstraintViolation s -> "ConstraintViolation: " ^ s - | StorageError s -> "StorageError: " ^ s - -let sexp_of_error e = - let open Sexplib.Sexp in - match e with - | RelationNotFound s -> List [ Atom "relation-not-found"; Atom s ] - | RelationAlreadyExists s -> List [ Atom "relation-already-exists"; Atom s ] - | TupleNotFound h -> List [ Atom "tuple-not-found"; Atom h ] - | DuplicateTuple h -> List [ Atom "duplicate-tuple"; Atom h ] - | ConstraintViolation s -> List [ Atom "constraint-violation"; Atom s ] - | StorageError s -> List [ Atom "storage-error"; Atom s ] - -type 'a result = ('a, error) Result.t -(** Result type for operations *) - -module Schema = Schema -(** Re-export Schema for convenience *) - -(** Re-export hashing functions for convenience *) -let hash_tuple = Hashing.hash_tuple - -(** Build base provenance for a schema *) -let build_base_provenance _schema relation_name = - Relation.Provenance.Base relation_name - -(** Build membership criteria from schema (validates type membership) *) -let build_membership_criteria _schema : Tuple.t -> bool = - (* For now, accept all tuples. Full implementation would check domain membership. *) - fun _ -> - true - -(** Mutation direction for cascade constraint checking *) -type mutation_kind = Insert | Delete - -type mutation_context = { - dep_rel : string; - transition : (string * Conventions.AbstractValue.t) list; - kind : mutation_kind; -} -(** Context for a single tuple mutation, used by cascade checking *) +(** Build membership criteria from schema (validates type membership). + Returns a function that takes a tree-lookup function and a tuple, checking + schema conformance and tree membership at call time. *) +let build_membership_criteria ~name schema : + (string -> Merkle.t option) -> Tuple.t -> bool = + let value_conforms_to_domain domain_name (value : Conventions.AbstractValue.t) + = + match domain_name with + | "integer" -> Obj.is_int value + | "natural" -> Obj.is_int value && (Obj.obj value : int) >= 0 + | "string" -> Obj.tag value = Obj.string_tag + | _ -> + (* Unknown domain: could be a relation-typed attribute (nested tuple) + or a user-defined domain. Accept for now; full validation would + require looking up the target relation's schema. *) + true + in + let n_expected = List.length schema in + fun tree_of tuple -> + match tuple with + | Tuple.NonMaterialized nm -> ( + match tree_of name with + | None -> false + | Some t -> + Merkle.member nm.hash t + && Tuple.AttributeMap.cardinal nm.attributes = n_expected + && List.for_all + (fun (attr_name, _) -> + Tuple.AttributeMap.mem attr_name nm.attributes) + schema) + | Tuple.Materialized m -> + Tuple.AttributeMap.cardinal m.attributes = n_expected + && List.for_all + (fun (attr_name, domain_name) -> + match Tuple.AttributeMap.find_opt attr_name m.attributes with + | None -> false + | Some attr -> + value_conforms_to_domain domain_name attr.Attribute.value) + schema + +let tree_of_db (db : Management.Database.t) name = + Option.bind (Management.Database.get_relation db name) (fun r -> r.Relation.tree) + +module Constraint = struct + include Constraint + + (** Mutation direction for cascade constraint checking *) + type mutation = Insert | Delete + + type mutation_context = { + target_relation : string; + transition : (string * Conventions.AbstractValue.t) list; + kind : mutation; + } + (** Context for a single tuple mutation, used by cascade checking *) +end (** Functor to create manipulation operations with a storage backend *) module Make (Storage : Management.Physical.S) = struct type storage = Storage.t - type nonrec error = error + type error = Error.t - let of_string_error s = StorageError s + let of_string_error s = Error.StorageError s let ( let* ) = Result.bind let fold_result (f : 'a -> 'b -> ('a, error) Result.t) (init : 'a) @@ -86,16 +87,40 @@ module Make (Storage : Management.Physical.S) = struct f a x) (Ok init) xs - (* Constraint evaluation context *) - (** Build an eval_context closed over a specific (storage, db) snapshot. All relation lookups resolve against this exact database version. *) let build_eval_context (storage : storage) (db : Management.Database.t) : Constraint.eval_context = + let load_value attr_hash = + match Storage.load_raw storage attr_hash with + | Error _ | Ok None -> None + | Ok (Some vbytes) -> + Some (Marshal.from_bytes vbytes 0 : Conventions.AbstractValue.t) + in + let rec load_attrs acc = function + | [] -> Some (List.rev acc) + | (name, attr_hash) :: rest -> ( + match load_value attr_hash with + | None -> None + | Some value -> load_attrs ((name, value) :: acc) rest) + in + let load_tuple hash = + match Storage.load_raw storage hash with + | Error _ | Ok None -> None + | Ok (Some bytes) -> + load_attrs [] (Storable.Tuple.of_bytes bytes).Storable.Tuple.attributes + in + let rec load_all acc = function + | [] -> Some (List.rev acc) + | h :: rest -> ( + match load_tuple h with + | None -> None + | Some pairs -> load_all (pairs :: acc) rest) + in let check_membership rel_name bound_pairs = match Management.Database.get_relation db rel_name with | None -> false - | Some rel -> ( + | Some rel -> let attributes = List.fold_left (fun acc (k, v) -> @@ -105,38 +130,7 @@ module Make (Storage : Management.Physical.S) = struct let tuple = Tuple.Materialized { Tuple.relation = rel_name; attributes } in - if not (rel.membership_criteria tuple) then false - else - (* For stored relations, check if a matching tuple actually exists *) - match rel.tree with - | None -> - (* No tree = ephemeral/generator relation, membership_criteria suffices *) - true - | Some tree -> - let hashes = Merkle.keys tree in - List.exists - (fun h -> - match Storage.load_raw storage h with - | Error _ | Ok None -> false - | Ok (Some bytes) -> - let stored = Storable.Tuple.of_bytes bytes in - (* Check if all bound pairs match the stored tuple's attributes *) - List.for_all - (fun (bk, bv) -> - List.exists - (fun (sk, sh) -> - sk = bk - && - match Storage.load_raw storage sh with - | Error _ | Ok None -> false - | Ok (Some vbytes) -> - let sv : Conventions.AbstractValue.t = - Marshal.from_bytes vbytes 0 - in - Stdlib.( = ) sv bv) - stored.Storable.Tuple.attributes) - bound_pairs) - hashes) + rel.membership_criteria (tree_of_db db) tuple in let iterate_finite rel_name = match Management.Database.get_relation db rel_name with @@ -144,39 +138,11 @@ module Make (Storage : Management.Physical.S) = struct | Some rel -> ( match rel.cardinality with | Conventions.Cardinality.Finite _ - | Conventions.Cardinality.ConstrainedFinite -> ( - match rel.tree with - | None -> Some [] - | Some tree -> - let hashes = Merkle.keys tree in - let rec load_all acc = function - | [] -> Some (List.rev acc) - | h :: rest -> ( - match Storage.load_raw storage h with - | Error _ -> None - | Ok None -> None - | Ok (Some bytes) -> ( - let stored = Storable.Tuple.of_bytes bytes in - let rec load_attrs attr_acc = function - | [] -> Some (List.rev attr_acc) - | (name, attr_hash) :: attr_rest -> ( - match Storage.load_raw storage attr_hash with - | Error _ | Ok None -> None - | Ok (Some vbytes) -> - let value : Conventions.AbstractValue.t = - Marshal.from_bytes vbytes 0 - in - load_attrs - ((name, value) :: attr_acc) - attr_rest) - in - match - load_attrs [] stored.Storable.Tuple.attributes - with - | None -> None - | Some pairs -> load_all (pairs :: acc) rest)) - in - load_all [] hashes) + | Conventions.Cardinality.ConstrainedFinite -> + let hashes = + match rel.tree with None -> [] | Some tree -> Merkle.keys tree + in + load_all [] hashes | _ -> None) in { Constraint.check_membership; iterate_finite } @@ -187,7 +153,7 @@ module Make (Storage : Management.Physical.S) = struct let store_relation (storage : storage) (relation : Relation.t) : (unit, error) Result.t = match relation.hash with - | None -> Error (StorageError "Relation has no hash") + | None -> Error (Error.StorageError "Relation has no hash") | Some rel_hash -> ( let tree_keys = match relation.tree with None -> [] | Some tree -> Merkle.keys tree @@ -203,7 +169,7 @@ module Make (Storage : Management.Physical.S) = struct match Storage.store_raw storage rel_hash (Storable.Relation.to_bytes stored) with - | Error _ -> Error (StorageError "Failed to store relation") + | Error _ -> Error (Error.StorageError "Failed to store relation") | Ok () -> Ok ()) (** Store a database state to storage *) @@ -232,14 +198,14 @@ module Make (Storage : Management.Physical.S) = struct match Storage.store_raw storage db.hash (Storable.Database.to_bytes stored) with - | Error _ -> Error (StorageError "Failed to store database") + | Error _ -> Error (Error.StorageError "Failed to store database") | Ok () -> Ok () (** Load a relation state from storage by hash *) let load_relation (storage : storage) (rel_hash : Conventions.Hash.t) : (Relation.t option, error) Result.t = match Storage.load_raw storage rel_hash with - | Error _ -> Error (StorageError "Failed to load relation") + | Error _ -> Error (Error.StorageError "Failed to load relation") | Ok None -> Ok None | Ok (Some bytes) -> let stored = Storable.Relation.of_bytes bytes in @@ -252,8 +218,8 @@ module Make (Storage : Management.Physical.S) = struct Relation.make ~hash:(Some rel_hash) ~name:stored.name ~schema:stored.schema ~tree:(Some tree) ~constraints:None ~cardinality:stored.cardinality ~generator:None - ~membership_criteria:(build_membership_criteria stored.schema) - ~provenance:(build_base_provenance stored.schema stored.name) + ~membership_criteria:(build_membership_criteria ~name:stored.name stored.schema) + ~provenance:(Relation.Provenance.Base stored.name) ~lineage:(Relation.Lineage.Base stored.name) in Ok (Some relation) @@ -264,7 +230,7 @@ module Make (Storage : Management.Physical.S) = struct let load_database (storage : storage) (db_hash : Conventions.Hash.t) : (Management.Database.t option, error) Result.t = match Storage.load_raw storage db_hash with - | Error _ -> Error (StorageError "Failed to load database") + | Error _ -> Error (Error.StorageError "Failed to load database") | Ok None -> Ok None | Ok (Some bytes) -> let stored = Storable.Database.of_bytes bytes in @@ -278,7 +244,8 @@ module Make (Storage : Management.Physical.S) = struct | (name, rel_hash) :: rest -> ( let* rel_opt = load_relation storage rel_hash in match rel_opt with - | None -> Error (StorageError ("Relation not found: " ^ name)) + | None -> + Error (Error.StorageError ("Relation not found: " ^ name)) | Some rel -> load_relations ((name, rel) :: acc) rest) in let* relation_list = load_relations [] stored.relations in @@ -307,9 +274,9 @@ module Make (Storage : Management.Physical.S) = struct (** Create a new empty relation with the given schema *) let create_relation_raw (storage : storage) (db : Management.Database.t) ~(name : string) ~(schema : Schema.t) : - (Management.Database.t * Relation.t) result = + (Management.Database.t * Relation.t, Error.t) Result.t = if Management.Database.has_relation db name then - Error (RelationAlreadyExists name) + Error (Error.RelationAlreadyExists name) else let tree = Merkle.empty in let relation_hash = Hashing.hash_relation ~name ~schema ~tree in @@ -317,8 +284,8 @@ module Make (Storage : Management.Physical.S) = struct Relation.make ~hash:(Some relation_hash) ~name ~schema ~tree:(Some tree) ~constraints:None ~cardinality:(Conventions.Cardinality.Finite 0) ~generator:None - ~membership_criteria:(build_membership_criteria schema) - ~provenance:(build_base_provenance schema name) + ~membership_criteria:(build_membership_criteria ~name schema) + ~provenance:(Relation.Provenance.Base name) ~lineage:(Relation.Lineage.Base name) in let new_db = Management.Database.add_relation db ~relation in @@ -342,7 +309,7 @@ module Make (Storage : Management.Physical.S) = struct in match Storage.store_raw storage value_hash value_bytes with | Error _ -> - Error (StorageError ("Failed to store attribute: " ^ name)) + Error (Error.StorageError ("Failed to store attribute: " ^ name)) | Ok () -> store_all ((name, value_hash) :: acc) rest) in store_all [] attrs @@ -351,7 +318,7 @@ module Make (Storage : Management.Physical.S) = struct let load_tuple (storage : storage) (tuple_hash : Conventions.Hash.t) : (Tuple.materialized option, error) Result.t = match Storage.load_raw storage tuple_hash with - | Error _ -> Error (StorageError "Failed to load tuple") + | Error _ -> Error (Error.StorageError "Failed to load tuple") | Ok None -> Ok None | Ok (Some tuple_bytes) -> let stored = Storable.Tuple.of_bytes tuple_bytes in @@ -360,8 +327,10 @@ module Make (Storage : Management.Physical.S) = struct | (name, attr_hash) :: rest -> ( match Storage.load_raw storage attr_hash with | Error _ -> - Error (StorageError ("Failed to load attribute: " ^ name)) - | Ok None -> Error (StorageError ("Attribute not found: " ^ name)) + Error + (Error.StorageError ("Failed to load attribute: " ^ name)) + | Ok None -> + Error (Error.StorageError ("Attribute not found: " ^ name)) | Ok (Some value_bytes) -> let value : Conventions.AbstractValue.t = Marshal.from_bytes value_bytes 0 @@ -385,7 +354,7 @@ module Make (Storage : Management.Physical.S) = struct | hash :: rest -> ( let* tup_opt = load_tuple storage hash in match tup_opt with - | None -> Error (TupleNotFound hash) + | None -> Error (Error.TupleNotFound hash) | Some tuple -> load_all (tuple :: acc) rest) in load_all [] hashes @@ -396,7 +365,8 @@ module Make (Storage : Management.Physical.S) = struct | Some tup -> on_some tup | None -> Error - (StorageError ("Tuple missing for referenced hash: " ^ tuple_hash)) + (Error.StorageError + ("Tuple missing for referenced hash: " ^ tuple_hash)) let transition_attrs (tuple : Tuple.materialized) = Tuple.AttributeMap.bindings tuple.attributes @@ -404,7 +374,7 @@ module Make (Storage : Management.Physical.S) = struct let ensure_relation_exists (db : Management.Database.t) name = if Management.Database.has_relation db name then Ok () - else Error (RelationNotFound name) + else Error (Error.RelationNotFound name) let bump_cardinality delta = function | Conventions.Cardinality.Finite n -> @@ -414,7 +384,13 @@ module Make (Storage : Management.Physical.S) = struct let rehashed_relation_with_tree ~name ~schema (relation : Relation.t) tree cardinality = let hash = Hashing.hash_relation ~name ~schema ~tree in - { relation with hash = Some hash; tree = Some tree; cardinality } + { + relation with + hash = Some hash; + tree = Some tree; + cardinality; + membership_criteria = build_membership_criteria ~name schema; + } let validate_tuple_constraints (storage : storage) (db : Management.Database.t) (relation : Relation.t) @@ -428,14 +404,15 @@ module Make (Storage : Management.Physical.S) = struct in match eval_result with | Ok true -> Ok () - | Ok false -> Error (ConstraintViolation "Constraint not satisfied") + | Ok false -> Error (Error.ConstraintViolation "Constraint not satisfied") | Error (Constraint.ConstraintFailures failures) -> let msg = String.concat "; " (List.map (fun (n, _) -> "constraint " ^ n ^ " violated") failures) in - Error (ConstraintViolation msg) - | Error _ -> Error (ConstraintViolation "Constraint evaluation failed") + Error (Error.ConstraintViolation msg) + | Error _ -> + Error (Error.ConstraintViolation "Constraint evaluation failed") let store_tuple_payload storage tuple_hash (tuple : Tuple.materialized) = let* attr_hashes = store_attributes storage tuple in @@ -444,23 +421,25 @@ module Make (Storage : Management.Physical.S) = struct in let tuple_bytes = Storable.Tuple.to_bytes stored_tuple in match Storage.store_raw storage tuple_hash tuple_bytes with - | Error _ -> Error (StorageError "Failed to store tuple") + | Error _ -> Error (Error.StorageError "Failed to store tuple") | Ok () -> Ok () (* Cascade constraint helpers *) let polarity_triggered_by kind pol = match (kind, pol) with - | Insert, (Constraint.Negative | Constraint.Both) -> true - | Delete, (Constraint.Positive | Constraint.Both) -> true + | Constraint.Insert, (Constraint.Negative | Constraint.Both) -> true + | Constraint.Delete, (Constraint.Positive | Constraint.Both) -> true | _ -> false - let cascade_violation_msg kind dep_rel cname constrained_name = + let cascade_violation_msg kind target_relation cname constrained_name = let verb = - match kind with Insert -> "inserting into" | Delete -> "deleting from" + match kind with + | Constraint.Insert -> "inserting into" + | Constraint.Delete -> "deleting from" in - Printf.sprintf "cascade: %s %s violates constraint %s on %s" verb dep_rel - cname constrained_name + Printf.sprintf "cascade: %s %s violates constraint %s on %s" verb + target_relation cname constrained_name let is_deferred (db : Management.Database.t) rel_name cname = List.exists @@ -477,13 +456,15 @@ module Make (Storage : Management.Physical.S) = struct | None -> false) filter - let recheck_one storage ctx mctx cname cbody constrained_name filter - (h : Conventions.Hash.t) : unit result = + let recheck_one storage ctx (mctx : Constraint.mutation_context) cname cbody + constrained_name filter (h : Conventions.Hash.t) : (unit, error) Result.t + = with_loaded_tuple storage h ~on_some:(fun tup -> if not (needs_recheck filter tup) then Ok () else let cbody' = - Constraint.substitute_transition cbody mctx.dep_rel mctx.transition + Constraint.substitute_transition cbody mctx.target_relation + mctx.transition in match Constraint.evaluate_first_failure ctx tup [ (cname, cbody') ] @@ -491,21 +472,21 @@ module Make (Storage : Management.Physical.S) = struct | Ok true -> Ok () | Ok false | Error _ -> Error - (ConstraintViolation - (cascade_violation_msg mctx.kind mctx.dep_rel cname + (Error.ConstraintViolation + (cascade_violation_msg mctx.kind mctx.target_relation cname constrained_name))) - let check_one_constraint storage ctx db mctx constrained_name constrained_rel - (cname, cbody) : unit result = + let check_one_constraint storage ctx db (mctx : Constraint.mutation_context) + constrained_name constrained_rel (cname, cbody) : (unit, error) Result.t = if is_deferred db constrained_name cname then Ok () else let pols = Constraint.polarity_of cbody in - match Constraint.polarity_find mctx.dep_rel pols with + match Constraint.polarity_find mctx.target_relation pols with | None -> Ok () | Some pol when not (polarity_triggered_by mctx.kind pol) -> Ok () | Some _ -> let filter = - Constraint.focused_filter cbody mctx.dep_rel mctx.transition + Constraint.focused_filter cbody mctx.target_relation mctx.transition in let hashes = match constrained_rel.Relation.tree with @@ -517,13 +498,14 @@ module Make (Storage : Management.Physical.S) = struct recheck_one storage ctx mctx cname cbody constrained_name filter h) () hashes - (** Scan every relation in [db] for constraints that reference [mctx.dep_rel] - with a polarity matched by the mutation kind, then re-evaluate those - constraints against the affected tuples in each constrained relation. + (** Scan every relation in [db] for constraints that reference + [mctx.target_relation] with a polarity matched by the mutation kind, then + re-evaluate those constraints against the affected tuples in each + constrained relation. Returns [Ok ()] when all cascade checks pass. *) let check_cascade_constraints (storage : storage) (db : Management.Database.t) - (mctx : mutation_context) : unit result = + (mctx : Constraint.mutation_context) = let ctx = build_eval_context storage db in Management.Database.RelationMap.fold (fun constrained_name constrained_rel acc -> @@ -541,11 +523,13 @@ module Make (Storage : Management.Physical.S) = struct (** Insert a tuple into a relation, storing it in the backend *) let create_tuple (storage : storage) (db : Management.Database.t) (relation : Relation.t) (tuple : Tuple.materialized) : - (Management.Database.t * Relation.t * Conventions.Hash.t) result = + (Management.Database.t * Relation.t * Conventions.Hash.t, error) Result.t + = let name = relation.name in let* () = ensure_relation_exists db name in - if not (relation.membership_criteria (Tuple.Materialized tuple)) then - Error (ConstraintViolation "Tuple does not satisfy membership criteria") + if not (relation.membership_criteria (tree_of_db db) (Tuple.Materialized tuple)) then + Error + (Error.ConstraintViolation "Tuple does not satisfy membership criteria") else let* () = validate_tuple_constraints storage db relation tuple in let tuple_hash = Hashing.hash_tuple tuple in @@ -553,7 +537,7 @@ module Make (Storage : Management.Physical.S) = struct match relation.tree with Some t -> t | None -> Merkle.empty in if Merkle.member tuple_hash current_tree then - Error (DuplicateTuple tuple_hash) + Error (Error.DuplicateTuple tuple_hash) else let* () = store_tuple_payload storage tuple_hash tuple in let new_tree = Merkle.insert tuple_hash current_tree in @@ -568,9 +552,9 @@ module Make (Storage : Management.Physical.S) = struct let* () = check_cascade_constraints storage new_db { - dep_rel = name; + target_relation = name; transition = transition_attrs tuple; - kind = Insert; + kind = Constraint.Insert; } in let* () = store_relation storage new_relation in @@ -580,7 +564,9 @@ module Make (Storage : Management.Physical.S) = struct (** Insert multiple tuples into a relation *) let create_tuples (storage : storage) (db : Management.Database.t) (relation : Relation.t) (tuples : Tuple.materialized list) : - (Management.Database.t * Relation.t * Conventions.Hash.t list) result = + ( Management.Database.t * Relation.t * Conventions.Hash.t list, + error ) + Result.t = let rec insert_all db rel hashes = function | [] -> Ok (db, rel, List.rev hashes) | tuple :: rest -> @@ -592,16 +578,16 @@ module Make (Storage : Management.Physical.S) = struct (** Remove a tuple from a relation by its hash *) let retract_tuple (storage : storage) (db : Management.Database.t) (relation : Relation.t) ~(tuple_hash : Conventions.Hash.t) : - (Management.Database.t * Relation.t) result = + (Management.Database.t * Relation.t, error) Result.t = let name = relation.name in let* () = ensure_relation_exists db name in let* current_tree = match relation.tree with - | None -> Error (TupleNotFound tuple_hash) + | None -> Error (Error.TupleNotFound tuple_hash) | Some tree -> Ok tree in if not (Merkle.member tuple_hash current_tree) then - Error (TupleNotFound tuple_hash) + Error (Error.TupleNotFound tuple_hash) else (* Note: We don't delete from storage - content-addressed storage is append-only. The tuple data remains for historical queries / garbage collection later. *) @@ -618,9 +604,9 @@ module Make (Storage : Management.Physical.S) = struct with_loaded_tuple storage tuple_hash ~on_some:(fun deleted_tuple -> check_cascade_constraints storage new_db { - dep_rel = name; + target_relation = name; transition = transition_attrs deleted_tuple; - kind = Delete; + kind = Constraint.Delete; }) in let* () = store_relation storage new_relation in @@ -724,7 +710,8 @@ module Make (Storage : Management.Physical.S) = struct let get_required (db : Management.Database.t) rel_name = match Management.Database.get_relation db rel_name with | Some rel -> Ok rel - | None -> Error (StorageError ("Catalog relation missing: " ^ rel_name)) + | None -> + Error (Error.StorageError ("Catalog relation missing: " ^ rel_name)) in let* relation_rel = get_required db Prelude.Catalog.relation_rel_name in let* db, _, _ = @@ -804,9 +791,9 @@ module Make (Storage : Management.Physical.S) = struct [sakura:relation] and [sakura:attribute] in the system catalog. *) let create_relation (storage : storage) (db : Management.Database.t) ~(name : string) ~(schema : Schema.t) : - (Management.Database.t * Relation.t) result = + (Management.Database.t * Relation.t, error) Result.t = if Management.Database.has_relation db name then - Error (RelationAlreadyExists name) + Error (Error.RelationAlreadyExists name) else let tree = Merkle.empty in let relation_hash = Hashing.hash_relation ~name ~schema ~tree in @@ -814,8 +801,8 @@ module Make (Storage : Management.Physical.S) = struct Relation.make ~hash:(Some relation_hash) ~name ~schema ~tree:(Some tree) ~constraints:None ~cardinality:(Conventions.Cardinality.Finite 0) ~generator:None - ~membership_criteria:(build_membership_criteria schema) - ~provenance:(build_base_provenance schema name) + ~membership_criteria:(build_membership_criteria ~name schema) + ~provenance:(Relation.Provenance.Base name) ~lineage:(Relation.Lineage.Base name) in let new_db = Management.Database.add_relation db ~relation in @@ -827,11 +814,11 @@ module Make (Storage : Management.Physical.S) = struct (** Create an immutable/generator-based relation *) let create_immutable_relation (storage : storage) (db : Management.Database.t) ~(name : string) ~(schema : Schema.t) ~(generator : Generator.t) - ~(membership_criteria : Tuple.t -> bool) + ~(membership_criteria : (string -> Merkle.t option) -> Tuple.t -> bool) ~(cardinality : Conventions.Cardinality.t) : - (Management.Database.t * Relation.t) result = + (Management.Database.t * Relation.t, error) Result.t = if Management.Database.has_relation db name then - Error (RelationAlreadyExists name) + Error (Error.RelationAlreadyExists name) else let relation_hash = Hashing.hash_relation ~name ~schema ~tree:Merkle.empty @@ -840,8 +827,7 @@ module Make (Storage : Management.Physical.S) = struct Relation.make ~hash:(Some relation_hash) ~name ~schema ~tree:None (* No tree for generator-based relations *) ~constraints:None ~cardinality ~generator:(Some generator) - ~membership_criteria - ~provenance:(build_base_provenance schema name) + ~membership_criteria ~provenance:(Relation.Provenance.Base name) ~lineage:(Relation.Lineage.Base name) in let new_db = Management.Database.add_relation db ~relation in @@ -853,9 +839,9 @@ module Make (Storage : Management.Physical.S) = struct (** Remove a relation from the database. Also removes its entries from [sakura:relation] and [sakura:attribute]. *) let retract_relation (storage : storage) (db : Management.Database.t) - ~(name : string) : Management.Database.t result = + ~(name : string) : (Management.Database.t, error) Result.t = match Management.Database.get_relation db name with - | None -> Error (RelationNotFound name) + | None -> Error (Error.RelationNotFound name) | Some relation -> let new_db = Management.Database.remove_relation db ~name in (* Update catalog on the db state that no longer contains the retracted relation *) @@ -865,10 +851,11 @@ module Make (Storage : Management.Physical.S) = struct (** Clear all tuples from a relation (truncate) *) let clear_relation (storage : storage) (db : Management.Database.t) - (relation : Relation.t) : (Management.Database.t * Relation.t) result = + (relation : Relation.t) : + (Management.Database.t * Relation.t, error) Result.t = let name = relation.name in match Management.Database.get_relation db name with - | None -> Error (RelationNotFound name) + | None -> Error (Error.RelationNotFound name) | Some _ -> let new_tree = Merkle.empty in let new_hash = @@ -880,6 +867,8 @@ module Make (Storage : Management.Physical.S) = struct hash = Some new_hash; tree = Some new_tree; cardinality = Conventions.Cardinality.Finite 0; + membership_criteria = + build_membership_criteria ~name relation.schema; } in let new_db = @@ -894,9 +883,9 @@ module Make (Storage : Management.Physical.S) = struct let update_relation_constraints (storage : storage) (db : Management.Database.t) ~(relation_name : string) ~(constraints : Relation.RelationConstraint.t) : - (Management.Database.t * Relation.t) result = + (Management.Database.t * Relation.t, error) Result.t = match Management.Database.get_relation db relation_name with - | None -> Error (RelationNotFound relation_name) + | None -> Error (Error.RelationNotFound relation_name) | Some relation -> let merged_constraints = match relation.constraints with @@ -1012,7 +1001,7 @@ module Make (Storage : Management.Physical.S) = struct | Ok true -> Ok () | Ok false | Error _ -> Error - (ConstraintViolation + (Error.ConstraintViolation (Printf.sprintf "deferred constraint %s on %s violated" entry.constraint_name entry.relation_name)))) @@ -1037,6 +1026,3 @@ module Make (Storage : Management.Physical.S) = struct let* () = store_database storage db' in Ok db' end - -module Memory = Make (Management.Physical.Memory) -(** Default instance using in-memory storage for convenience/testing *) diff --git a/lib/manipulation.mli b/lib/manipulation.mli deleted file mode 100644 index 2a52b5c..0000000 --- a/lib/manipulation.mli +++ /dev/null @@ -1,137 +0,0 @@ -type error = - | RelationNotFound of string - | RelationAlreadyExists of string - | TupleNotFound of Conventions.Hash.t - | DuplicateTuple of Conventions.Hash.t - | ConstraintViolation of string - | StorageError of string - -val string_of_error : error -> string -val sexp_of_error : error -> Sexplib.Sexp.t - -type 'a result = ('a, error) Result.t - -module Schema = Schema - -val hash_tuple : Tuple.materialized -> Conventions.Hash.t - -module Make (Storage : Management.Physical.S) : sig - type storage = Storage.t - type nonrec error = error - - val of_string_error : string -> error - - val store_database : - storage -> Management.Database.t -> (unit, error) Result.t - - val load_relation : - storage -> Conventions.Hash.t -> (Relation.t option, error) Result.t - - val load_database : - storage -> - Conventions.Hash.t -> - (Management.Database.t option, error) Result.t - - val load_tuple : - storage -> Conventions.Hash.t -> (Tuple.materialized option, error) Result.t - - val load_tuples : - storage -> - Conventions.Hash.t list -> - (Tuple.materialized list, error) Result.t - - val create_tuple : - storage -> - Management.Database.t -> - Relation.t -> - Tuple.materialized -> - (Management.Database.t * Relation.t * Conventions.Hash.t, error) Result.t - - val create_tuples : - storage -> - Management.Database.t -> - Relation.t -> - Tuple.materialized list -> - ( Management.Database.t * Relation.t * Conventions.Hash.t list, - error ) - Result.t - - val retract_tuple : - storage -> - Management.Database.t -> - Relation.t -> - tuple_hash:Conventions.Hash.t -> - (Management.Database.t * Relation.t, error) Result.t - - val register_domain : - storage -> - Management.Database.t -> - Domain.t -> - (Management.Database.t, error) Result.t - - val create_database : - storage -> name:string -> (Management.Database.t, error) Result.t - - val database_history : Management.Database.t -> Conventions.Hash.t list - - val create_relation : - storage -> - Management.Database.t -> - name:string -> - schema:Schema.t -> - (Management.Database.t * Relation.t, error) Result.t - - val create_immutable_relation : - storage -> - Management.Database.t -> - name:string -> - schema:Schema.t -> - generator:Generator.t -> - membership_criteria:(Tuple.t -> bool) -> - cardinality:Conventions.Cardinality.t -> - (Management.Database.t * Relation.t, error) Result.t - - val retract_relation : - storage -> - Management.Database.t -> - name:string -> - Management.Database.t result - - val clear_relation : - storage -> - Management.Database.t -> - Relation.t -> - (Management.Database.t * Relation.t, error) Result.t - - val register_constraint : - storage -> - Management.Database.t -> - constraint_name:string -> - relation_name:string -> - body:Constraint.t -> - (Management.Database.t, error) Result.t - - val tuple_hashes : Relation.t -> Conventions.Hash.t list - val get_relation : Management.Database.t -> name:string -> Relation.t option - val tuple_count : Relation.t -> int - val tuple_exists : Relation.t -> Conventions.Hash.t -> bool - - val attach_constraint : - storage -> - Management.Database.t -> - constraint_name:string -> - relation_name:string -> - body:Constraint.t -> - timing:Constraint.timing -> - (Management.Database.t, error) Result.t - - val check_deferred_constraints : - storage -> Management.Database.t -> (unit, error) Result.t - - val commit : - storage -> Management.Database.t -> (Management.Database.t, error) Result.t -end - -module Memory : sig - include module type of Make (Management.Physical.Memory) -end diff --git a/lib/prelude/standard.ml b/lib/prelude/standard.ml index 6bdab0c..6822f19 100644 --- a/lib/prelude/standard.ml +++ b/lib/prelude/standard.ml @@ -15,7 +15,8 @@ let make_comparison ~name ~domain_name ~pred ~cardinality ~generator = |> Schema.add "left" domain_name |> Schema.add "right" domain_name in - let membership_criteria : Tuple.t -> bool = function + let membership_criteria : (string -> Merkle.t option) -> Tuple.t -> bool = + fun _tree_of -> function | Tuple.Materialized m -> ( match ( Tuple.AttributeMap.find_opt "left" m.attributes, @@ -176,7 +177,8 @@ let plus_natural : Relation.t = (Tuple.make_materialized ~relation:"plus" ~attributes, generator) | None -> Error "Cannot enumerate plus randomly." in - let membership_criteria : Tuple.t -> bool = function + let membership_criteria : (string -> Merkle.t option) -> Tuple.t -> bool = + fun _tree_of -> function | Tuple.Materialized m -> ( match ( Tuple.AttributeMap.find_opt "a" m.attributes, @@ -214,7 +216,8 @@ let times_natural : Relation.t = (Tuple.make_materialized ~relation:"times" ~attributes, generator) | None -> Error "Cannot enumerate times randomly." in - let membership_criteria : Tuple.t -> bool = function + let membership_criteria : (string -> Merkle.t option) -> Tuple.t -> bool = + fun _tree_of -> function | Tuple.Materialized m -> ( match ( Tuple.AttributeMap.find_opt "a" m.attributes, @@ -253,7 +256,8 @@ let minus_natural : Relation.t = (Tuple.make_materialized ~relation:"minus" ~attributes, generator) | None -> Error "Cannot enumerate minus randomly." in - let membership_criteria : Tuple.t -> bool = function + let membership_criteria : (string -> Merkle.t option) -> Tuple.t -> bool = + fun _tree_of -> function | Tuple.Materialized m -> ( match ( Tuple.AttributeMap.find_opt "a" m.attributes, @@ -301,7 +305,8 @@ let divide_natural : Relation.t = (Tuple.make_materialized ~relation:"divide" ~attributes, generator) | None -> Error "Cannot enumerate divide randomly." in - let membership_criteria : Tuple.t -> bool = function + let membership_criteria : (string -> Merkle.t option) -> Tuple.t -> bool = + fun _tree_of -> function | Tuple.Materialized m -> ( match ( Tuple.AttributeMap.find_opt "a" m.attributes, diff --git a/lib/relation.ml b/lib/relation.ml index 285bcef..0855b57 100644 --- a/lib/relation.ml +++ b/lib/relation.ml @@ -36,7 +36,7 @@ type t = { constraints : RelationConstraint.t option; cardinality : Conventions.Cardinality.t; generator : Generator.t option; - membership_criteria : Tuple.t -> bool; + membership_criteria : (string -> Tree.t option) -> Tuple.t -> bool; provenance : Provenance.t; lineage : Lineage.t; } diff --git a/lib/transport.ml b/lib/transport.ml new file mode 100644 index 0000000..5130a2d --- /dev/null +++ b/lib/transport.ml @@ -0,0 +1,41 @@ +module type TRANSPORT = sig + type t + type connection + type options + + val create : options -> t + + val listen : t -> unit + val accept : t -> connection + + val input : connection -> in_channel + val output : connection -> out_channel +end + +(* TODO: move this into TCP once we have a proper configuration system *) +type tcp_options = { + address_family : Unix.socket_domain; + address : Unix.sockaddr +} + +module TCP : (TRANSPORT with type options = tcp_options) = struct + type t = Unix.file_descr + type connection = Unix.file_descr + type options = tcp_options + + let create { address_family; address } = + let socket = Unix.socket address_family Unix.SOCK_STREAM 0 in + Unix.setsockopt socket Unix.SO_REUSEADDR true; + Unix.setsockopt socket Unix.SO_REUSEPORT true; + Unix.bind socket address; + socket + + let listen fd = Unix.listen fd 5 + + let accept fd = + let (conn, _) = Unix.accept fd in + conn + + let input = Unix.in_channel_of_descr + let output = Unix.out_channel_of_descr +end -- cgit v1.2.3