Propagate map changes through cached outputs

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 =

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;

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"

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 =

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

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"

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

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

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 );
+  ]

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)