diff --git a/src/anf.ml b/src/anf.ml index f340a40..1fefdcf 100644 --- a/src/anf.ml +++ b/src/anf.ml @@ -29,6 +29,8 @@ and desc = type program = { ap_input : Ident.t; ap_input_element : Types.t; + ap_types : (int * Types.t) list; + ap_records : Types.records; ap_query : Ident.t; ap_query_body : expr; ap_helpers : (Ident.t * Types.scheme * expr) list; @@ -38,11 +40,18 @@ let make a aty aspan = { a = a; aty = aty; aspan = aspan } let temp_counter = ref 0 -let reset () = temp_counter := 0 +let type_log : (int * Types.t) list ref = ref [] + +let reset () = + temp_counter := 0; + type_log := [] + +let record_type ident ty = type_log := (Ident.stamp ident, ty) :: !type_log let fresh_temp span ty = incr temp_counter; let ident = Ident.fresh (Printf.sprintf "t%d" !temp_counter) span in + record_type ident ty; (ident, ty) let rec bind (typed : Typed.expr) (k : expr -> expr) : expr = @@ -95,6 +104,7 @@ let rec bind (typed : Typed.expr) (k : expr -> expr) : expr = | Typed.TCount collection -> bind collection (fun collection_atom -> force (make (ACount (atom_of collection_atom)) ty span) k) | Typed.TLet (ident, bound, body) -> + record_type ident ty; k (make (ALet (ident, normalize bound, normalize body)) ty span) and atom_of expr = @@ -121,20 +131,31 @@ and normalize expr = bind expr (fun value -> value) and lambda_body expr = match expr.Typed.te with - | Typed.TLambda (parameter, body) -> (parameter, normalize body) + | Typed.TLambda (parameter, body) -> + (match Types.repr expr.Typed.ty with + | Types.TArrow (element, _) -> record_type parameter element + | _ -> ()); + (parameter, normalize body) | _ -> Diagnostic.error expr.Typed.tspan "internal error: expected a function literal after specialization" let program typed = + reset (); + let body = normalize typed.Typed.tp_query_body in + let helpers = + List.map + (fun helper -> (helper.Typed.th_ident, helper.Typed.th_scheme, normalize helper.Typed.th_body)) + typed.Typed.tp_helpers + in + record_type typed.Typed.tp_input (Types.TCollection typed.Typed.tp_input_element); { ap_input = typed.Typed.tp_input; ap_input_element = typed.Typed.tp_input_element; + ap_types = !type_log; + ap_records = typed.Typed.tp_records; ap_query = typed.Typed.tp_query; - ap_query_body = normalize typed.Typed.tp_query_body; - ap_helpers = - List.map - (fun helper -> (helper.Typed.th_ident, helper.Typed.th_scheme, normalize helper.Typed.th_body)) - typed.Typed.tp_helpers; + ap_query_body = body; + ap_helpers = helpers; } let atom_to_string atom = diff --git a/src/anf.mli b/src/anf.mli index 106e65b..eed695d 100644 --- a/src/anf.mli +++ b/src/anf.mli @@ -29,6 +29,8 @@ and desc = type program = { ap_input : Ident.t; ap_input_element : Types.t; + ap_types : (int * Types.t) list; + ap_records : Types.records; ap_query : Ident.t; ap_query_body : expr; ap_helpers : (Ident.t * Types.scheme * expr) list; diff --git a/src/change.ml b/src/change.ml index d7f2d9f..69766fc 100644 --- a/src/change.ml +++ b/src/change.ml @@ -287,7 +287,7 @@ let to_string change = let rec render change = match change with | CEmpty -> "empty" - | CInt delta -> Printf.sprintf "+%d" delta + | CInt delta -> Printf.sprintf "%+d" delta | CBool truth -> Printf.sprintf "bool %b" truth | CString text -> Printf.sprintf "string %S" text | CUnit -> "unit" diff --git a/src/graph.ml b/src/graph.ml index db342b0..f7d0e04 100644 --- a/src/graph.ml +++ b/src/graph.ml @@ -29,6 +29,9 @@ type plan = { pl_output : Types.t; pl_output_element : Types.t option; pl_consumers : int list array; + pl_types : (int * Types.t) list; + pl_records : Types.records; + pl_scalar_bindings : (int * int) list; } let node_of_id plan id = @@ -38,6 +41,13 @@ let node_of_id plan id = in search plan.pl_nodes +let type_of plan stamp = + let rec search = function + | [] -> Diagnostic.error Location.none "internal error: no type recorded for this identifier" + | (candidate, ty) :: rest -> if candidate = stamp then ty else search rest + in + search plan.pl_types + let build program = let nodes = ref [] in let table = Hashtbl.create 16 in @@ -118,11 +128,17 @@ let build program = if Ident.stamp ident = Ident.stamp program.Anf.ap_input then Some (source span) else None) | _ -> None in + let scalar_bindings = ref [] in let rec walk expr = match expr.Anf.a with | Anf.ALet (ident, bound, body) -> (match bind_bound bound with - | Some id -> Hashtbl.replace table (Ident.stamp ident) id + | Some id -> + Hashtbl.replace table (Ident.stamp ident) id; + let node = find id in + (match node.n_kind with + | Sum | Count -> scalar_bindings := (Ident.stamp ident, id) :: !scalar_bindings + | Source | Filter _ | Map _ -> ()) | None -> ()); walk body | Anf.AAtom (Anf.AVar ident) -> ( @@ -134,6 +150,7 @@ let build program = else Result_scalar expr) | _ -> Result_scalar expr in + let scalar_bindings = ref [] in let result = walk program.Anf.ap_query_body in let output = program.Anf.ap_query_body.Anf.aty in let output_element = match Types.repr output with Types.TCollection element -> Some element | _ -> None in @@ -165,6 +182,9 @@ let build program = pl_output = output; pl_output_element = output_element; pl_consumers = consumers; + pl_types = program.Anf.ap_types; + pl_records = program.Anf.ap_records; + pl_scalar_bindings = List.rev !scalar_bindings; } let kind_to_string node = diff --git a/src/graph.mli b/src/graph.mli index a21c6e3..5cd1af6 100644 --- a/src/graph.mli +++ b/src/graph.mli @@ -29,9 +29,13 @@ type plan = { pl_output : Types.t; pl_output_element : Types.t option; pl_consumers : int list array; + pl_types : (int * Types.t) list; + pl_records : Types.records; + pl_scalar_bindings : (int * int) list; } val build : Anf.program -> plan +val type_of : plan -> int -> Types.t val node_of_id : plan -> int -> node val dump : plan -> string val kind_to_string : node -> string diff --git a/src/incremental.ml b/src/incremental.ml new file mode 100644 index 0000000..c80fa4d --- /dev/null +++ b/src/incremental.ml @@ -0,0 +1,420 @@ +type state = { + s_input : Value.t Delta_runtime.Pure_map.t; + s_caches : Value.t Delta_runtime.Pure_map.t option array; + s_accumulators : int array; +} + +type node_change = + | Keyed of Value.t option * Value.t option + | Delta of int + +let upstream node = + match node.Graph.n_input with + | Some id -> id + | None -> + Diagnostic.error node.Graph.n_span "internal error: node %d has no input collection" + node.Graph.n_id + +let keyed state plan node_id = + match state.s_caches.(node_id) with + | Some map -> map + | None -> + Diagnostic.error (Graph.node_of_id plan node_id).Graph.n_span + "internal error: node %d has no cached collection" node_id + +let with_cache state node_id map = + let caches = Array.copy state.s_caches in + caches.(node_id) <- Some map; + { state with s_caches = caches } + +let with_accumulator state node_id value = + let accumulators = Array.copy state.s_accumulators in + accumulators.(node_id) <- value; + { state with s_accumulators = accumulators } + +let lookup name env = + let rec search = function + | [] -> None + | (candidate, value) :: rest -> if candidate = name then Some value else search rest + in + search env + +let int_of_value span value = + match value with + | Value.VInt number -> number + | other -> Diagnostic.error span "runtime error: expected an integer but got %s" (Value.to_string other) + +let bool_of_value span value = + match value with + | Value.VBool truth -> truth + | other -> Diagnostic.error span "runtime error: expected a boolean but got %s" (Value.to_string other) + +let rec eval_scalar env expr = + match expr.Anf.a with + | Anf.AAtom atom -> eval_atom env expr.Anf.aspan atom + | Anf.ABinop (operator, left, right) -> + let left = eval_atom env expr.Anf.aspan left in + let right = eval_atom env expr.Anf.aspan right in + eval_binop expr.Anf.aspan operator left right + | Anf.AIf (condition, then_branch, else_branch) -> + if bool_of_value expr.Anf.aspan (eval_atom env expr.Anf.aspan condition) then + eval_scalar env then_branch + else eval_scalar env else_branch + | Anf.ATuple atoms -> Value.VTuple (List.map (eval_atom env expr.Anf.aspan) atoms) + | Anf.ARecord (name, fields) -> + Value.VRecord + (name, List.map (fun (label, atom) -> (label, eval_atom env expr.Anf.aspan atom)) fields) + | Anf.AField (record, label) -> ( + match Value.field (eval_atom env expr.Anf.aspan record) label with + | Some value -> value + | None -> Diagnostic.error expr.Anf.aspan "runtime error: the record has no field `%s`" label) + | Anf.ALet (ident, bound, body) -> + let value = eval_scalar env bound in + eval_scalar ((Ident.stamp ident, value) :: env) body + | Anf.AApp _ | Anf.ALambda _ | Anf.AFilter _ | Anf.AMap _ | Anf.ASum _ | Anf.ACount _ -> + Diagnostic.error expr.Anf.aspan + "internal error: collection operations may not appear inside scalar expressions" + +and eval_atom env span atom = + match atom with + | Anf.AInt value -> Value.VInt value + | Anf.ABool value -> Value.VBool value + | Anf.AString value -> Value.VString value + | Anf.AUnit -> Value.VUnit + | Anf.AVar ident -> ( + match lookup (Ident.stamp ident) env with + | Some value -> value + | None -> Diagnostic.error span "runtime error: `%s` has no value" (Ident.display ident)) + +and eval_binop span operator left right = + match operator with + | Syntax.Add -> Value.VInt (int_of_value span left + int_of_value span right) + | Syntax.Sub -> Value.VInt (int_of_value span left - int_of_value span right) + | Syntax.Mul -> Value.VInt (int_of_value span left * int_of_value span right) + | Syntax.Div -> Value.VInt (int_of_value span left / int_of_value span right) + | Syntax.Eq -> Value.VBool (Value.equal left right) + | Syntax.Ne -> Value.VBool (not (Value.equal left right)) + | Syntax.Lt -> Value.VBool (int_of_value span left < int_of_value span right) + | Syntax.Le -> Value.VBool (int_of_value span left <= int_of_value span right) + | Syntax.Gt -> Value.VBool (int_of_value span left > int_of_value span right) + | Syntax.Ge -> Value.VBool (int_of_value span left >= int_of_value span right) + | Syntax.And -> Value.VBool (bool_of_value span left && bool_of_value span right) + | Syntax.Or -> Value.VBool (bool_of_value span left || bool_of_value span right) + +let int_delta change = match change with Change.CInt delta -> delta | _ -> 0 + +let rec delta_scalar plan env_old env_new expr = + let span = expr.Anf.aspan in + match expr.Anf.a with + | Anf.AAtom (Anf.AInt _ | Anf.ABool _ | Anf.AString _ | Anf.AUnit) -> Change.CEmpty + | Anf.AAtom (Anf.AVar ident) -> ( + match (lookup (Ident.stamp ident) env_old, lookup (Ident.stamp ident) env_new) with + | Some old_value, Some new_value -> + if Value.equal old_value new_value then Change.CEmpty + else + Change.of_values plan.Graph.pl_records expr.Anf.aty old_value new_value + | _ -> Change.CEmpty) + | Anf.ABinop (Syntax.Add, left, right) -> + Change.CInt + (int_delta (delta_atom plan env_old env_new span left) + + int_delta (delta_atom plan env_old env_new span right)) + | Anf.ABinop (Syntax.Sub, left, right) -> + Change.CInt + (int_delta (delta_atom plan env_old env_new span left) + - int_delta (delta_atom plan env_old env_new span right)) + | Anf.ABinop (Syntax.Mul, left, right) -> + let left_delta = int_delta (delta_atom plan env_old env_new span left) in + let right_delta = int_delta (delta_atom plan env_old env_new span right) in + let left_old = int_of_value span (eval_atom env_old span left) in + let right_old = int_of_value span (eval_atom env_old span right) in + Change.CInt ((left_old * right_delta) + (right_old * left_delta) + (left_delta * right_delta)) + | _ -> + let old_value = eval_scalar env_old expr in + let new_value = eval_scalar env_new expr in + if Value.equal old_value new_value then Change.CEmpty + else Change.of_values plan.Graph.pl_records expr.Anf.aty old_value new_value + +and delta_atom plan env_old env_new span atom = + match atom with + | Anf.AInt _ | Anf.ABool _ | Anf.AString _ | Anf.AUnit -> Change.CEmpty + | Anf.AVar ident -> ( + match (lookup (Ident.stamp ident) env_old, lookup (Ident.stamp ident) env_new) with + | Some old_value, Some new_value -> + if Value.equal old_value new_value then Change.CEmpty + else + let ty = Graph.type_of plan (Ident.stamp ident) in + Change.of_values plan.Graph.pl_records ty old_value new_value + | _ -> Change.CEmpty) + +type context = { + ctx_plan : Graph.plan; + ctx_counters : Delta_runtime.counters; + ctx_root : int option; + ctx_output : Change.change ref; +} + +let keyed_change key old_value new_value = + match (old_value, new_value) with + | None, None -> [] + | None, Some value -> [ Change.CInsert (key, value) ] + | Some value, None -> [ Change.CRemove (key, value) ] + | Some old_value, Some new_value -> + if Value.equal old_value new_value then [] + else [ Change.CReplace (key, old_value, new_value) ] + +let unchanged old_value new_value = + match (old_value, new_value) with + | None, None -> true + | Some old_value, Some new_value -> Value.equal old_value new_value + | _ -> false + +let record_output context node_id key change = + match context.ctx_root with + | Some root when root = node_id -> ( + match change with + | Keyed (old_value, new_value) -> + let items = keyed_change key old_value new_value in + if items <> [] then + context.ctx_output := Change.compose !(context.ctx_output) (Change.CCollection items) + | Delta delta -> + if delta <> 0 then + context.ctx_output := Change.compose !(context.ctx_output) (Change.CInt delta)) + | _ -> () + +let rec propagate context state key up_old up_new node_id = + let plan = context.ctx_plan in + let node = Graph.node_of_id plan node_id in + Delta_runtime.count_changed_key context.ctx_counters; + match node.Graph.n_kind with + | Graph.Source -> + let input = + match up_new with + | None -> Delta_runtime.Pure_map.remove key state.s_input + | Some value -> Delta_runtime.Pure_map.add key value state.s_input + in + let state = with_cache { state with s_input = input } node_id input in + let change = Keyed (up_old, up_new) in + record_output context node_id key change; + fan_out context state key node_id change + | Graph.Filter (parameter, body) -> + let cached = keyed state plan node_id in + let old_member = Delta_runtime.Pure_map.find_opt key cached in + let new_member = + match up_new with + | None -> None + | Some value -> + Delta_runtime.count_predicate context.ctx_counters; + if bool_of_value node.Graph.n_span (eval_scalar [ (Ident.stamp parameter, value) ] body) then + Some value + else None + in + let cached = + match (old_member, new_member) with + | None, None -> cached + | None, Some value -> Delta_runtime.Pure_map.add key value cached + | Some _, None -> Delta_runtime.Pure_map.remove key cached + | Some _, Some value -> Delta_runtime.Pure_map.add key value cached + in + let state = with_cache state node_id cached in + let change = Keyed (old_member, new_member) in + record_output context node_id key change; + if unchanged old_member new_member then state else fan_out context state key node_id change + | Graph.Map (parameter, body) -> + let cached = keyed state plan node_id in + let old_value = Delta_runtime.Pure_map.find_opt key cached in + let new_value = + match up_new with + | None -> None + | Some value -> ( + match up_old with + | None -> + Delta_runtime.count_mapping context.ctx_counters; + Some (eval_scalar [ (Ident.stamp parameter, value) ] body) + | Some previous -> + Delta_runtime.count_scalar_delta context.ctx_counters; + let delta = + delta_scalar plan + [ (Ident.stamp parameter, previous) ] + [ (Ident.stamp parameter, value) ] + body + in + match old_value with + | Some cached_value -> Some (Change.apply cached_value delta) + | None -> Some (eval_scalar [ (Ident.stamp parameter, value) ] body)) + in + let cached = + match (old_value, new_value) with + | None, None -> cached + | None, Some value -> Delta_runtime.Pure_map.add key value cached + | Some _, None -> Delta_runtime.Pure_map.remove key cached + | Some _, Some value -> Delta_runtime.Pure_map.add key value cached + in + let state = with_cache state node_id cached in + let change = Keyed (old_value, new_value) in + record_output context node_id key change; + if unchanged old_value new_value then state else fan_out context state key node_id change + | Graph.Sum -> + let previous = match up_old with Some value -> int_of_value node.Graph.n_span value | None -> 0 in + let current = match up_new with Some value -> int_of_value node.Graph.n_span value | None -> 0 in + let delta = current - previous in + let state = with_accumulator state node_id (state.s_accumulators.(node_id) + delta) in + let change = Delta delta in + record_output context node_id key change; + if delta = 0 then state else fan_out context state key node_id change + | Graph.Count -> + let delta = + match (up_old, up_new) with + | None, None -> 0 + | None, Some _ -> 1 + | Some _, None -> -1 + | Some _, Some _ -> 0 + in + let state = with_accumulator state node_id (state.s_accumulators.(node_id) + delta) in + let change = Delta delta in + record_output context node_id key change; + if delta = 0 then state else fan_out context state key node_id change + +and fan_out context state key node_id change = + let old_value = match change with Keyed (old_value, _) -> old_value | Delta _ -> None in + let new_value = match change with Keyed (_, new_value) -> new_value | Delta _ -> None in + List.fold_left + (fun state consumer -> propagate context state key old_value new_value consumer) + state context.ctx_plan.Graph.pl_consumers.(node_id) + +let init plan counters entries = + let input = Value.collection_of_list entries in + let state = + { + s_input = input; + s_caches = Array.make (List.length plan.Graph.pl_nodes) None; + s_accumulators = Array.make (List.length plan.Graph.pl_nodes) 0; + } + in + let state = ref state in + List.iter + (fun node -> + Delta_runtime.count_full_traversal counters; + match node.Graph.n_kind with + | Graph.Source -> state := with_cache !state node.Graph.n_id input + | Graph.Filter (parameter, body) -> + let upstream = keyed !state plan (upstream node) in + let kept = + Delta_runtime.Pure_map.fold + (fun key value acc -> + Delta_runtime.count_predicate counters; + if + bool_of_value node.Graph.n_span + (eval_scalar [ (Ident.stamp parameter, value) ] body) + then Delta_runtime.Pure_map.add key value acc + else acc) + upstream Delta_runtime.Pure_map.empty + in + state := with_cache !state node.Graph.n_id kept + | Graph.Map (parameter, body) -> + let upstream = keyed !state plan (upstream node) in + let mapped = + Delta_runtime.Pure_map.fold + (fun key value acc -> + Delta_runtime.count_mapping counters; + Delta_runtime.Pure_map.add key (eval_scalar [ (Ident.stamp parameter, value) ] body) acc) + upstream Delta_runtime.Pure_map.empty + in + state := with_cache !state node.Graph.n_id mapped + | Graph.Sum -> + let upstream = keyed !state plan (upstream node) in + let total = + Delta_runtime.Pure_map.fold + (fun _ value acc -> acc + int_of_value node.Graph.n_span value) + upstream 0 + in + state := with_accumulator !state node.Graph.n_id total + | Graph.Count -> + let upstream = keyed !state plan (upstream node) in + state := with_accumulator !state node.Graph.n_id (Delta_runtime.Pure_map.cardinal upstream)) + plan.Graph.pl_nodes; + !state + +let scalar_env plan state = + List.map + (fun (stamp, node_id) -> (stamp, Value.VInt state.s_accumulators.(node_id))) + plan.Graph.pl_scalar_bindings + +let result plan state = + match plan.Graph.pl_result with + | Graph.Result_collection id -> ( + let node = Graph.node_of_id plan id in + match node.Graph.n_kind with + | Graph.Sum | Graph.Count -> Value.VInt state.s_accumulators.(id) + | Graph.Source | Graph.Filter _ | Graph.Map _ -> Value.VCollection (keyed state plan id)) + | Graph.Result_scalar expr -> eval_scalar (scalar_env plan state) expr + +let root_node plan = + match plan.Graph.pl_result with Graph.Result_collection id -> Some id | Graph.Result_scalar _ -> None + +let apply_keyed plan counters state changes = + let root = root_node plan in + let scalar_before = + match plan.Graph.pl_result with + | Graph.Result_scalar expr -> eval_scalar (scalar_env plan state) expr + | _ -> Value.VUnit + in + let source = + match plan.Graph.pl_nodes with + | node :: _ when node.Graph.n_kind = Graph.Source -> node.Graph.n_id + | _ -> -1 + in + let context = + { ctx_plan = plan; ctx_counters = counters; ctx_root = root; ctx_output = ref Change.CEmpty } + in + let state = + List.fold_left + (fun state (key, up_old, up_new) -> + propagate context state key up_old up_new source) + state changes + in + let output_change = + match plan.Graph.pl_result with + | Graph.Result_collection _ -> !(context.ctx_output) + | Graph.Result_scalar expr -> + let after = eval_scalar (scalar_env plan state) expr in + Change.CInt (int_of_value Location.none after - int_of_value Location.none scalar_before) + in + (state, output_change) + +let apply_batch plan counters state ops = + match Change.validate_batch ~existing:state.s_input ops with + | Change.Failure message -> Change.Failure message + | Change.Success (_, changes) -> + let changes = + List.map + (fun item -> + match item with + | Change.CInsert (key, value) -> (key, None, Some value) + | Change.CRemove (key, value) -> (key, Some value, None) + | Change.CReplace (key, old_value, new_value) -> (key, Some old_value, Some new_value)) + changes + in + (try + let state_after, output_change = apply_keyed plan counters state changes in + Change.Success (state_after, output_change) + with + | Division_by_zero -> Change.Failure "division by zero while applying a batch" + | Diagnostic.Error diagnostic -> Change.Failure diagnostic.Diagnostic.message) + +let rule_to_string node = + match node.Graph.n_kind with + | Graph.Source -> "the input map is the source of every change" + | Graph.Filter _ -> + "membership is read from the cached values; the predicate is evaluated for inserted and replaced keys only" + | Graph.Map _ -> + "cached values are updated by the change rule of the mapping expression for replacements" + | Graph.Sum -> "the accumulator is updated by the new contribution minus the old one" + | Graph.Count -> "the accumulator changes only when membership changes" + +let dump_delta plan = + Graph.dump plan + ^ Util.join "\n" + (List.map + (fun node -> Printf.sprintf " delta %d: %s" node.Graph.n_id (rule_to_string node)) + (List.filter (fun node -> node.Graph.n_kind <> Graph.Source) plan.Graph.pl_nodes)) + ^ "\n" diff --git a/src/incremental.mli b/src/incremental.mli new file mode 100644 index 0000000..34164ae --- /dev/null +++ b/src/incremental.mli @@ -0,0 +1,21 @@ +type state = { + s_input : Value.t Delta_runtime.Pure_map.t; + s_caches : Value.t Delta_runtime.Pure_map.t option array; + s_accumulators : int array; +} + +val init : Graph.plan -> Delta_runtime.counters -> (int * Value.t) list -> state +val result : Graph.plan -> state -> Value.t +val apply_keyed : + Graph.plan -> + Delta_runtime.counters -> + state -> + (int * Value.t option * Value.t option) list -> + state * Change.change +val apply_batch : + Graph.plan -> + Delta_runtime.counters -> + state -> + Change.input_op list -> + (state * Change.change) Change.outcome +val dump_delta : Graph.plan -> string diff --git a/src/main.ml b/src/main.ml index 56bdea8..e5c306f 100644 --- a/src/main.ml +++ b/src/main.ml @@ -106,7 +106,7 @@ let check path = let dump stage path = match stage with | "anf" -> print_string (Anf.program_to_string (anf_source path)) - | "delta" -> print_string (Graph.dump (plan_source path)) + | "delta" -> print_string (Incremental.dump_delta (plan_source path)) | _ -> let program = infer_source path in match stage with diff --git a/test/test_incremental.ml b/test/test_incremental.ml index 1076f02..4261eb3 100644 --- a/test/test_incremental.ml +++ b/test/test_incremental.ml @@ -197,3 +197,236 @@ let graph_cases = in check_equal_int "four nodes" 4 (List.length plan.Graph.pl_nodes) ); ] + +type fixture = { + fx_program : Typed.program; + fx_plan : Graph.plan; + fx_counters : Delta_runtime.counters; + mutable fx_state : Incremental.state; + fx_entries : (int * Value.t) list; +} + +let build_fixture text entries = + let program = infer text in + let plan = Graph.build (Anf.program (Specialize.program program)) in + let counters = Delta_runtime.new_counters () in + let state = Incremental.init plan counters entries in + { fx_program = program; fx_plan = plan; fx_counters = counters; fx_state = state; fx_entries = entries } + +let fixture_entries plan state = Delta_runtime.Pure_map.bindings state.Incremental.s_input + +let reference_result fixture state = + Interpret.program fixture.fx_program (fixture_entries fixture.fx_plan state) + +let reference_after fixture ops = + match Change.validate_batch ~existing:fixture.fx_state.Incremental.s_input ops with + | Change.Failure message -> Error message + | Change.Success (temp, _) -> + let program = fixture.fx_program in + ignore program; + Ok (Interpret.program fixture.fx_program (Delta_runtime.Pure_map.bindings temp)) + +let step fixture ops = + let before = Incremental.result fixture.fx_plan fixture.fx_state in + match Incremental.apply_batch fixture.fx_plan fixture.fx_counters fixture.fx_state ops with + | Change.Failure message -> Error message + | Change.Success (state, change) -> + let applied = Change.apply before change in + let cached_after = Incremental.result fixture.fx_plan state in + fixture.fx_state <- state; + Ok (applied, cached_after, change) + +let order_row customer total = + Value.VRecord ("order", [ ("customer", Value.VString customer); ("total", Value.VInt total) ]) + +let executor_cases = + let orders text entries = + let fixture = + build_fixture + ("input orders : collection order\n" ^ text) + entries + in + fixture + in + [ + ( "initialization matches the reference interpreter", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") + [ (1, order_row "Ada" 1500); (2, order_row "Bo" 900) ] + in + check_equal_string "initial result" + (Value.to_string (reference_result fixture fixture.fx_state)) + (Value.to_string (Incremental.result fixture.fx_plan fixture.fx_state)) ); + ( "inserting a key appends a mapped contribution", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] + in + (match step fixture [ Change.OpInsert (2, order_row "Lin" 2200) ] with + | Error message -> fail "insert" message + | Ok (applied, cached, change) -> + let expected = reference_result fixture fixture.fx_state in + check_equal_string "change" "collection [insert 2 (tuple \"Lin\" 440)]" + (Change.to_string change); + check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); + check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); + ( "removing a key retracts the cached contribution", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") + [ (1, order_row "Ada" 1500); (2, order_row "Lin" 2200) ] + in + (match step fixture [ Change.OpRemove 2 ] with + | Error message -> fail "remove" message + | Ok (applied, cached, change) -> + let expected = reference_result fixture fixture.fx_state in + check_equal_string "change" "collection [remove 2 (tuple \"Lin\" 440)]" + (Change.to_string change); + check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); + check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); + ( "replacing a retained key forwards the new contribution", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] + in + (match step fixture [ Change.OpReplace (1, order_row "Ada" 2500) ] with + | Error message -> fail "replace" message + | Ok (applied, cached, change) -> + let expected = reference_result fixture fixture.fx_state in + check_equal_string "change" "collection [replace 1 (tuple \"Ada\" 300) (tuple \"Ada\" 500)]" + (Change.to_string change); + check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); + check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); + ( "replacing a key with an equal row produces no change", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] + in + (match step fixture [ Change.OpReplace (1, order_row "Ada" 1500) ] with + | Error message -> fail "equal" message + | Ok (applied, cached, change) -> + check "empty change" (Change.is_empty change); + check_equal_string "applied" (Value.to_string (reference_result fixture fixture.fx_state)) + (Value.to_string applied); + check_equal_string "cached" (Value.to_string (reference_result fixture fixture.fx_state)) + (Value.to_string cached)) ); + ( "a row that stops matching leaves the collection only once", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] + in + (match step fixture [ Change.OpReplace (1, order_row "Ada" 100) ] with + | Error message -> fail "drop" message + | Ok (applied, cached, change) -> + let expected = reference_result fixture fixture.fx_state in + check_equal_string "change" "collection [remove 1 (tuple \"Ada\" 300)]" + (Change.to_string change); + check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); + check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); + ( "a row that starts matching is inserted once", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 100) ] + in + (match step fixture [ Change.OpReplace (1, order_row "Ada" 5000) ] with + | Error message -> fail "gain" message + | Ok (applied, cached, change) -> + let expected = reference_result fixture fixture.fx_state in + check_equal_string "change" "collection [insert 1 (tuple \"Ada\" 1000)]" + (Change.to_string change); + check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); + check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); + ( "several batches keep the state consistent", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] + in + (match step fixture [ Change.OpInsert (2, order_row "Bo" 3000) ] with + | Error message -> fail "insert" message + | Ok _ -> ()); + (match step fixture [ Change.OpRemove 1 ] with + | Error message -> fail "remove" message + | Ok _ -> ()); + (match step fixture [ Change.OpReplace (2, order_row "Bo" 900) ] with + | Error message -> fail "replace" message + | Ok (applied, cached, _) -> + let expected = reference_result fixture fixture.fx_state in + check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); + check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); + ( "an identity query forwards the input change", + fun () -> + let fixture = + build_fixture "input rows : collection int\nquery q = rows" [ (1, Value.VInt 5) ] + in + (match step fixture [ Change.OpInsert (2, Value.VInt 9) ] with + | Error message -> fail "identity" message + | Ok (applied, cached, change) -> + check_equal_string "change" "collection [insert 2 9]" (Change.to_string change); + check_equal_string "applied" "(collection (1 5) (2 9))" (Value.to_string applied); + check_equal_string "cached" "(collection (1 5) (2 9))" (Value.to_string cached)) ); + ( "an integer query reports an additive change", + fun () -> + let fixture = + build_fixture (read_fixture "revenue.delta") [ (1, order_row "Ada" 1000) ] + in + (match step fixture [ Change.OpInsert (2, order_row "Bo" 250) ] with + | Error message -> fail "sum" message + | Ok (applied, cached, change) -> + check_equal_string "change" "+50" (Change.to_string change); + check_equal_string "applied" "250" (Value.to_string applied); + check_equal_string "cached" "250" (Value.to_string cached)) ); + ( "count ignores replacements that keep a row", + fun () -> + let fixture = + build_fixture (read_fixture "count_large.delta") [ (1, order_row "Ada" 1000) ] + in + (match step fixture [ Change.OpReplace (1, order_row "Ada" 2000) ] with + | Error message -> fail "count" message + | Ok (_, _, change) -> check "empty change" (Change.is_empty change)); + (match step fixture [ Change.OpReplace (1, order_row "Ada" 10) ] with + | Error message -> fail "count" message + | Ok (_, cached, change) -> + check_equal_string "change" "-1" (Change.to_string change); + check_equal_string "cached" "0" (Value.to_string cached)) ); + ( "an invalid batch leaves the state usable", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] + in + let before = Value.to_string (Incremental.result fixture.fx_plan fixture.fx_state) in + (match step fixture [ Change.OpRemove 99 ] with + | Ok _ -> fail "invalid" "expected a failure" + | Error _ -> ()); + check_equal_string "result unchanged" before + (Value.to_string (Incremental.result fixture.fx_plan fixture.fx_state)); + (match step fixture [ Change.OpInsert (2, order_row "Bo" 2000) ] with + | Error message -> fail "recovery" message + | Ok (applied, _, _) -> + check_equal_string "still incremental" (Value.to_string (reference_result fixture fixture.fx_state)) + (Value.to_string applied)) ); + ( "updates visit one key per node", + fun () -> + let fixture = + build_fixture (read_fixture "expensive_order.delta") + [ (1, order_row "Ada" 1500); (2, order_row "Bo" 2000); (3, order_row "Cy" 500) ] + in + Delta_runtime.reset_counters fixture.fx_counters; + (match step fixture [ Change.OpReplace (2, order_row "Bo" 2500) ] with + | Error message -> fail "visit" message + | Ok _ -> + let counters = fixture.fx_counters in + check_equal_int "changed key visits" 3 counters.Delta_runtime.changed_key_visits; + check_equal_int "predicate evaluations" 1 counters.Delta_runtime.predicate_evaluations; + check_equal_int "mapping evaluations" 0 counters.Delta_runtime.mapping_evaluations; + check_equal_int "scalar deltas" 1 counters.Delta_runtime.scalar_deltas; + check_equal_int "full traversals" 0 counters.Delta_runtime.full_traversals) ); + ( "initialization counts one traversal per node", + fun () -> + let plan = plan_of_fixture "expensive_order.delta" in + let counters = Delta_runtime.new_counters () in + let _ = Incremental.init plan counters [ (1, order_row "Ada" 1500) ] in + check_equal_int "full traversals" 3 counters.Delta_runtime.full_traversals; + check_equal_int "mapping evaluations" 1 counters.Delta_runtime.mapping_evaluations; + check_equal_int "predicate evaluations" 1 counters.Delta_runtime.predicate_evaluations ); + ] diff --git a/test/test_main.ml b/test/test_main.ml index 419acba..54e83a2 100644 --- a/test/test_main.ml +++ b/test/test_main.ml @@ -380,5 +380,6 @@ let () = Test_harness.run_suite "batches" Test_change.batch_cases; Test_harness.run_suite "interpret" Test_incremental.cases; Test_harness.run_suite "graph" Test_incremental.graph_cases; + Test_harness.run_suite "executor" Test_incremental.executor_cases; Printf.printf "%d cases, %d failures\n" (Test_harness.case_count ()) (Test_harness.failure_count ()); exit (if Test_harness.failure_count () = 0 then 0 else 1)