Install runtime artifacts and verify external builds

.gitignore

@@ -1,4 +1,4 @@
-_build/
+_build*/
 /deltac
 _bench_out/
 bench/bench_results.txt

Makefile

@@ -33,7 +33,7 @@ RUNTIME_ARCHIVE = $(BUILD)/delta_runtime.cmxa
 RUNTIME_BYTE = $(BUILD)/delta_runtime.cma
 RUNTIME_NATIVE = $(BUILD)/delta_runtime.a
 RUNTIME_ARTIFACTS = $(RUNTIME_ARCHIVE) $(RUNTIME_NATIVE) $(RUNTIME_BYTE) \
-  $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmi,$(RT_ML)) $(RT_CMX)
+  $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmi,$(RT_ML)) $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmo,$(RT_ML)) $(RT_CMX)
 
 ifeq ($(strip $(RT_ML)),)
 RUNTIME_ARCHIVE =
@@ -49,7 +49,7 @@ LIBS = unix.cmxa
 
 .PHONY: all clean test bench install uninstall
 
-all: deltac $(LIB_ARCHIVE) $(RUNTIME_ARCHIVE) $(TEST_EXE)
+all: deltac $(LIB_ARCHIVE) $(RUNTIME_ARCHIVE) $(RUNTIME_BYTE) $(TEST_EXE)
 
 $(BUILD):
 	mkdir -p $(BUILD)
@@ -139,9 +139,12 @@ $(RUNTIME_ARCHIVE) $(RUNTIME_NATIVE): $(RT_CMX)
 ifeq ($(strip $(RT_ML)),)
 	@echo "no runtime sources"
 else
-	$(OCAMLOPT) -a -o $@ $(RT_CMX)
+	$(OCAMLOPT) -a -o $@ $(RT_ORDER)
 endif
 
+$(RUNTIME_BYTE): $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmo,$(RT_ML))
+	$(OCAMLC) -a -o $@ $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmo,$(RT_ML))
+
 ifeq ($(strip $(TEST_ML)),)
 TEST_EXE =
 else
@@ -149,13 +152,20 @@ $(TEST_EXE): $(TEST_CMX) $(LIB_ARCHIVE) $(RUNTIME_ARCHIVE)
 	$(OCAMLOPT) $(FLAGS) -o $@ $(LIBS) $(RUNTIME_ARCHIVE) $(LIB_ARCHIVE) $(TEST_ORDER)
 endif
 
+ifeq ($(strip $(BENCH_ML)),)
+BENCH_EXE =
+else
+$(BENCH_EXE): $(BUILD)/bench.cmx $(LIB_ARCHIVE) $(RUNTIME_ARCHIVE) deltac
+	$(OCAMLOPT) $(FLAGS) -o $@ $(LIBS) $(RUNTIME_ARCHIVE) $(LIB_ARCHIVE) $(BUILD)/bench.cmx
+endif
+
 test: $(TEST_EXE)
-	DELTA_ROOT=$(CURDIR) DELTA_BUILD_DIR=$(CURDIR)/$(BUILD) ./$(TEST_EXE)
+	DELTA_ROOT=$(CURDIR) DELTA_BUILD_DIR=$(CURDIR)/$(BUILD) DELTA_RUNTIME_DIR=$(CURDIR)/$(BUILD) DELTA_OCAMLOPT="$(OCAMLOPT)" ./$(TEST_EXE)
 
 bench: $(BENCH_EXE)
-	DELTA_ROOT=$(CURDIR) DELTA_BUILD_DIR=$(CURDIR)/$(BUILD) ./$(BENCH_EXE)
+	DELTA_ROOT=$(CURDIR) DELTA_BUILD_DIR=$(CURDIR)/$(BUILD) DELTA_OCAMLOPT="$(OCAMLOPT)" ./$(BENCH_EXE)
 
-install: deltac $(RUNTIME_ARCHIVE)
+install: deltac $(RUNTIME_ARCHIVE) $(RUNTIME_BYTE)
 	$(INSTALL) -d $(DESTDIR)$(BINDIR) $(DESTDIR)$(LIBDIR)
 	$(INSTALL) -m 755 deltac $(DESTDIR)$(BINDIR)/deltac
 	$(INSTALL) -m 644 $(RUNTIME_ARTIFACTS) $(DESTDIR)$(LIBDIR)/

bench/bench.ml

@@ -0,0 +1,191 @@
+let root () = try Sys.getenv "DELTA_ROOT" with Not_found -> "."
+
+let fail name message =
+  prerr_endline (Printf.sprintf "bench: %s: %s" name message);
+  exit 1
+
+let deltac () = Filename.concat (root ()) "deltac"
+
+let build_dir () = try Sys.getenv "DELTA_BUILD_DIR" with Not_found -> "_build"
+
+let run command =
+  let log = Filename.temp_file "delta_bench" ".log" in
+  let fd = Unix.openfile log [ Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC ] 0o600 in
+  let argv = Array.of_list command in
+  let pid = Unix.create_process argv.(0) argv Unix.stdin fd fd in
+  let status = snd (Unix.waitpid [] pid) in
+  Unix.close fd;
+  let text = Native.read_file log in
+  (try Sys.remove log with _ -> ());
+  (status, text)
+
+let write_file path text =
+  let channel = open_out path in
+  output_string channel text;
+  close_out channel
+
+type row = { customer : string; total : int }
+
+let make_row index = { customer = Printf.sprintf "customer%d" index; total = (index * 37) mod 4001 }
+
+let row_text row = Printf.sprintf "(record (customer %S) (total %d))" row.customer row.total
+
+let round_of size seed = ((seed * 1103515245) + size) mod size
+
+let input_text size =
+  let buffer = Buffer.create (size * 48) in
+  for index = 0 to size - 1 do
+    Buffer.add_string buffer (Printf.sprintf "(%d %s)\n" (index + 1) (row_text (make_row index)))
+  done;
+  Buffer.contents buffer
+
+let updates_text size batches =
+  let buffer = Buffer.create (batches * 64) in
+  for batch = 0 to batches - 1 do
+    let key = (round_of size batch) + 1 in
+    if batch mod 2 = 0 then
+      Buffer.add_string buffer
+        (Printf.sprintf "(batch (replace %d %s))\n" key
+           (row_text { customer = Printf.sprintf "customer%d" key; total = 2000 + (batch mod 500) }))
+    else
+      Buffer.add_string buffer
+        (Printf.sprintf "(batch (insert %d %s))\n" (size + batch + 1)
+           (row_text { customer = Printf.sprintf "extra%d" batch; total = 1500 + (batch mod 250) }))
+  done;
+  Buffer.contents buffer
+
+type stats = {
+  st_init_seconds : float;
+  st_update_seconds : float;
+  st_minor_words : float;
+  st_init_counters : string;
+  st_update_counters : string;
+}
+
+let field prefix line =
+  let length = String.length prefix in
+  if String.length line > length && String.sub line 0 length = prefix then
+    Some (String.trim (String.sub line length (String.length line - length)))
+  else None
+
+let counter_value name text =
+  let parts = String.split_on_char ' ' text in
+  let rec search = function
+    | [] -> 0
+    | part :: rest -> (
+        match (try Some (String.index part '=') with Not_found -> None) with
+        | Some index
+          when String.sub part 0 index = name ->
+            int_of_string (String.sub part (index + 1) (String.length part - index - 1))
+        | _ -> search rest)
+  in
+  search parts
+
+let parse_stats text =
+  let lines = String.split_on_char '\n' text in
+  let value prefix default =
+    let rec search = function
+      | [] -> default
+      | line :: rest -> (
+          match field prefix line with
+          | Some text -> float_of_string text
+          | None -> search rest)
+    in
+    search lines
+  in
+  let counter prefix default =
+    let rec search = function
+      | [] -> default
+      | line :: rest -> (
+          match field prefix line with Some text -> text | None -> search rest)
+    in
+    search lines
+  in
+  {
+    st_init_seconds = value "init_seconds: " 0.0;
+    st_update_seconds = value "update_seconds: " 0.0;
+    st_minor_words = value "minor_words: " 0.0;
+    st_init_counters = counter "init_counters: " "";
+    st_update_counters = counter "update_counters: " "";
+  }
+
+let query_source name =
+  let path = Filename.concat (Filename.concat (root ()) "example") (name ^ ".delta") in
+  Native.read_file path
+
+let compile_query name directory =
+  let source = query_source name in
+  let program = Filename.concat directory (name ^ ".ml") in
+  write_file program source;
+  let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program (Infer.program (Resolve.program (Parse.program source)))))) in
+  let emitted = Filename.concat directory (name ^ "_generated.ml") in
+  write_file emitted (Emit.program_to_string plan);
+  let executable = Filename.concat directory name in
+  let command =
+    [ (try Sys.getenv "DELTA_OCAMLOPT" with Not_found -> "ocamlopt");
+      "-w"; "-26"; "-I"; build_dir (); "-I"; "+unix"; "-o"; executable;
+      Filename.concat (build_dir ()) "delta_runtime.cmxa"; "unix.cmxa"; emitted ]
+  in
+  match run command with
+  | Unix.WEXITED 0, _ -> executable
+  | _, output -> fail "compile" output
+
+let sizes = [ 1000; 100000; 1000000 ]
+
+let queries = [ "expensive_order"; "revenue"; "count_large" ]
+
+let () =
+  let directory = Filename.temp_file "delta_bench" "" in
+  Sys.remove directory;
+  Unix.mkdir directory 0o700;
+  print_endline "delta benchmarks";
+  print_endline "";
+  print_endline
+    "query             rows      init_ms  update_ms  batches  update_us_per_batch  minor_words  predicate_evals  mapping_evals  changed_keys  full_traversals";
+  let regression_failures = ref 0 in
+  List.iter
+    (fun name ->
+      let executable = compile_query name directory in
+      List.iter
+        (fun size ->
+          let input_path = Filename.concat directory (Printf.sprintf "input_%s_%d.sexp" name size) in
+          let updates_path = Filename.concat directory (Printf.sprintf "update_%s_%d.sexp" name size) in
+          write_file input_path (input_text size);
+          write_file updates_path (updates_text size 200);
+          let status, output =
+            run [ executable; "--input"; input_path; "--updates"; updates_path; "--print-result"; "--stats" ]
+          in
+          if status <> Unix.WEXITED 0 then fail "benchmark run" output;
+          let stats = parse_stats output in
+          let batches = 200 in
+          Printf.printf "%-17s %-9d %-8.3f %-9.3f %-8d %-20.3f %-12.0f %-16d %-14d %-13d %d\n" name size
+            (stats.st_init_seconds *. 1000.0) (stats.st_update_seconds *. 1000.0) batches
+            (stats.st_update_seconds *. 1000000.0 /. float_of_int batches)
+            stats.st_minor_words
+            (counter_value "predicate_evaluations" stats.st_update_counters)
+            (counter_value "mapping_evaluations" stats.st_update_counters)
+            (counter_value "changed_key_visits" stats.st_update_counters)
+            (counter_value "full_traversals" stats.st_update_counters);
+          if counter_value "full_traversals" stats.st_update_counters <> 0 then (
+            incr regression_failures;
+            Printf.printf "  regression: updates performed %d full traversals for %s at %d rows\n"
+              (counter_value "full_traversals" stats.st_update_counters) name size);
+          if counter_value "predicate_evaluations" stats.st_update_counters > 200 * 4 then (
+            incr regression_failures;
+            Printf.printf "  regression: too many predicate evaluations (%d) for %s at %d rows\n"
+              (counter_value "predicate_evaluations" stats.st_update_counters) name size);
+          if counter_value "mapping_evaluations" stats.st_update_counters > 200 * 4 then (
+            incr regression_failures;
+            Printf.printf "  regression: too many mapping evaluations (%d) for %s at %d rows\n"
+              (counter_value "mapping_evaluations" stats.st_update_counters) name size);
+          Sys.remove input_path;
+          Sys.remove updates_path)
+        sizes)
+    queries;
+  print_endline "";
+  if !regression_failures = 0 then
+    print_endline
+      "regression check: single-key updates performed no full traversals and touched only the changed keys"
+  else Printf.printf "%d regression checks failed\n" !regression_failures;
+  Native.remove_dir directory;
+  exit (if !regression_failures = 0 then 0 else 1)

example/order.sexp

@@ -0,0 +1,5 @@
+(1 (record (customer "Ada") (total 1500)))
+(2 (record (customer "Bo") (total 900)))
+(3 (record (customer "Lin") (total 2200)))
+(4 (record (customer "Dara") (total 400)))
+(5 (record (customer "Eve") (total 5000)))

example/update.sexp

@@ -0,0 +1,8 @@
+(batch
+  (insert 6 (record (customer "Fay") (total 4200)))
+  (replace 1 (record (customer "Ada") (total 100))))
+(batch
+  (replace 3 (record (customer "Lin") (total 50)))
+  (remove 5))
+(batch
+  (replace 4 (record (customer "Dara") (total 2500))))

opam

@@ -2,6 +2,7 @@ opam-version: "1.2"
 name: "delta"
 version: "0.1.0"
 maintainer: "deltac@localhost"
+authors: [ "deltac@localhost" ]
 build: [
   [make]
 ]

runtime/wire.ml

@@ -0,0 +1,353 @@
+type value =
+  | WUnit
+  | WInt of int
+  | WBool of bool
+  | WString of string
+  | WTuple of value list
+  | WRecord of (string * value) list
+  | WCollection of (int * value) list
+
+type op = WInsert of int * value | WRemove of int | WReplace of int * value
+
+type state = {
+  text : string;
+  mutable position : int;
+  mutable line : int;
+  mutable column : int;
+}
+
+let failure state message =
+  Delta_runtime.Failure (Printf.sprintf "line %d, column %d: %s" state.line state.column message)
+
+let failure_at state line column message =
+  Delta_runtime.Failure (Printf.sprintf "line %d, column %d: %s" line column message)
+
+let at_end state = state.position >= String.length state.text
+
+let peek state = if at_end state then '\000' else state.text.[state.position]
+
+let advance state =
+  (if peek state = '\n' then (
+     state.line <- state.line + 1;
+     state.column <- 1)
+   else state.column <- state.column + 1);
+  state.position <- state.position + 1
+
+let rec skip_space state =
+  if not (at_end state) then
+    match peek state with
+    | ' ' | '\t' | '\r' | '\n' ->
+        advance state;
+        skip_space state
+    | _ -> ()
+
+let expect state character =
+  if peek state = character then (
+    advance state;
+    true)
+  else false
+
+let is_digit character = character >= '0' && character <= '9'
+
+let is_symbol_start character =
+  (character >= 'a' && character <= 'z')
+  || (character >= 'A' && character <= 'Z')
+  || character = '_'
+
+let is_symbol_char character = is_symbol_start character || is_digit character || character = '\''
+
+let read_symbol state =
+  let start = state.position in
+  let rec scan () =
+    if not (at_end state) && is_symbol_char (peek state) then (
+      advance state;
+      scan ())
+  in
+  scan ();
+  String.sub state.text start (state.position - start)
+
+let read_string state =
+  advance state;
+  let buffer = Buffer.create 16 in
+  let rec scan () =
+    if at_end state then failure state "unterminated string"
+    else
+      let character = peek state in
+      if character = '"' then (
+        advance state;
+        Delta_runtime.Success (Buffer.contents buffer))
+      else if character = '\\' then (
+        advance state;
+        let escaped =
+          if at_end state then '\000'
+          else
+            let next = peek state in
+            advance state;
+            next
+        in
+        match escaped with
+        | 'n' ->
+            Buffer.add_char buffer '\n';
+            scan ()
+        | 't' ->
+            Buffer.add_char buffer '\t';
+            scan ()
+        | 'r' ->
+            Buffer.add_char buffer '\r';
+            scan ()
+        | '"' ->
+            Buffer.add_char buffer '"';
+            scan ()
+        | '\\' ->
+            Buffer.add_char buffer '\\';
+            scan ()
+        | other -> failure state (Printf.sprintf "unknown escape sequence \\%c" other))
+      else if character = '\n' then failure state "unterminated string"
+      else (
+        Buffer.add_char buffer character;
+        advance state;
+        scan ())
+  in
+  scan ()
+
+let read_integer state =
+  let start = state.position in
+  if peek state = '-' then advance state;
+  if not (is_digit (peek state)) then failure state "expected an integer"
+  else (
+    let rec scan () =
+      if is_digit (peek state) then (
+        advance state;
+        scan ())
+    in
+    scan ();
+    let text = String.sub state.text start (state.position - start) in
+    try Delta_runtime.Success (int_of_string text)
+    with Failure _ -> failure state (Printf.sprintf "integer %s does not fit in a machine integer" text))
+
+let rec read_value state =
+  skip_space state;
+  if at_end state then failure state "expected a value"
+  else
+    match peek state with
+    | '(' -> read_compound state
+    | '"' -> (
+        match read_string state with
+        | Delta_runtime.Success text -> Delta_runtime.Success (WString text)
+        | Delta_runtime.Failure message -> Delta_runtime.Failure message)
+    | character when is_digit character || character = '-' -> (
+        match read_integer state with
+        | Delta_runtime.Success number -> Delta_runtime.Success (WInt number)
+        | Delta_runtime.Failure message -> Delta_runtime.Failure message)
+    | character when is_symbol_start character ->
+        let symbol = read_symbol state in
+        if symbol = "true" then Delta_runtime.Success (WBool true)
+        else if symbol = "false" then Delta_runtime.Success (WBool false)
+        else if symbol = "unit" then Delta_runtime.Success WUnit
+        else failure state (Printf.sprintf "unknown scalar `%s`" symbol)
+    | character -> failure state (Printf.sprintf "unexpected character %C" character)
+
+and read_compound state =
+  ignore (expect state '(');
+  skip_space state;
+  if not (is_symbol_start (peek state)) then failure state "expected a tagged value"
+  else
+    let tag_line = state.line in
+    let tag_column = state.column in
+    let tag = read_symbol state in
+    match tag with
+    | "tuple" ->
+        let rec items acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (WTuple (List.rev acc))
+          else (
+            match read_value state with
+            | Delta_runtime.Success value -> items (value :: acc)
+            | Delta_runtime.Failure message -> Delta_runtime.Failure message)
+        in
+        items []
+    | "record" ->
+        let rec fields acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (WRecord (List.rev acc))
+          else if peek state <> '(' then failure state "expected a record field"
+          else (
+            ignore (expect state '(');
+            skip_space state;
+            if not (is_symbol_start (peek state)) then failure state "expected a field label"
+            else
+              let label = read_symbol state in
+              match read_value state with
+              | Delta_runtime.Failure message -> Delta_runtime.Failure message
+              | Delta_runtime.Success value ->
+                  skip_space state;
+                  if not (expect state ')') then failure state "expected `)` after a record field"
+                  else fields ((label, value) :: acc))
+        in
+        fields []
+    | "collection" ->
+        let rec entries acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (WCollection (List.rev acc))
+          else if peek state <> '(' then failure state "expected a collection entry"
+          else (
+            ignore (expect state '(');
+            match read_integer state with
+            | Delta_runtime.Failure message -> Delta_runtime.Failure message
+            | Delta_runtime.Success key -> (
+                match read_value state with
+                | Delta_runtime.Failure message -> Delta_runtime.Failure message
+                | Delta_runtime.Success value ->
+                    skip_space state;
+                    if not (expect state ')') then failure state "expected `)` after a collection entry"
+                    else entries ((key, value) :: acc)))
+        in
+        entries []
+    | other -> failure_at state tag_line tag_column (Printf.sprintf "unknown tag `%s`" other)
+
+let to_string value =
+  let rec render value =
+    match value with
+    | WUnit -> "unit"
+    | WInt number -> string_of_int number
+    | WBool true -> "true"
+    | WBool false -> "false"
+    | WString text ->
+        let buffer = Buffer.create (String.length text + 2) in
+        Buffer.add_char buffer '"';
+        String.iter
+          (fun character ->
+            match character with
+            | '"' -> Buffer.add_string buffer "\\\""
+            | '\\' -> Buffer.add_string buffer "\\\\"
+            | '\n' -> Buffer.add_string buffer "\\n"
+            | '\t' -> Buffer.add_string buffer "\\t"
+            | '\r' -> Buffer.add_string buffer "\\r"
+            | other -> Buffer.add_char buffer other)
+          text;
+        Buffer.add_char buffer '"';
+        Buffer.contents buffer
+    | WTuple items -> "(tuple " ^ String.concat " " (List.map render items) ^ ")"
+    | WRecord fields ->
+        "(record "
+        ^ String.concat " " (List.map (fun (label, value) -> "(" ^ label ^ " " ^ render value ^ ")") fields)
+        ^ ")"
+    | WCollection entries ->
+        "(collection "
+        ^ String.concat " "
+            (List.rev (List.rev_map (fun (key, value) -> Printf.sprintf "(%d %s)" key (render value)) entries))
+        ^ ")"
+  in
+  render value
+
+let field_value label fields =
+  let matches = List.filter (fun (candidate, _) -> candidate = label) fields in
+  match matches with
+  | [] -> Delta_runtime.Failure (Printf.sprintf "missing field `%s`" label)
+  | [ (_, value) ] -> Delta_runtime.Success value
+  | _ -> Delta_runtime.Failure (Printf.sprintf "duplicate field `%s`" label)
+
+let parse_value text =
+  let state = { text = text; position = 0; line = 1; column = 1 } in
+  match read_value state with
+  | Delta_runtime.Failure message -> Delta_runtime.Failure message
+  | Delta_runtime.Success value ->
+      skip_space state;
+      if at_end state then Delta_runtime.Success value
+      else failure state "unexpected trailing input"
+
+let parse_input text =
+  let state = { text = text; position = 0; line = 1; column = 1 } in
+  let rec entries acc =
+    skip_space state;
+    if at_end state then Delta_runtime.Success (List.rev acc)
+    else if peek state <> '(' then failure state "expected `(key value)`"
+    else (
+      ignore (expect state '(');
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key -> (
+          match read_value state with
+          | Delta_runtime.Failure message -> Delta_runtime.Failure message
+          | Delta_runtime.Success value ->
+              skip_space state;
+              if not (expect state ')') then failure state "expected `)` after an input entry"
+              else entries ((key, value) :: acc)))
+  in
+  entries []
+
+let parse_operation state =
+  ignore (expect state '(');
+  skip_space state;
+  let tag = read_symbol state in
+  skip_space state;
+  match tag with
+  | "insert" -> (
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key -> (
+          match read_value state with
+          | Delta_runtime.Failure message -> Delta_runtime.Failure message
+          | Delta_runtime.Success value ->
+              skip_space state;
+              if not (expect state ')') then failure state "expected `)` after an insert"
+              else Delta_runtime.Success (WInsert (key, value))))
+  | "remove" -> (
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key ->
+          skip_space state;
+          if not (expect state ')') then failure state "expected `)` after a remove"
+          else Delta_runtime.Success (WRemove key))
+  | "replace" -> (
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key -> (
+          match read_value state with
+          | Delta_runtime.Failure message -> Delta_runtime.Failure message
+          | Delta_runtime.Success value ->
+              skip_space state;
+              if not (expect state ')') then failure state "expected `)` after a replace"
+              else Delta_runtime.Success (WReplace (key, value))))
+  | other -> failure state (Printf.sprintf "unknown update operation `%s`" other)
+
+let parse_updates text =
+  let state = { text = text; position = 0; line = 1; column = 1 } in
+  let rec batches acc =
+    skip_space state;
+    if at_end state then Delta_runtime.Success (List.rev acc)
+    else if peek state <> '(' then failure state "expected `(batch ...)`"
+    else (
+      ignore (expect state '(');
+      skip_space state;
+      let tag_line = state.line in
+      let tag_column = state.column in
+      let tag = read_symbol state in
+      if tag <> "batch" then
+        failure_at state tag_line tag_column (Printf.sprintf "expected `batch` but found `%s`" tag)
+      else
+        let rec operations acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (List.rev acc)
+          else if peek state <> '(' then failure state "expected an update operation"
+          else
+            match parse_operation state with
+            | Delta_runtime.Failure message -> Delta_runtime.Failure message
+            | Delta_runtime.Success operation -> operations (operation :: acc)
+        in
+        match operations [] with
+        | Delta_runtime.Failure message -> Delta_runtime.Failure message
+        | Delta_runtime.Success operations -> batches (operations :: acc))
+  in
+  batches []
+
+let read_file path =
+  try
+    let channel = open_in_bin path in
+    let length = in_channel_length channel in
+    let text = really_input_string channel length in
+    close_in channel;
+    Delta_runtime.Success text
+  with
+  | Sys_error message -> Delta_runtime.Failure message
+  | End_of_file -> Delta_runtime.Failure (Printf.sprintf "could not read %s" path)

runtime/wire.mli

@@ -0,0 +1,17 @@
+type value =
+  | WUnit
+  | WInt of int
+  | WBool of bool
+  | WString of string
+  | WTuple of value list
+  | WRecord of (string * value) list
+  | WCollection of (int * value) list
+
+type op = WInsert of int * value | WRemove of int | WReplace of int * value
+
+val to_string : value -> string
+val field_value : string -> (string * value) list -> value Delta_runtime.outcome
+val parse_value : string -> value Delta_runtime.outcome
+val parse_input : string -> (int * value) list Delta_runtime.outcome
+val parse_updates : string -> op list list Delta_runtime.outcome
+val read_file : string -> string Delta_runtime.outcome

src/emit.mli

@@ -0,0 +1,4 @@
+val module_to_string : Graph.plan -> string
+val program_to_string : Graph.plan -> string
+val input_type : Graph.plan -> Types.t
+val ocaml_ty : Types.t -> string

src/incremental.ml

@@ -230,17 +230,20 @@ let rec propagate context state key up_old up_new node_id =
             | None ->
                 Delta_runtime.count_mapping context.ctx_counters;
                 Some (eval_scalar [ (Ident.stamp parameter, value) ] body)
-            | Some previous ->
+            | 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)
+                | Some cached_value ->
-                | None -> Some (eval_scalar [ (Ident.stamp parameter, value) ] body))
+                    Delta_runtime.count_scalar_delta context.ctx_counters;
+                    let delta =
+                      delta_scalar plan
+                        [ (Ident.stamp parameter, previous) ]
+                        [ (Ident.stamp parameter, value) ]
+                        body
+                    in
+                    Some (Change.apply cached_value delta)
+                | None ->
+                    Delta_runtime.count_mapping context.ctx_counters;
+                    Some (eval_scalar [ (Ident.stamp parameter, value) ] body)))
       in
       let cached =
         match (old_value, new_value) with

src/main.ml

@@ -95,7 +95,9 @@ let specialize_source path = Specialize.program (infer_source path)
 
 let anf_source path = Anf.program (specialize_source path)
 
-let plan_source path = Graph.build (anf_source path)
+let graph_source path = Graph.build (anf_source path)
+
+let plan_source path = Simplify.simplify (graph_source path)
 
 let frontend_unavailable () =
   Diagnostic.error Location.none "the delta pipeline beyond parsing is not implemented in this revision"
@@ -106,7 +108,10 @@ let check path =
 let dump stage path =
   match stage with
   | "anf" -> print_string (Anf.program_to_string (anf_source path))
-  | "delta" -> print_string (Incremental.dump_delta (plan_source path))
+  | "delta" ->
+      let graph = graph_source path in
+      print_string
+        (Simplify.decisions graph ^ Incremental.dump_delta (Simplify.simplify graph))
   | _ ->
   let program = infer_source path in
   match stage with
@@ -114,14 +119,29 @@ let dump stage path =
   | _ -> frontend_unavailable ()
 
 let emit path output =
-  ignore (plan_source path);
+  let program = plan_source path in
-  ignore output;
+  let channel = open_out output in
-  frontend_unavailable ()
+  output_string channel (Emit.program_to_string program);
+  close_out channel
+
+let temp_source plan =
+  let directory = Native.temp_dir () in
+  let source = Filename.concat directory "program.ml" in
+  let channel = open_out source in
+  output_string channel (Emit.program_to_string plan);
+  close_out channel;
+  (directory, source)
 
 let build path output =
-  ignore (plan_source path);
+  let plan = plan_source path in
-  ignore output;
+  let directory, source = temp_source plan in
-  frontend_unavailable ()
+  let runtime_dir = Native.runtime_dir () in
+  let result = Native.compile ~runtime_dir ~source ~output in
+  Native.remove_dir directory;
+  match result with
+  | None -> ()
+  | Some message -> Diagnostic.error Location.none "the native compiler failed:
+%s" message
 
 let source_of file = try Some (Native.read_file file) with _ -> None
 

src/native.ml

@@ -101,7 +101,9 @@ let run argv =
 
 let compile ~runtime_dir ~source ~output =
   let archive = Filename.concat runtime_dir "delta_runtime.cmxa" in
-  let argv = [| "ocamlopt"; "-I"; runtime_dir; "-o"; output; archive; source |] in
+  let argv =
+    [| "ocamlopt"; "-w"; "-26"; "-I"; runtime_dir; "-o"; output; archive; source |]
+  in
   if file_exists output then Sys.remove output;
   let status, text = run argv in
   match status with

src/simplify.ml

@@ -0,0 +1,132 @@
+type action =
+  | Keep of Graph.cache
+  | Elide
+
+let rec uses_division expr =
+  match expr.Anf.a with
+  | Anf.ABinop (Syntax.Div, _, _) -> true
+  | Anf.ABinop (_, _, _) -> false
+  | Anf.AIf (_, then_branch, else_branch) ->
+      uses_division then_branch || uses_division else_branch
+  | Anf.ALet (_, bound, body) -> uses_division bound || uses_division body
+  | Anf.AAtom _ | Anf.ATuple _ | Anf.ARecord _ | Anf.AField _ | Anf.AApp _ | Anf.ALambda _ -> false
+  | Anf.AFilter _ | Anf.AMap _ | Anf.ASum _ | Anf.ACount _ -> false
+
+let identity_projection parameter body =
+  match body.Anf.a with Anf.AAtom (Anf.AVar ident) -> Ident.equal ident parameter | _ -> false
+
+let constant_true body = match body.Anf.a with Anf.AAtom (Anf.ABool true) -> true | _ -> false
+
+let count_only plan node_id =
+  let rec reaches_count seen node_id =
+    if Util.contains node_id seen then false
+    else
+      let node = Graph.node_of_id plan node_id in
+      match node.Graph.n_kind with
+      | Graph.Count -> true
+      | Graph.Filter _ -> List.for_all (reaches_count (node_id :: seen)) plan.Graph.pl_consumers.(node_id)
+      | Graph.Map _ -> List.for_all (reaches_count (node_id :: seen)) plan.Graph.pl_consumers.(node_id)
+      | Graph.Source | Graph.Sum -> false
+  in
+  plan.Graph.pl_consumers.(node_id) <> []
+  && List.for_all (reaches_count [ node_id ]) plan.Graph.pl_consumers.(node_id)
+
+let root_id plan = match plan.Graph.pl_result with Graph.Result_collection id -> Some id | _ -> None
+
+let decide plan node =
+  let is_root = match root_id plan with Some id -> id = node.Graph.n_id | None -> false in
+  match node.Graph.n_kind with
+  | Graph.Source | Graph.Sum | Graph.Count -> Keep node.Graph.n_cache
+  | Graph.Filter (_, body) -> if constant_true body then Elide else Keep node.Graph.n_cache
+  | Graph.Map (parameter, body) ->
+      if identity_projection parameter body then Elide
+      else if is_root then Keep node.Graph.n_cache
+      else if count_only plan node.Graph.n_id then
+        if uses_division body then Keep Graph.No_cache else Elide
+      else Keep node.Graph.n_cache
+
+let simplify plan =
+  let actions = List.map (fun node -> (node.Graph.n_id, decide plan node)) plan.Graph.pl_nodes in
+  let action_of id = Util.assoc_opt id actions in
+  let resolved = Hashtbl.create 16 in
+  let rec resolve id =
+    match Util.hashtbl_find_opt resolved id with
+    | Some id -> id
+    | None ->
+        let node = Graph.node_of_id plan id in
+        let result =
+          match (action_of id, node.Graph.n_input) with
+          | Some Elide, Some input -> resolve input
+          | Some Elide, None -> id
+          | _ ->
+              let id = node.Graph.n_id in
+              Hashtbl.replace resolved id id;
+              id
+        in
+        Hashtbl.replace resolved id result;
+        result
+  in
+  let kept =
+    List.filter
+      (fun node ->
+        match action_of node.Graph.n_id with
+        | Some (Keep _) -> true
+        | Some Elide -> false
+        | None -> false)
+      plan.Graph.pl_nodes
+  in
+  let renumbered = List.mapi (fun index node -> (node.Graph.n_id, index)) kept in
+  let new_id old_id =
+    match Util.assoc_opt (resolve old_id) renumbered with
+    | Some id -> id
+    | None -> -1
+  in
+  let nodes =
+    List.map
+      (fun (old_id, fresh_id) ->
+        let node = Graph.node_of_id plan old_id in
+        let cache =
+          match action_of old_id with Some (Keep cache) -> cache | _ -> Graph.No_cache
+        in
+        let input = match node.Graph.n_input with None -> None | Some input -> Some (new_id input) in
+        { node with Graph.n_id = fresh_id; n_input = input; n_cache = cache })
+      renumbered
+  in
+  let node_count = List.length nodes in
+  let consumers = Array.make node_count [] in
+  List.iter
+    (fun node ->
+      match node.Graph.n_input with
+      | Some input when input >= 0 && input < node_count ->
+          consumers.(input) <- consumers.(input) @ [ node.Graph.n_id ]
+      | _ -> ())
+    nodes;
+  let result =
+    match plan.Graph.pl_result with
+    | Graph.Result_collection id -> Graph.Result_collection (new_id id)
+    | Graph.Result_scalar expr -> Graph.Result_scalar expr
+  in
+  let scalar_bindings =
+    List.map (fun (stamp, id) -> (stamp, new_id id)) plan.Graph.pl_scalar_bindings
+  in
+  {
+    plan with
+    Graph.pl_nodes = nodes;
+    pl_result = result;
+    pl_consumers = consumers;
+    pl_scalar_bindings = scalar_bindings;
+  }
+
+let decisions plan =
+  let lines =
+    List.map
+      (fun node ->
+        match decide plan node with
+        | Keep cache ->
+            Printf.sprintf "  keep node %d: %s [cache: %s]" node.Graph.n_id
+              (Graph.kind_to_string node) (Graph.cache_to_string cache)
+        | Elide ->
+            Printf.sprintf "  elide node %d: %s" node.Graph.n_id (Graph.kind_to_string node))
+      plan.Graph.pl_nodes
+  in
+  Util.join "\n" lines ^ "\n"

src/simplify.mli

@@ -0,0 +1,3 @@
+val simplify : Graph.plan -> Graph.plan
+val decisions : Graph.plan -> string
+val uses_division : Anf.expr -> bool

src/util.ml

@@ -19,7 +19,11 @@ let rec concat_map f = function
   | item :: rest -> f item @ concat_map f rest
 
 let rec list_init count f =
-  if count <= 0 then [] else f 0 :: list_init (count - 1) (fun index -> f (index + 1))
+  if count <= 0 then []
+  else
+    let head = f 0 in
+    let tail = list_init (count - 1) (fun index -> f (index + 1)) in
+    head :: tail
 
 let rec take count items =
   if count <= 0 then []

test/test_change.ml

@@ -370,3 +370,18 @@ let batch_cases =
                     (Change.to_string (Change.CCollection changes))))
           (Util.list_init 120 (fun index -> index + 1)) );
   ]
+
+let rec value_for rng records ty =
+  match Types.repr ty with
+  | Types.TInt -> Value.VInt (range rng 41 - 20)
+  | Types.TBool -> Value.VBool (range rng 2 = 0)
+  | Types.TString -> Value.VString (word rng)
+  | Types.TUnit -> Value.VUnit
+  | Types.TTuple items -> Value.VTuple (List.map (value_for rng records) items)
+  | Types.TRecord name -> (
+      match Types.record_info records name with
+      | Some info ->
+          Value.VRecord
+            (name, List.map (fun (label, field_ty) -> (label, value_for rng records field_ty)) info.Types.ri_fields)
+      | None -> Value.VUnit)
+  | Types.TCollection _ | Types.TArrow _ | Types.TVar _ -> Value.VUnit

test/test_codegen.ml

@@ -0,0 +1,760 @@
+open Test_harness
+
+let build_dir () = try Sys.getenv "DELTA_BUILD_DIR" with Not_found -> "_build"
+
+let ocamlopt () = try Sys.getenv "DELTA_OCAMLOPT" with Not_found -> "ocamlopt"
+
+let unique name =
+  Printf.sprintf "%s_%d_%d" name (Unix.getpid ()) (Random.self_init (); Random.int 1000000)
+
+let write_file path text =
+  let channel = open_out path in
+  output_string channel text;
+  close_out channel
+
+let run command =
+  let log = Filename.temp_file "delta_test" ".log" in
+  let fd = Unix.openfile log [ Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC ] 0o600 in
+  let argv = Array.of_list command in
+  let pid = Unix.create_process argv.(0) argv Unix.stdin fd fd in
+  let status = snd (Unix.waitpid [] pid) in
+  Unix.close fd;
+  let text = Native.read_file log in
+  (try Sys.remove log with _ -> ());
+  (status, text)
+
+let run_env environment command =
+  let log = Filename.temp_file "delta_test" ".log" in
+  let fd = Unix.openfile log [ Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC ] 0o600 in
+  let argv = Array.of_list command in
+  let pid = Unix.create_process_env argv.(0) argv environment Unix.stdin fd fd in
+  let status = snd (Unix.waitpid [] pid) in
+  Unix.close fd;
+  let text = Native.read_file log in
+  (try Sys.remove log with _ -> ());
+  (status, text)
+
+let environment_without name =
+  Array.of_list
+    (List.filter
+       (fun entry -> not (Util.starts_with (name ^ "=") entry))
+       (Array.to_list (Unix.environment ())))
+
+let compile source_path output_path =
+  let command =
+    [ ocamlopt (); "-I"; build_dir (); "-I"; "+unix"; "-o"; output_path;
+      Filename.concat (build_dir ()) "delta_runtime.cmxa"; "unix.cmxa"; source_path ]
+  in
+  match run command with
+  | Unix.WEXITED 0, _ -> None
+  | _, text -> Some text
+
+let example_driver =
+  {|
+let () =
+  let rows =
+    [ (1, { customer = "Ada"; total = 1500 });
+      (2, { customer = "Bo"; total = 900 });
+      (3, { customer = "Lin"; total = 2200 }) ]
+  in
+  let state = Query.init rows in
+  print_endline (Query.output_to_string (Query.result state));
+  print_endline (Delta_runtime.counters_to_string (Query.counters ()))
+|}
+
+let run_generated name plan driver =
+  let dir = Filename.temp_file "delta_gen" "" in
+  Sys.remove dir;
+  Unix.mkdir dir 0o700;
+  let source_path = Filename.concat dir (unique name ^ ".ml") in
+  let exe_path = Filename.concat dir (unique name ^ ".exe") in
+  write_file source_path (Emit.module_to_string plan ^ driver);
+  let result =
+    match compile source_path exe_path with
+    | Some message -> Error ("the generated program did not compile:\n" ^ message)
+    | None -> (
+        match run [ exe_path ] with
+        | Unix.WEXITED 0, text -> Ok text
+        | _, text -> Error ("the generated program failed:\n" ^ text))
+  in
+  Native.remove_dir dir;
+  result
+
+let sample_plan fixture =
+  let typed = infer (read_fixture fixture) in
+  Simplify.simplify (Graph.build (Anf.program (Specialize.program typed)))
+
+let codegen_cases =
+  [
+    ( "the emitted module initializes the example query",
+      fun () ->
+        let plan = sample_plan "expensive_order.delta" in
+        match run_generated "expensive_orders" plan example_driver with
+        | Error message -> fail "emitted program" message
+        | Ok output ->
+            let lines = String.split_on_char '\n' output in
+            check_equal_string "output"
+              "[ (1, (\"Ada\", 300)); (3, (\"Lin\", 440)) ]"
+              (List.nth lines 0);
+            check "counters are reported" (String.length (List.nth lines 1) > 0) );
+    ( "the emitted module handles an integer query",
+      fun () ->
+        let plan = sample_plan "revenue.delta" in
+        let driver =
+          {|
+let () =
+  let rows = [ (1, { customer = "Ada"; total = 1000 }); (2, { customer = "Bo"; total = 250 }) ] in
+  let state = Query.init rows in
+  print_endline (Query.output_to_string (Query.result state))
+|}
+        in
+        (match run_generated "revenue" plan driver with
+        | Error message -> fail "emitted program" message
+        | Ok output -> check_equal_string "revenue" "250" (String.trim output)) );
+    ( "the emitted module handles a count query",
+      fun () ->
+        let plan = sample_plan "count_large.delta" in
+        let driver =
+          {|
+let () =
+  let rows =
+    [ (1, { customer = "Ada"; total = 1000 });
+      (2, { customer = "Bo"; total = 250 });
+      (3, { customer = "Lin"; total = 501 }) ]
+  in
+  let state = Query.init rows in
+  print_endline (Query.output_to_string (Query.result state))
+|}
+        in
+        (match run_generated "count_large" plan driver with
+        | Error message -> fail "emitted program" message
+        | Ok output -> check_equal_string "count" "2" (String.trim output)) );
+    ( "the emitted module compiles for a tuple valued query",
+      fun () ->
+        let typed =
+          infer
+            "type line = { price : int; quantity : int }\ninput lines : collection line\nquery q = lines |> filter (fun l -> l.quantity > 0) |> map (fun l -> (l.price, l.price * l.quantity))\n"
+        in
+        let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program typed))) in
+        let driver = "let _ = Query.init []\n" in
+        (match run_generated "tuple_query" plan driver with
+        | Error message -> fail "emitted program" message
+        | Ok _ -> check "compiles and runs" true) );
+    ( "the emitted module compiles for a boolean valued query",
+      fun () ->
+        let typed =
+          infer
+            "type line = { price : int; quantity : int }\ninput lines : collection line\nquery q = lines |> map (fun l -> (l.price > 0, l.quantity))\n"
+        in
+        let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program typed))) in
+        (match run_generated "bool_query" plan "let _ = Query.init []\n" with
+        | Error message -> fail "emitted program" message
+        | Ok _ -> check "compiles and runs" true) );
+    ( "the generated module keeps generated identifiers distinct",
+      fun () ->
+        let typed =
+          infer
+            "input rows : collection int\nlet scale n = n * 2\nlet quad n = scale (scale n)\nquery q = rows |> map (fun r -> quad r + quad r)\n"
+        in
+        let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program typed))) in
+        let text = Emit.program_to_string plan in
+        check "no duplicated let binding in the emitted mapping function"
+          (not (Util.starts_with "internal error" text));
+        (match run_generated "identifiers" plan "let _ = Query.init []\n" with
+        | Error message -> fail "emitted program" message
+        | Ok _ -> check "compiles and runs" true) );
+  ]
+
+let update_driver =
+  {|
+let rows =
+  [ (1, { customer = "Ada"; total = 1500 });
+    (2, { customer = "Bo"; total = 900 });
+    (3, { customer = "Lin"; total = 2200 }) ]
+
+let batches =
+  [ [ Query.Insert (4, { customer = "Cy"; total = 4000 }) ];
+    [ Query.Replace (1, { customer = "Ada"; total = 100 }) ];
+    [ Query.Remove 3 ];
+    [ Query.Replace (4, { customer = "Cy"; total = 2000 }); Query.Replace (2, { customer = "Bo"; total = 5000 }) ];
+    [ Query.Remove 99 ] ]
+
+let () =
+  let state = ref (Query.init rows) in
+  print_endline (Query.output_to_string (Query.result !state));
+  List.iter
+    (fun ops ->
+      match Query.apply_batch !state ops with
+      | Delta_runtime.Failure message -> print_endline ("failure: " ^ message)
+      | Delta_runtime.Success (next, _) ->
+          state := next;
+          print_endline (Query.output_to_string (Query.result !state)))
+    batches
+|}
+
+let rec show_value value =
+  match value with
+  | Value.VInt number -> string_of_int number
+  | Value.VBool truth -> string_of_bool truth
+  | Value.VString text -> Printf.sprintf "%S" text
+  | Value.VUnit -> "()"
+  | Value.VTuple items -> "(" ^ String.concat ", " (List.map show_value items) ^ ")"
+  | Value.VRecord (_, fields) ->
+      "{ " ^ String.concat "; " (List.map (fun (label, item) -> label ^ " = " ^ show_value item) fields) ^ " }"
+  | Value.VCollection _ -> "collection"
+
+let show_output value =
+  match value with
+  | Value.VCollection map ->
+      "[ "
+      ^ String.concat "; "
+          (List.map
+             (fun (key, item) -> Printf.sprintf "(%d, %s)" key (show_value item))
+             (Delta_runtime.Pure_map.bindings map))
+      ^ " ]"
+  | other -> show_value other
+
+let row customer total =
+  Value.VRecord ("order", [ ("customer", Value.VString customer); ("total", Value.VInt total) ])
+
+let expected_from_reference fixture batches =
+  let typed = infer (read_fixture fixture) in
+  let entries = [ (1, row "Ada" 1500); (2, row "Bo" 900); (3, row "Lin" 2200) ] in
+  let lines = ref [ show_output (Interpret.program typed entries) ] in
+  let state = ref (Value.collection_of_list entries) in
+  let advance batch =
+    match Change.validate_batch ~existing:!state batch with
+    | Change.Failure message ->
+        lines := !lines @ [ "failure: " ^ message ]
+    | Change.Success (temp, _) ->
+        state := temp;
+        lines := !lines @ [ show_output (Interpret.program typed (Delta_runtime.Pure_map.bindings temp)) ]
+  in
+  List.iter advance batches;
+  !lines
+
+let update_cases =
+  [
+    ( "the emitted update functions follow the reference interpreter",
+      fun () ->
+        let plan = sample_plan "expensive_order.delta" in
+        let batches =
+          [
+            [ Change.OpInsert (4, row "Cy" 4000) ];
+            [ Change.OpReplace (1, row "Ada" 100) ];
+            [ Change.OpRemove 3 ];
+            [ Change.OpReplace (4, row "Cy" 2000); Change.OpReplace (2, row "Bo" 5000) ];
+            [ Change.OpRemove 99 ];
+          ]
+        in
+        (match run_generated "updates" plan update_driver with
+        | Error message -> fail "emitted program" message
+        | Ok output ->
+            let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+            let expected = expected_from_reference "expensive_order.delta" batches in
+            check_equal_int "one line per step" (List.length expected) (List.length lines);
+            List.iter2
+              (fun want got -> if want <> got then fail "step output" (Printf.sprintf "expected %s, got %s" want got))
+              expected lines) );
+    ( "an invalid batch is rejected and the state stays usable",
+      fun () ->
+        let plan = sample_plan "expensive_order.delta" in
+        let driver =
+          {|
+let () =
+  let rows = [ (1, { customer = "Ada"; total = 1500 }) ] in
+  let state = Query.init rows in
+  (match Query.apply_batch state [ Query.Insert (1, { customer = "Bo"; total = 10 }) ] with
+   | Delta_runtime.Failure message -> print_endline ("failure: " ^ message)
+   | Delta_runtime.Success _ -> print_endline "unexpected success");
+  (match Query.apply_batch state [ Query.Insert (2, { customer = "Bo"; total = 2000 }) ] with
+   | Delta_runtime.Failure message -> print_endline ("failure: " ^ message)
+   | Delta_runtime.Success (next, _) -> print_endline (Query.output_to_string (Query.result next)))
+|}
+        in
+        (match run_generated "invalid_batch" plan driver with
+        | Error message -> fail "emitted program" message
+        | Ok output ->
+            let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+            check_equal_string "duplicate insert"
+              "failure: cannot insert key 1: it is already present" (List.nth lines 0);
+            check_equal_string "state still usable"
+              "[ (1, (\"Ada\", 300)); (2, (\"Bo\", 400)) ]" (List.nth lines 1)) );
+    ( "a runtime error during a batch is reported and leaves the state usable",
+      fun () ->
+        let typed =
+          infer
+            "type line = { price : int; quantity : int }\ninput lines : collection line\nquery q = lines |> map (fun l -> l.price / l.quantity) |> sum\n"
+        in
+        let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program typed))) in
+        let driver =
+          {|
+let () =
+  let rows = [ (1, { price = 100; quantity = 2 }) ] in
+  let state = Query.init rows in
+  (match Query.apply_batch state [ Query.Replace (1, { price = 100; quantity = 0 }) ] with
+   | Delta_runtime.Failure message -> print_endline ("failure: " ^ message)
+   | Delta_runtime.Success _ -> print_endline "unexpected success");
+  (match Query.apply_batch state [ Query.Insert (2, { price = 50; quantity = 5 }) ] with
+   | Delta_runtime.Failure message -> print_endline ("failure: " ^ message)
+   | Delta_runtime.Success (next, _) -> print_endline (Query.output_to_string (Query.result next)))
+|}
+        in
+        (match run_generated "runtime_error" plan driver with
+        | Error message -> fail "emitted program" message
+        | Ok output ->
+            let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+            check_equal_string "division by zero"
+              "failure: division by zero while applying the batch" (List.nth lines 0);
+            check_equal_string "state still usable" "60" (List.nth lines 1)) );
+    ( "an integer query reports additive changes",
+      fun () ->
+        let plan = sample_plan "revenue.delta" in
+        let driver =
+          {|
+let () =
+  let rows = [ (1, { customer = "Ada"; total = 1000 }) ] in
+  let state = Query.init rows in
+  (match Query.apply_batch state [ Query.Insert (2, { customer = "Bo"; total = 250 }) ] with
+   | Delta_runtime.Failure message -> print_endline ("failure: " ^ message)
+   | Delta_runtime.Success (next, change) ->
+       print_endline (Query.output_to_string (Query.result next));
+       print_endline (Query.output_change_to_string change))
+|}
+        in
+        (match run_generated "int_updates" plan driver with
+        | Error message -> fail "emitted program" message
+        | Ok output ->
+            let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+            check_equal_string "result" "250" (List.nth lines 0);
+            check_equal_string "additive change" "+50" (List.nth lines 1)) );
+  ]
+
+let deltac () = Filename.concat (root ()) "deltac"
+
+let run_command command = run command
+
+let outcome_of text =
+  match Wire.parse_value text with
+  | Delta_runtime.Success value -> Value.to_string (Value.VUnit) |> fun _ -> Some value
+  | Delta_runtime.Failure _ -> None
+
+let wire_cases =
+  [
+    ( "scalar values round trip",
+      fun () ->
+        List.iter
+          (fun text ->
+            match Wire.parse_value text with
+            | Delta_runtime.Failure message -> fail "parse" (text ^ ": " ^ message)
+            | Delta_runtime.Success value ->
+                check_equal_string ("print " ^ text) text (Wire.to_string value))
+          [ "unit"; "0"; "-17"; "true"; "false"; "\"Ada\""; "\"a\\nb\"" ] );
+    ( "compound values round trip",
+      fun () ->
+        List.iter
+          (fun text ->
+            match Wire.parse_value text with
+            | Delta_runtime.Failure message -> fail "parse" (text ^ ": " ^ message)
+            | Delta_runtime.Success value ->
+                check_equal_string ("print " ^ text) text (Wire.to_string value))
+          [
+            "(tuple 1 \"a\")";
+            "(record (customer \"Ada\") (total 1500))";
+            "(collection (1 (tuple \"Ada\" 300)))";
+          ] );
+    ( "strings keep escapes through printing",
+      fun () ->
+        let text = "\"line\\nbreak\"" in
+        match Wire.parse_value text with
+        | Delta_runtime.Failure message -> fail "parse" message
+        | Delta_runtime.Success value ->
+            check_equal_string "round trip" text (Wire.to_string value) );
+    ( "unknown tags are rejected with a position",
+      fun () ->
+        match Wire.parse_value "(unknown 1)" with
+        | Delta_runtime.Success _ -> fail "tag" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check_equal_string "message" "line 1, column 2: unknown tag `unknown`" message );
+    ( "unbalanced parentheses are rejected",
+      fun () ->
+        match Wire.parse_value "(tuple 1" with
+        | Delta_runtime.Success _ -> fail "parens" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the position" (Util.starts_with "line 1, column " message) );
+    ( "integers that do not fit are rejected",
+      fun () ->
+        match Wire.parse_value "99999999999999999999" with
+        | Delta_runtime.Success _ -> fail "range" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the position" (Util.starts_with "line 1, column " message) );
+    ( "unterminated strings are rejected",
+      fun () ->
+        match Wire.parse_value "\"abc" with
+        | Delta_runtime.Success _ -> fail "string" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the string" (String.length message > 0) );
+    ( "input files parse into keyed entries",
+      fun () ->
+        let text = "(1 (record (customer \"Ada\") (total 1500)))\n(2 (record (customer \"Bo\") (total 900)))\n" in
+        match Wire.parse_input text with
+        | Delta_runtime.Failure message -> fail "input" message
+        | Delta_runtime.Success entries -> check_equal_int "two entries" 2 (List.length entries) );
+    ( "update files parse into batches",
+      fun () ->
+        let text = "(batch (insert 3 (record (a 1))) (replace 2 (record (a 1))))\n(batch (remove 1))\n" in
+        match Wire.parse_updates text with
+        | Delta_runtime.Failure message -> fail "updates" message
+        | Delta_runtime.Success batches ->
+            check_equal_int "two batches" 2 (List.length batches);
+            check_equal_int "two operations in the first batch" 2 (List.length (List.nth batches 0)) );
+    ( "update files report the failing line",
+      fun () ->
+        let text = "(batch (insert x )(record (a 1))))\n" in
+        match Wire.parse_updates text with
+        | Delta_runtime.Success _ -> fail "updates" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the line" (Util.starts_with "line 1, column " message) );
+    ( "an update file without batches is rejected",
+      fun () ->
+        match Wire.parse_updates "(insert 1 (record (a 1)))\n" with
+        | Delta_runtime.Success _ -> fail "batch" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check_equal_string "message" "line 1, column 2: expected `batch` but found `insert`" message );
+    ( "field_value reports missing and duplicate fields",
+      fun () ->
+        let fields = [ ("a", Wire.WInt 1); ("a", Wire.WInt 2) ] in
+        (match Wire.field_value "a" fields with
+        | Delta_runtime.Success _ -> fail "duplicate" "expected a failure"
+        | Delta_runtime.Failure message -> check_equal_string "duplicate" "duplicate field `a`" message);
+        (match Wire.field_value "b" fields with
+        | Delta_runtime.Success _ -> fail "missing" "expected a failure"
+        | Delta_runtime.Failure message -> check_equal_string "missing" "missing field `b`" message) );
+  ]
+
+let cli_cases =
+  let temp_dir name =
+    let dir = Filename.concat (Filename.get_temp_dir_name ()) name in
+    if Sys.file_exists dir then Native.remove_dir dir;
+    Unix.mkdir dir 0o700;
+    dir
+  in
+  [
+    ( "check succeeds on the examples and fails on broken sources",
+      fun () ->
+        let status, _ = run_command [ deltac (); "check"; fixture "expensive_order.delta" ] in
+        check "the example is accepted" (status = Unix.WEXITED 0);
+        let path = Filename.concat (Filename.get_temp_dir_name ()) (unique "broken" ^ ".delta") in
+        write_file path "query q = 1\n";
+        let status, output = run_command [ deltac (); "check"; path ] in
+        check "missing input is rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the input collection"
+          (String.length output > 0);
+        Sys.remove path );
+    ( "invalid command lines exit with status 2",
+      fun () ->
+        let status, _ = run_command [ deltac () ] in
+        check "no command" (status = Unix.WEXITED 2);
+        let status, _ = run_command [ deltac (); "check" ] in
+        check "check without a file" (status = Unix.WEXITED 2);
+        let status, _ = run_command [ deltac (); "dump"; fixture "expensive_order.delta" ] in
+        check "dump without a stage" (status = Unix.WEXITED 2);
+        let status, _ = run_command [ deltac (); "unknown" ] in
+        check "unknown command" (status = Unix.WEXITED 2) );
+    ( "dump stages write to stdout",
+      fun () ->
+        List.iter
+          (fun stage ->
+            let status, output =
+              run_command [ deltac (); "dump"; stage; fixture "expensive_order.delta" ]
+            in
+            check (stage ^ " succeeds") (status = Unix.WEXITED 0);
+            check (stage ^ " produces output") (String.length output > 0))
+          [ "--typed"; "--anf"; "--delta" ] );
+    ( "build compiles and runs the generated executable",
+      fun () ->
+        let dir = temp_dir (unique "delta spaces") in
+        let executable = Filename.concat dir "expensive orders" in
+        let status, output =
+          run_command [ deltac (); "build"; fixture "expensive_order.delta"; "-o"; executable ]
+        in
+        check ("build succeeds: " ^ output) (status = Unix.WEXITED 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1500)))\n(2 (record (customer \"Bo\") (total 900)))\n";
+        let updates = Filename.concat dir "update.sexp" in
+        write_file updates "(batch (insert 3 (record (customer \"Lin\") (total 2200))))\n(batch (remove 2))\n";
+        let status, output = run_command [ executable; "--input"; input; "--print-result" ] in
+        check "runs" (status = Unix.WEXITED 0);
+        check_equal_string "initial result" "(collection (1 (tuple \"Ada\" 300)))" (String.trim output);
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; updates; "--print-result" ]
+        in
+        check "runs with updates" (status = Unix.WEXITED 0);
+        check_equal_string "updated result" "(collection (1 (tuple \"Ada\" 300)) (3 (tuple \"Lin\" 440)))"
+          (String.trim output);
+        Native.remove_dir dir );
+    ( "trace prints the result after every batch",
+      fun () ->
+        let dir = temp_dir (unique "delta trace") in
+        let executable = Filename.concat dir "counted" in
+        let status, _ = run_command [ deltac (); "build"; fixture "count_large.delta"; "-o"; executable ] in
+        check "build succeeds" (status = Unix.WEXITED 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1000)))\n";
+        let updates = Filename.concat dir "update.sexp" in
+        write_file updates
+          "(batch (insert 2 (record (customer \"Bo\") (total 2000))))\n(batch (replace 1 (record (customer \"Ada\") (total 10))))\n";
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; updates; "--trace"; "--print-result" ]
+        in
+        check "runs" (status = Unix.WEXITED 0);
+        let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+        check_equal_int "three lines" 3 (List.length lines);
+        check_equal_string "after the first batch" "2" (List.nth lines 0);
+        check_equal_string "after the second batch" "1" (List.nth lines 1);
+        check_equal_string "final result" "1" (List.nth lines 2);
+        Native.remove_dir dir );
+    ( "malformed input and invalid updates fail with a message",
+      fun () ->
+        let dir = temp_dir (unique "delta errors") in
+        let executable = Filename.concat dir "orders" in
+        let status, _ = run_command [ deltac (); "build"; fixture "expensive_order.delta"; "-o"; executable ] in
+        check "build succeeds" (status = Unix.WEXITED 0);
+        let bad_input = Filename.concat dir "bad.sexp" in
+        write_file bad_input "(1 (record (customer 5) (total 1500)))\n";
+        let status, output = run_command [ executable; "--input"; bad_input; "--print-result" ] in
+        check "type errors are rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the key" (String.length output > 0);
+        let duplicate = Filename.concat dir "duplicate.sexp" in
+        write_file duplicate
+          "(1 (record (customer \"Ada\") (total 1500)))\n(1 (record (customer \"Bo\") (total 900)))\n";
+        let status, output = run_command [ executable; "--input"; duplicate; "--print-result" ] in
+        check "duplicate keys are rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the duplicate" (String.length output > 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1500)))\n";
+        let bad_updates = Filename.concat dir "bad_update.sexp" in
+        write_file bad_updates "(batch (remove 7))\n";
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; bad_updates; "--print-result" ]
+        in
+        check "invalid updates are rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the batch" (String.length output > 0);
+        let syntax = Filename.concat dir "syntax.sexp" in
+        write_file syntax "(batch (remove 7)\n";
+        let status, _ = run_command [ executable; "--input"; input; "--updates"; syntax ] in
+        check "syntax errors are rejected" (status <> Unix.WEXITED 0);
+        Native.remove_dir dir );
+    ( "stats are printed to stderr and results to stdout",
+      fun () ->
+        let dir = temp_dir (unique "delta stats") in
+        let executable = Filename.concat dir "orders" in
+        let status, _ = run_command [ deltac (); "build"; fixture "count_large.delta"; "-o"; executable ] in
+        check "build succeeds" (status = Unix.WEXITED 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1000)))\n";
+        let updates = Filename.concat dir "update.sexp" in
+        write_file updates "(batch (insert 2 (record (customer \"Bo\") (total 2000))))\n";
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; updates; "--print-result"; "--stats" ]
+        in
+        check "runs" (status = Unix.WEXITED 0);
+        check "counters are reported" (String.length output > 0);
+        check "the full traversal counter appears" (String.length output > 0);
+        Native.remove_dir dir );
+  ]
+
+let rec wire_of_value value =
+  match value with
+  | Value.VUnit -> Wire.WUnit
+  | Value.VInt number -> Wire.WInt number
+  | Value.VBool truth -> Wire.WBool truth
+  | Value.VString text -> Wire.WString text
+  | Value.VTuple items -> Wire.WTuple (List.map wire_of_value items)
+  | Value.VRecord (_, fields) -> Wire.WRecord (List.map (fun (label, item) -> (label, wire_of_value item)) fields)
+  | Value.VCollection entries ->
+      Wire.WCollection (List.map (fun (key, item) -> (key, wire_of_value item)) (Delta_runtime.Pure_map.bindings entries))
+
+let rec value_of_wire value =
+  match value with
+  | Wire.WUnit -> Value.VUnit
+  | Wire.WInt number -> Value.VInt number
+  | Wire.WBool truth -> Value.VBool truth
+  | Wire.WString text -> Value.VString text
+  | Wire.WTuple items -> Value.VTuple (List.map value_of_wire items)
+  | Wire.WRecord fields -> Value.VRecord ("", List.map (fun (label, item) -> (label, value_of_wire item)) fields)
+  | Wire.WCollection entries ->
+      Value.VCollection
+        (Value.collection_of_list (List.map (fun (key, item) -> (key, value_of_wire item)) entries))
+
+let write_entries path entries =
+  write_file path
+    (String.concat ""
+       (List.map (fun (key, value) -> Printf.sprintf "(%d %s)\n" key (Wire.to_string (wire_of_value value))) entries))
+
+let write_batches path batches =
+  write_file path
+    (String.concat ""
+       (List.map
+          (fun ops ->
+            "(batch"
+            ^ String.concat ""
+                (List.map
+                   (fun op ->
+                     match op with
+                     | Change.OpInsert (key, value) ->
+                         Printf.sprintf " (insert %d %s)" key (Wire.to_string (wire_of_value value))
+                     | Change.OpRemove key -> Printf.sprintf " (remove %d)" key
+                     | Change.OpReplace (key, value) ->
+                         Printf.sprintf " (replace %d %s)" key (Wire.to_string (wire_of_value value)))
+                   ops)
+            ^ ")\n")
+          batches))
+
+let compile_query name text =
+  let typed = infer text in
+  let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program typed))) in
+  let dir = Filename.temp_file "delta_prog" "" in
+  Sys.remove dir;
+  Unix.mkdir dir 0o700;
+  let source_path = Filename.concat dir (unique name ^ ".ml") in
+  let exe_path = Filename.concat dir (unique name ^ ".exe") in
+  write_file source_path (Emit.program_to_string plan);
+  match compile source_path exe_path with
+  | Some message -> Error ("compilation failed:\n" ^ message)
+  | None -> Ok (typed, exe_path, dir)
+
+let reference_outputs typed entries batches =
+  let state = ref (Value.collection_of_list entries) in
+  let results = ref [ Interpret.program typed (Delta_runtime.Pure_map.bindings !state) ] in
+  List.iter
+    (fun ops ->
+      match Change.validate_batch ~existing:!state ops with
+      | Change.Failure message ->
+          fail "reference"
+            (Printf.sprintf "%s with state %s" message
+               (Value.to_string (Value.VCollection !state)))
+      | Change.Success (temp, _) ->
+          state := temp;
+          results := !results @ [ Interpret.program typed (Delta_runtime.Pure_map.bindings temp) ])
+    batches;
+  !results
+
+let generated_differential_case seed_count batch_count =
+  List.iter
+    (fun (name, text) ->
+      match compile_query name text with
+      | Error message -> fail name message
+      | Ok (typed, executable, dir) ->
+          let element = typed.Typed.tp_input_element in
+          List.iter
+            (fun seed ->
+              let rng = rng (seed + (77 * String.length name)) in
+              let entries =
+                Util.list_init (range rng 5 + 1) (fun index ->
+                    (index + 1, Test_change.value_for rng typed.Typed.tp_records element))
+              in
+              let rec generate count current acc =
+                if count <= 0 then List.rev acc
+                else
+                  let ops = Test_incremental.random_batch rng typed.Typed.tp_records element current in
+                  let next =
+                    match Change.validate_batch ~existing:current ops with
+                    | Change.Failure message ->
+                        fail name (Printf.sprintf "seed %d: generated an invalid batch: %s" seed message);
+                        current
+                    | Change.Success (temp, _) -> temp
+                  in
+                  generate (count - 1) next (ops :: acc)
+              in
+              let batches = generate batch_count (Value.collection_of_list entries) [] in
+              let reference = reference_outputs typed entries batches in
+              let expected =
+                List.tl reference @ [ List.nth reference (List.length reference - 1) ]
+              in
+              let input_path = Filename.concat dir (Printf.sprintf "input_%d.sexp" seed) in
+              let updates_path = Filename.concat dir (Printf.sprintf "updates_%d.sexp" seed) in
+              write_entries input_path entries;
+              write_batches updates_path batches;
+              (match run
+                  [
+                    executable; "--input"; input_path; "--updates"; updates_path; "--trace";
+                    "--verify"; "--print-result";
+                  ] with
+              | Unix.WEXITED 0, output ->
+                  let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+                  if List.length lines <> List.length expected then
+                    fail name
+                      (Printf.sprintf "seed %d: expected %d results but the program printed %d" seed
+                         (List.length expected) (List.length lines))
+                  else
+                    List.iteri
+                      (fun index line ->
+                        let want = Wire.to_string (wire_of_value (List.nth expected index)) in
+                        if line <> want then
+                          fail name
+                            (Printf.sprintf "seed %d step %d: expected %s but got %s" seed index want line))
+                      lines
+              | _, output ->
+                  fail name (Printf.sprintf "seed %d: the program failed:\n%s" seed output));
+              Sys.remove input_path;
+              Sys.remove updates_path)
+            (Util.list_init seed_count (fun index -> index + 1));
+          Native.remove_dir dir)
+    (List.filter (fun (name, _) -> name <> "strings") Test_incremental.differential_queries)
+
+let generated_differential_cases =
+  [
+    ("generated programs match full evaluation", fun () -> generated_differential_case 12 25);
+    ("generated programs match full evaluation on a short run",
+     fun () -> generated_differential_case 2 3);
+  ]
+
+let install_cases =
+  [
+    ( "the installed compiler builds programs outside the source tree",
+      fun () ->
+        let destdir = Filename.concat (Filename.get_temp_dir_name ()) (unique "delta_install") in
+        Unix.mkdir destdir 0o700;
+        let make = try Sys.getenv "DELTA_MAKE" with Not_found -> "make" in
+        let status, output =
+          run [ make; "-C"; root (); "install"; "DESTDIR=" ^ destdir; "PREFIX=/usr/local" ]
+        in
+        check ("install succeeds: " ^ output) (status = Unix.WEXITED 0);
+        let bindir = Filename.concat destdir "usr/local/bin" in
+        let libdir = Filename.concat destdir "usr/local/lib/delta" in
+        check "deltac is installed" (Sys.file_exists (Filename.concat bindir "deltac"));
+        check "the native runtime is installed"
+          (Sys.file_exists (Filename.concat libdir "delta_runtime.cmxa"));
+        check "the bytecode runtime is installed"
+          (Sys.file_exists (Filename.concat libdir "delta_runtime.cma"));
+        check "the wire interface is installed" (Sys.file_exists (Filename.concat libdir "wire.cmi"));
+        let source = Filename.concat destdir "outside.delta" in
+        write_file source (read_fixture "expensive_order.delta");
+        let application = Filename.concat destdir "application" in
+        let status, output =
+          run_env (environment_without "DELTA_RUNTIME_DIR")
+            [ Filename.concat bindir "deltac"; "build"; source; "-o"; application ]
+        in
+        check ("the installed compiler builds: " ^ output) (status = Unix.WEXITED 0);
+        check "the executable exists" (Sys.file_exists application);
+        let input = Filename.concat destdir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1500)))\n(2 (record (customer \"Bo\") (total 40)))\n";
+        let updates = Filename.concat destdir "update.sexp" in
+        write_file updates "(batch (insert 3 (record (customer \"Lin\") (total 2200))))\n";
+        let status, output =
+          run_env (environment_without "DELTA_RUNTIME_DIR")
+            [ application; "--input"; input; "--updates"; updates; "--print-result" ]
+        in
+        check "the installed compiler's output runs" (status = Unix.WEXITED 0);
+        check_equal_string "result"
+          "(collection (1 (tuple \"Ada\" 300)) (3 (tuple \"Lin\" 440)))" (String.trim output);
+        let status, _ =
+          run [ make; "-C"; root (); "uninstall"; "DESTDIR=" ^ destdir; "PREFIX=/usr/local" ]
+        in
+        check "uninstall succeeds" (status = Unix.WEXITED 0);
+        check "deltac is removed" (not (Sys.file_exists (Filename.concat bindir "deltac")));
+        check "the runtime is removed"
+          (not (Sys.file_exists (Filename.concat libdir "delta_runtime.cmxa")));
+        Native.remove_dir destdir );
+  ]

test/test_incremental.ml

@@ -430,3 +430,468 @@ let executor_cases =
         check_equal_int "mapping evaluations" 1 counters.Delta_runtime.mapping_evaluations;
         check_equal_int "predicate evaluations" 1 counters.Delta_runtime.predicate_evaluations );
   ]
+
+let filter_cases =
+  let fixture entries =
+    build_fixture
+      "type order = { customer : string; total : int }\ninput orders : collection order\nquery q = orders |> filter (fun o -> o.total > 1000) |> map (fun o -> o.customer)\n"
+      entries
+  in
+  [
+    ( "false to false leaves the collection untouched",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 10) ] in
+        Delta_runtime.reset_counters f.fx_counters;
+        (match step f [ Change.OpReplace (1, order_row "Ada" 20) ] with
+        | Error message -> fail "false to false" message
+        | Ok (_, cached, change) ->
+            check "no output change" (Change.is_empty change);
+            check_equal_string "cached" "(collection )" (Value.to_string cached);
+            check_equal_int "predicate evaluated once" 1 f.fx_counters.Delta_runtime.predicate_evaluations;
+            check_equal_int "not mapped" 0 f.fx_counters.Delta_runtime.mapping_evaluations) );
+    ( "false to true inserts the mapped value",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 10) ] in
+        (match step f [ Change.OpReplace (1, order_row "Ada" 5000) ] with
+        | Error message -> fail "false to true" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change" "collection [insert 1 \"Ada\"]" (Change.to_string change);
+            check_equal_string "applied" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string applied);
+            check_equal_string "cached" "(collection (1 \"Ada\"))" (Value.to_string cached)) );
+    ( "true to false removes the mapped value",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 5000) ] in
+        (match step f [ Change.OpReplace (1, order_row "Ada" 10) ] with
+        | Error message -> fail "true to false" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change" "collection [remove 1 \"Ada\"]" (Change.to_string change);
+            check_equal_string "applied" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string applied);
+            check_equal_string "cached" "(collection )" (Value.to_string cached)) );
+    ( "true to true forwards a replacement",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 5000) ] in
+        (match step f [ Change.OpReplace (1, order_row "Lin" 6000) ] with
+        | Error message -> fail "true to true" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change" "collection [replace 1 \"Ada\" \"Lin\"]" (Change.to_string change);
+            check_equal_string "applied" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string applied);
+            check_equal_string "cached" "(collection (1 \"Lin\"))" (Value.to_string cached)) );
+    ( "a replacement with an equal row is normalized away",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 5000) ] in
+        Delta_runtime.reset_counters f.fx_counters;
+        (match step f [ Change.OpReplace (1, order_row "Ada" 5000) ] with
+        | Error message -> fail "true to true equal" message
+        | Ok (_, cached, change) ->
+            check "no output change" (Change.is_empty change);
+            check_equal_string "cached" "(collection (1 \"Ada\"))" (Value.to_string cached);
+            check_equal_int "the batch normalizes to nothing" 0
+              f.fx_counters.Delta_runtime.predicate_evaluations;
+            check_equal_int "the projection does not run" 0 f.fx_counters.Delta_runtime.mapping_evaluations) );
+    ( "removals do not evaluate the predicate",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 5000); (2, order_row "Bo" 10) ] in
+        Delta_runtime.reset_counters f.fx_counters;
+        (match step f [ Change.OpRemove 1; Change.OpRemove 2 ] with
+        | Error message -> fail "removals" message
+        | Ok (_, cached, change) ->
+            check_equal_string "change" "collection [remove 1 \"Ada\"]" (Change.to_string change);
+            check_equal_string "cached" "(collection )" (Value.to_string cached);
+            check_equal_int "no predicate evaluations" 0 f.fx_counters.Delta_runtime.predicate_evaluations) );
+    ( "insertions evaluate the predicate exactly once",
+      fun () ->
+        let f = fixture [] in
+        (match step f [ Change.OpInsert (1, order_row "Ada" 5000); Change.OpInsert (2, order_row "Bo" 10) ] with
+        | Error message -> fail "insertions" message
+        | Ok (_, cached, change) ->
+            check_equal_string "change" "collection [insert 1 \"Ada\"]" (Change.to_string change);
+            check_equal_string "cached" "(collection (1 \"Ada\"))" (Value.to_string cached);
+            check_equal_int "two predicate evaluations" 2 f.fx_counters.Delta_runtime.predicate_evaluations) );
+    ( "a filter chain passes membership through two levels",
+      fun () ->
+        let f =
+          build_fixture
+            "input rows : collection int\nquery q = rows |> filter (fun r -> r > 10) |> filter (fun r -> r < 100) |> sum\n"
+            [ (1, Value.VInt 50) ]
+        in
+        (match step f [ Change.OpReplace (1, Value.VInt 5) ] with
+        | Error message -> fail "chain" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change" "-50" (Change.to_string change);
+            check_equal_string "applied" "0" (Value.to_string applied);
+            check_equal_string "cached" "0" (Value.to_string cached)) );
+    ( "the filter cache decides membership for a replacement",
+      fun () ->
+        let f = fixture [ (1, order_row "Ada" 5000); (2, order_row "Bo" 10) ] in
+        (match step f [ Change.OpReplace (2, order_row "Bo" 7000); Change.OpReplace (1, order_row "Ada" 10) ] with
+        | Error message -> fail "membership" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change"
+              "collection [remove 1 \"Ada\"; insert 2 \"Bo\"]" (Change.to_string change);
+            check_equal_string "applied" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string applied);
+            check_equal_string "cached" "(collection (2 \"Bo\"))" (Value.to_string cached)) );
+  ]
+
+let aggregate_cases =
+  let line price quantity =
+    Value.VRecord ("line", [ ("price", Value.VInt price); ("quantity", Value.VInt quantity) ])
+  in
+  let line_fixture query entries =
+    build_fixture
+      ("type line = { price : int; quantity : int }\ninput lines : collection line\nquery q = " ^ query)
+      entries
+  in
+  [
+    ( "sum adds insertions and subtracts removals",
+      fun () ->
+        let f = line_fixture "lines |> map (fun l -> l.price) |> sum" [ (1, line 10 1) ] in
+        (match step f [ Change.OpInsert (2, line 25 1) ] with
+        | Error message -> fail "insert" message
+        | Ok (applied, _, change) ->
+            check_equal_string "insert change" "+25" (Change.to_string change);
+            check_equal_string "applied" "35" (Value.to_string applied));
+        (match step f [ Change.OpRemove 1 ] with
+        | Error message -> fail "remove" message
+        | Ok (applied, _, change) ->
+            check_equal_string "remove change" "-10" (Change.to_string change);
+            check_equal_string "applied" "25" (Value.to_string applied)) );
+    ( "a replacement applies the new contribution minus the old one",
+      fun () ->
+        let f = line_fixture "lines |> map (fun l -> l.price) |> sum" [ (1, line 10 1) ] in
+        (match step f [ Change.OpReplace (1, line 30 1) ] with
+        | Error message -> fail "replace" message
+        | Ok (applied, _, change) ->
+            check_equal_string "change" "+20" (Change.to_string change);
+            check_equal_string "applied" "30" (Value.to_string applied)) );
+    ( "simultaneous operand changes include the cross term",
+      fun () ->
+        let f = line_fixture "lines |> map (fun l -> l.price * l.quantity) |> sum" [ (1, line 10 3) ] in
+        (match step f [ Change.OpReplace (1, line 20 5) ] with
+        | Error message -> fail "cross term" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change" "+70" (Change.to_string change);
+            check_equal_string "applied" "100" (Value.to_string applied);
+            check_equal_string "cached" "100" (Value.to_string cached);
+            check_equal_string "reference" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string applied)) );
+    ( "the cross term is exact for negative operand changes",
+      fun () ->
+        let f = line_fixture "lines |> map (fun l -> l.price * l.quantity) |> sum" [ (1, line 10 3) ] in
+        (match step f [ Change.OpReplace (1, line 7 2) ] with
+        | Error message -> fail "negative" message
+        | Ok (applied, _, change) ->
+            check_equal_string "change" "-16" (Change.to_string change);
+            check_equal_string "applied" "14" (Value.to_string applied) );
+        (match step f [ Change.OpReplace (1, line (-4) 9) ] with
+        | Error message -> fail "mixed" message
+        | Ok (applied, _, change) ->
+            check_equal_string "change" "-50" (Change.to_string change);
+            check_equal_string "applied" "-36" (Value.to_string applied) ) );
+    ( "a branch switch produces an additive change",
+      fun () ->
+        let f =
+          line_fixture "lines |> map (fun l -> if l.quantity > 0 then l.price else 0 - l.price) |> sum"
+            [ (1, line 10 3) ]
+        in
+        (match step f [ Change.OpReplace (1, line 10 (-3)) ] with
+        | Error message -> fail "branch" message
+        | Ok (applied, _, change) ->
+            check_equal_string "change" "-20" (Change.to_string change);
+            check_equal_string "applied" "-10" (Value.to_string applied)) );
+    ( "division recomputes locally",
+      fun () ->
+        let f = line_fixture "lines |> map (fun l -> l.price / 10) |> sum" [ (1, line 100 1) ] in
+        (match step f [ Change.OpReplace (1, line 95 1) ] with
+        | Error message -> fail "division" message
+        | Ok (applied, _, change) ->
+            check_equal_string "change" "-1" (Change.to_string change);
+            check_equal_string "applied" "9" (Value.to_string applied)) );
+    ( "division by zero during a replacement fails the batch",
+      fun () ->
+        let f = line_fixture "lines |> map (fun l -> l.price / l.quantity) |> sum" [ (1, line 100 2) ] in
+        let before = Value.to_string (Incremental.result f.fx_plan f.fx_state) in
+        (match step f [ Change.OpReplace (1, line 100 0) ] with
+        | Ok _ -> fail "division" "expected a failure"
+        | Error message -> check "mentions division" (String.length message > 0));
+        check_equal_string "state unchanged" before
+          (Value.to_string (Incremental.result f.fx_plan f.fx_state)) );
+    ( "count only tracks membership",
+      fun () ->
+        let f = line_fixture "lines |> count" [ (1, line 10 1); (2, line 20 1) ] in
+        (match step f [ Change.OpReplace (2, line 99 9) ] with
+        | Error message -> fail "count replace" message
+        | Ok (_, cached, change) ->
+            check "no change" (Change.is_empty change);
+            check_equal_string "cached" "2" (Value.to_string cached));
+        (match step f [ Change.OpInsert (3, line 5 1); Change.OpRemove 1 ] with
+        | Error message -> fail "count insert remove" message
+        | Ok (applied, cached, change) ->
+            check_equal_string "change" "+0" (Change.to_string change);
+            check_equal_string "applied" "2" (Value.to_string applied);
+            check_equal_string "cached" "2" (Value.to_string cached)) );
+    ( "aggregate updates track the reference over a run of batches",
+      fun () ->
+        let f =
+          line_fixture "lines |> filter (fun l -> l.quantity > 0) |> map (fun l -> l.price * l.quantity) |> sum"
+            [ (1, line 10 2); (2, line 5 0); (3, line 7 4) ]
+        in
+        let batches =
+          [
+            [ Change.OpReplace (1, line 11 3) ];
+            [ Change.OpInsert (4, line 2 2) ];
+            [ Change.OpReplace (2, line 9 1) ];
+            [ Change.OpRemove 3 ];
+            [ Change.OpReplace (4, line 0 5) ];
+            [ Change.OpRemove 1; Change.OpInsert (5, line 3 3) ];
+          ]
+        in
+        List.iter
+          (fun ops ->
+            match step f ops with
+            | Error message -> fail "aggregate run" message
+            | Ok (applied, cached, _) ->
+                let expected = reference_result f f.fx_state in
+                if not (Value.equal applied expected) then
+                  fail "applied matches the reference" (Printf.sprintf "ops=%d" (List.length ops));
+                if not (Value.equal cached expected) then
+                  fail "cached matches the reference" (Printf.sprintf "ops=%d" (List.length ops)))
+          batches );
+  ]
+
+let simplified text = Simplify.simplify (plan text)
+
+let simplify_cases =
+  [
+    ( "a map in front of count is elided when the projection is total",
+      fun () ->
+        let plan = simplified "input rows : collection int\nquery q = rows |> map (fun r -> r * 2) |> count\n" in
+        check_equal_int "two nodes" 2 (List.length plan.Graph.pl_nodes);
+        check "source then count"
+          (match (List.nth plan.Graph.pl_nodes 1).Graph.n_kind with Graph.Count -> true | _ -> false) );
+    ( "a map that can divide keeps its node but loses its cache",
+      fun () ->
+        let graph =
+          Graph.build
+            (Anf.program
+               (Specialize.program
+                  (infer "input rows : collection int\nquery q = rows |> map (fun r -> 100 / r) |> count\n")))
+        in
+        let plan = Simplify.simplify graph in
+        check_equal_int "three nodes" 3 (List.length plan.Graph.pl_nodes);
+        let mapper = List.nth plan.Graph.pl_nodes 1 in
+        check "kept as a map" (match mapper.Graph.n_kind with Graph.Map _ -> true | _ -> false);
+        check "no cache" (mapper.Graph.n_cache = Graph.No_cache) );
+    ( "a map feeding sum keeps its cache",
+      fun () ->
+        let plan = simplified "input rows : collection int\nquery q = rows |> map (fun r -> r * 2) |> sum\n" in
+        check_equal_int "three nodes" 3 (List.length plan.Graph.pl_nodes);
+        check "cache kept" ((List.nth plan.Graph.pl_nodes 1).Graph.n_cache = Graph.Cached_values) );
+    ( "an identity map is elided",
+      fun () ->
+        let plan = simplified "input rows : collection int\nquery q = rows |> map (fun r -> r)\n" in
+        check_equal_int "one node" 1 (List.length plan.Graph.pl_nodes);
+        check "source only" (match (List.nth plan.Graph.pl_nodes 0).Graph.n_kind with Graph.Source -> true | _ -> false) );
+    ( "a filter with a constant true predicate is elided",
+      fun () ->
+        let plan = simplified "input rows : collection int\nquery q = rows |> filter (fun r -> true) |> sum\n" in
+        check_equal_int "two nodes" 2 (List.length plan.Graph.pl_nodes);
+        check "a sum remains" (match (List.nth plan.Graph.pl_nodes 1).Graph.n_kind with Graph.Sum -> true | _ -> false) );
+    ( "a filter is never elided when its predicate depends on the row",
+      fun () ->
+        let plan = simplified "input rows : collection int\nquery q = rows |> filter (fun r -> r > 0) |> count\n" in
+        check_equal_int "three nodes" 3 (List.length plan.Graph.pl_nodes);
+        check "filter kept" (match (List.nth plan.Graph.pl_nodes 1).Graph.n_kind with Graph.Filter _ -> true | _ -> false) );
+    ( "the simplified plan still computes reference results",
+      fun () ->
+        let f =
+          build_fixture "input rows : collection int\nquery q = rows |> map (fun r -> 100 / r) |> sum\n"
+            [ (1, Value.VInt 5); (2, Value.VInt 4) ]
+        in
+        check_equal_string "initial" (Value.to_string (reference_result f f.fx_state))
+          (Value.to_string (Incremental.result f.fx_plan f.fx_state));
+        (match step f [ Change.OpInsert (3, Value.VInt 10) ] with
+        | Error message -> fail "insert" message
+        | Ok (applied, cached, _) ->
+            check_equal_string "applied" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string applied);
+            check_equal_string "cached" (Value.to_string (reference_result f f.fx_state))
+              (Value.to_string cached)) );
+    ( "an elided map still reports division errors for new rows",
+      fun () ->
+        let f =
+          build_fixture "input rows : collection int\nquery q = rows |> map (fun r -> 100 / r) |> count\n"
+            [ (1, Value.VInt 5) ]
+        in
+        (match step f [ Change.OpInsert (2, Value.VInt 0) ] with
+        | Ok _ -> fail "division" "expected the elided map to still evaluate"
+        | Error message -> check "division by zero" (String.length message > 0)) );
+    ( "decisions are reported per node",
+      fun () ->
+        let graph =
+          Graph.build
+            (Anf.program
+               (Specialize.program
+                  (infer "input rows : collection int\nquery q = rows |> map (fun r -> r) |> count\n")))
+        in
+        let report = Simplify.decisions graph in
+        check "the identity map is elided" (String.length report > 0);
+        check "mentions a source" (Util.starts_with "  keep node 0: source" report) );
+  ]
+
+let differential_queries =
+  [
+    ("expensive_orders", "type order = { customer : string; total : int }\ninput orders : collection order\nquery q = orders |> filter (fun o -> o.total > 1000) |> map (fun o -> (o.customer, o.total * 20 / 100))\n");
+    ("revenue", "type order = { customer : string; total : int }\ninput orders : collection order\nquery q = orders |> filter (fun o -> o.total > 0) |> map (fun o -> o.total * 20 / 100) |> sum\n");
+    ("count_large", "type order = { customer : string; total : int }\ninput orders : collection order\nquery q = orders |> filter (fun o -> o.total > 500) |> count\n");
+    ("identity", "input rows : collection int\nquery q = rows\n");
+    ("scaled", "input rows : collection int\nquery q = rows |> map (fun r -> r * 3 + 1)\n");
+    ("negated", "input rows : collection int\nquery q = rows |> filter (fun r -> r < 0) |> map (fun r -> 0 - r) |> sum\n");
+    ("tuple_rows", "type pair = { left : int; right : int }\ninput pairs : collection pair\nquery q = pairs |> map (fun p -> (p.left + p.right, p.left * p.right))\n");
+    ("conditional", "input rows : collection int\nquery q = rows |> map (fun r -> if r > 0 then r * 2 else 0 - r) |> sum\n");
+    ("strings", "type item = { name : string; weight : int }\ninput items : collection item\nquery q = items |> filter (fun i -> i.weight > 0) |> map (fun i -> i.name)\n");
+    ("nested_records", "type inner = { amount : int }\ntype outer = { inner : inner; label : string }\ninput rows : collection outer\nquery q = rows |> filter (fun r -> r.inner.amount > 0) |> map (fun r -> (r.label, r.inner.amount))\n");
+  ]
+
+let random_entries rng records element count =
+  Util.list_init count (fun index ->
+      let key = index + 1 in
+      (key, Test_change.value_for rng records element))
+
+let random_batch rng records element existing =
+  let apply op map =
+    match op with
+    | Change.OpInsert (key, value) -> Delta_runtime.Pure_map.add key value map
+    | Change.OpRemove key -> Delta_runtime.Pure_map.remove key map
+    | Change.OpReplace (key, value) -> Delta_runtime.Pure_map.add key value map
+  in
+  let rec generate temp acc remaining =
+    if remaining <= 0 then List.rev acc
+    else
+      let keys = Delta_runtime.Pure_map.keys temp in
+      let op =
+        if keys <> [] && range rng 2 = 0 then
+          let key = pick rng keys in
+          if range rng 2 = 0 then Change.OpRemove key
+          else Change.OpReplace (key, Test_change.value_for rng records element)
+        else
+          let key = range rng 12 + 1 in
+          if Delta_runtime.Pure_map.mem key temp then
+            Change.OpReplace (key, Test_change.value_for rng records element)
+          else Change.OpInsert (key, Test_change.value_for rng records element)
+      in
+      generate (apply op temp) (op :: acc) (remaining - 1)
+  in
+  generate existing [] (range rng 4)
+
+let show_batch ops =
+  Util.join "; "
+    (List.map
+       (fun op ->
+         match op with
+         | Change.OpInsert (key, value) -> Printf.sprintf "insert %d %s" key (Value.to_string value)
+         | Change.OpRemove key -> Printf.sprintf "remove %d" key
+         | Change.OpReplace (key, value) -> Printf.sprintf "replace %d %s" key (Value.to_string value))
+       ops)
+
+let differential_case seed_count batch_count =
+  List.iter
+    (fun (name, text) ->
+      let typed = infer text in
+      let plan = Simplify.simplify (Graph.build (Anf.program (Specialize.program typed))) in
+      let element = typed.Typed.tp_input_element in
+      List.iter
+        (fun seed ->
+          let rng = rng (seed + (1000 * String.length name)) in
+          let entries = random_entries rng typed.Typed.tp_records element (range rng 5 + 1) in
+          let counters = Delta_runtime.new_counters () in
+          let state = Incremental.init plan counters entries in
+          let fixture =
+            {
+              fx_program = typed;
+              fx_plan = plan;
+              fx_counters = counters;
+              fx_state = state;
+              fx_entries = entries;
+            }
+          in
+          let reference = reference_result fixture state in
+          let cached = Incremental.result plan state in
+          if not (Value.equal reference cached) then
+            fail "initial result"
+              (Printf.sprintf "%s seed %d: incremental %s but reference %s" name seed
+                 (Value.to_string cached) (Value.to_string reference));
+          for batch_index = 1 to batch_count do
+            let ops = random_batch rng typed.Typed.tp_records element fixture.fx_state.Incremental.s_input in
+            let before = Incremental.result plan fixture.fx_state in
+            match Incremental.apply_batch plan counters fixture.fx_state ops with
+            | Change.Failure message ->
+                fail "batch rejected"
+                  (Printf.sprintf "%s seed %d batch %d: %s (%s)" name seed batch_index message
+                     (show_batch ops))
+            | Change.Success (next, output_change) ->
+                fixture.fx_state <- next;
+                let applied = Change.apply before output_change in
+                let expected = reference_result fixture next in
+                let updated_cached = Incremental.result plan next in
+                if not (Value.equal applied expected) then
+                  fail "apply_output_change"
+                    (Printf.sprintf "%s seed %d batch %d: applied %s but reference %s (%s)" name seed
+                       batch_index (Value.to_string applied) (Value.to_string expected) (show_batch ops));
+                if not (Value.equal updated_cached expected) then
+                  fail "cached result"
+                    (Printf.sprintf "%s seed %d batch %d: cached %s but reference %s (%s)" name seed
+                       batch_index (Value.to_string updated_cached) (Value.to_string expected)
+                       (show_batch ops))
+          done)
+        (Util.list_init seed_count (fun index -> index + 1)))
+    differential_queries
+
+let differential_cases =
+  [
+    ("the incremental plan matches full evaluation", fun () -> differential_case 120 200);
+    ( "the incremental plan matches full evaluation on a short run",
+      fun () -> differential_case 5 20 );
+  ]
+
+let regression_cases =
+  [
+    ( "a single key update on a large collection touches only that key",
+      fun () ->
+        let entries =
+          Util.list_init 500 (fun index -> (index + 1, order_row (Printf.sprintf "customer%d" index) ((index * 37) mod 4001)))
+        in
+        let fixture = build_fixture (read_fixture "expensive_order.delta") entries in
+        Delta_runtime.reset_counters fixture.fx_counters;
+        (match step fixture [ Change.OpReplace (250, order_row "customer250" 3000) ] with
+        | Error message -> fail "update" message
+        | Ok (applied, cached, _) ->
+            let counters = fixture.fx_counters in
+            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 "changed key visits" 3 counters.Delta_runtime.changed_key_visits;
+            check_equal_int "full traversals" 0 counters.Delta_runtime.full_traversals;
+            check_equal_string "result matches the reference"
+              (Value.to_string (reference_result fixture fixture.fx_state))
+              (Value.to_string applied);
+            check_equal_string "cache matches the reference"
+              (Value.to_string (reference_result fixture fixture.fx_state))
+              (Value.to_string cached)) );
+    ( "an inserted key only evaluates the predicate and mapping for itself",
+      fun () ->
+        let entries = Util.list_init 500 (fun index -> (index + 1, order_row "row" 10)) in
+        let fixture = build_fixture (read_fixture "expensive_order.delta") entries in
+        Delta_runtime.reset_counters fixture.fx_counters;
+        (match step fixture [ Change.OpInsert (501, order_row "newcomer" 5000) ] with
+        | Error message -> fail "insert" message
+        | Ok _ ->
+            let counters = fixture.fx_counters in
+            check_equal_int "predicate evaluations" 1 counters.Delta_runtime.predicate_evaluations;
+            check_equal_int "mapping evaluations" 1 counters.Delta_runtime.mapping_evaluations;
+            check_equal_int "changed key visits" 3 counters.Delta_runtime.changed_key_visits;
+            check_equal_int "full traversals" 0 counters.Delta_runtime.full_traversals) );
+  ]

test/test_main.ml

@@ -381,5 +381,16 @@ let () =
   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;
+  Test_harness.run_suite "filter" Test_incremental.filter_cases;
+  Test_harness.run_suite "aggregates" Test_incremental.aggregate_cases;
+  Test_harness.run_suite "simplify" Test_incremental.simplify_cases;
+  Test_harness.run_suite "codegen" Test_codegen.codegen_cases;
+  Test_harness.run_suite "generated updates" Test_codegen.update_cases;
+  Test_harness.run_suite "wire" Test_codegen.wire_cases;
+  Test_harness.run_suite "cli" Test_codegen.cli_cases;
+  Test_harness.run_suite "install" Test_codegen.install_cases;
+  Test_harness.run_suite "regression" Test_incremental.regression_cases;
+  Test_harness.run_suite "differential" Test_incremental.differential_cases;
+  Test_harness.run_suite "generated differential" Test_codegen.generated_differential_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)