Maintain filter membership across replacements

test/test_incremental.ml

@@ -430,3 +430,108 @@ 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)) );
+  ]

test/test_main.ml

@@ -381,5 +381,6 @@ 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;
   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)