Check incremental results against full evaluation

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_incremental.ml

@@ -741,3 +741,118 @@ let simplify_cases =
         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 );
+  ]

test/test_main.ml

@@ -388,5 +388,6 @@ let () =
   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 "differential" Test_incremental.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)