open Test_harness let row customer total = Value.VRecord ("order", [ ("customer", Value.VString customer); ("total", Value.VInt total) ]) let input entries = List.mapi (fun index entry -> (index + 1, entry)) entries let run fixture_name entries = let program = infer (read_fixture fixture_name) in Interpret.program program (input entries) let run_text text entries = let program = infer text in Interpret.program program (input entries) let cases = [ ( "the example query selects and projects rows", fun () -> let result = run "expensive_order.delta" [ row "Ada" 1500; row "Bo" 900; row "Lin" 2200 ] in check_equal_string "result" "(collection (1 (tuple \"Ada\" 300)) (3 (tuple \"Lin\" 440)))" (Value.to_string result) ); ( "a filter that keeps nothing yields an empty collection", fun () -> let result = run "expensive_order.delta" [ row "Bo" 10 ] in check_equal_string "empty" "(collection )" (Value.to_string result) ); ( "an empty input yields an empty collection and zero aggregates", fun () -> check_equal_string "no rows" "(collection )" (Value.to_string (run "expensive_order.delta" [])); check_equal_int "revenue" 0 (match run "revenue.delta" [] with Value.VInt total -> total | _ -> -1); check_equal_int "count" 0 (match run "count_large.delta" [] with Value.VInt total -> total | _ -> -1) ); ( "revenue sums the taxed totals", fun () -> let result = run "revenue.delta" [ row "Ada" 1000; row "Bo" 250 ] in check_equal_string "revenue" "250" (Value.to_string result) ); ( "count_large counts the retained rows", fun () -> let result = run "count_large.delta" [ row "Ada" 1000; row "Bo" 250; row "Lin" 501 ] in check_equal_string "count" "2" (Value.to_string result) ); ( "negative values are preserved", fun () -> let result = run_text "input rows : collection int\nquery q = rows |> filter (fun r -> r < 0) |> sum\n" [ Value.VInt (-5); Value.VInt 3; Value.VInt (-7) ] in check_equal_string "sum" "-12" (Value.to_string result) ); ( "integer division truncates towards zero", fun () -> let result = run_text "input rows : collection int\nquery q = rows |> map (fun r -> r / 3) |> sum\n" [ Value.VInt 10; Value.VInt (-10) ] in check_equal_string "sum" "0" (Value.to_string result) ); ( "division by zero is a runtime error", fun () -> (try ignore (run_text "input rows : collection int\nquery q = rows |> map (fun r -> 100 / r) |> sum\n" [ Value.VInt 5; Value.VInt 0 ]); fail "division" "expected a runtime error" with Division_by_zero -> check "raised" true) ); ( "the mapping expression runs only where the filter keeps rows", fun () -> let result = run_text "input rows : collection int\nquery q = rows |> filter (fun r -> r > 10) |> map (fun r -> 1000 / r) |> sum\n" [ Value.VInt 5; Value.VInt 20 ] in check_equal_string "sum" "50" (Value.to_string result) ); ( "boolean operators short circuit", fun () -> let result = run_text "input rows : collection int\nquery q = rows |> count\n" [ Value.VInt 1 ] in check_equal_string "count" "1" (Value.to_string result); let result = run_text "input rows : collection int\nquery q = rows |> filter (fun r -> false && 1 / r > 0) |> count\n" [ Value.VInt 0 ] in check_equal_string "short circuit and" "0" (Value.to_string result); let result = run_text "input rows : collection int\nquery q = rows |> filter (fun r -> true || 1 / r > 0) |> count\n" [ Value.VInt 0 ] in check_equal_string "short circuit or" "1" (Value.to_string result) ); ( "map preserves keys and filter keeps the original keys", fun () -> let result = run_text "input rows : collection int\nquery q = rows |> filter (fun r -> r > 1) |> map (fun r -> r * 10)\n" [ Value.VInt 1; Value.VInt 2; Value.VInt 3 ] in check_equal_string "keys" "(collection (2 20) (3 30))" (Value.to_string result) ); ( "helpers are applied at their call sites", fun () -> let result = run_text "input rows : collection int\nlet scale n = n * 3\nlet offset n = scale n + 1\nquery q = rows |> map offset |> sum\n" [ Value.VInt 1; Value.VInt 2 ] in check_equal_string "sum" "11" (Value.to_string result) ); ( "records and tuples are compared structurally", fun () -> let result = run_text "type pair = { first : int; second : int }\ninput rows : collection pair\nquery q = rows |> filter (fun r -> r = { first = 1; second = 2 }) |> count\n" [ Value.VRecord ("pair", [ ("first", Value.VInt 1); ("second", Value.VInt 2) ]); Value.VRecord ("pair", [ ("first", Value.VInt 2); ("second", Value.VInt 1) ]) ] in check_equal_string "count" "1" (Value.to_string result) ); ( "a constant query ignores the input", fun () -> let result = run_text "input rows : collection int\nquery q = 6 * 7\n" [ Value.VInt 1 ] in check_equal_string "constant" "42" (Value.to_string result) ); ( "out of range arithmetic follows machine integers", fun () -> let result = run_text "input rows : collection int\nquery q = rows |> sum\n" [ Value.VInt max_int; Value.VInt 1 ] in check_equal_string "wrapped" (string_of_int min_int) (Value.to_string result) ); ] let plan text = Graph.build (Anf.program (Specialize.program (infer text))) let plan_of_fixture name = Graph.build (Anf.program (Specialize.program (infer (read_fixture name)))) let graph_cases = [ ( "the example plan is a source, a filter and a map", fun () -> let plan = plan_of_fixture "expensive_order.delta" in check_equal_int "three nodes" 3 (List.length plan.Graph.pl_nodes); let nodes = plan.Graph.pl_nodes in check "source first" (match (List.nth nodes 0).Graph.n_kind with Graph.Source -> true | _ -> false); check "filter second" (match (List.nth nodes 1).Graph.n_kind with Graph.Filter _ -> true | _ -> false); check "map last" (match (List.nth nodes 2).Graph.n_kind with Graph.Map _ -> true | _ -> false); check_equal_string "output" "collection (string, int)" (Types.pp plan.Graph.pl_output) ); ( "the revenue plan ends in a sum accumulator", fun () -> let plan = plan_of_fixture "revenue.delta" in let root = Graph.node_of_id plan (match plan.Graph.pl_result with Graph.Result_collection id -> id | _ -> -1) in check "sum" (match root.Graph.n_kind with Graph.Sum -> true | _ -> false); check "accumulator cache" (root.Graph.n_cache = Graph.Accumulator); check_equal_string "linear output" "int" (Types.pp plan.Graph.pl_output) ); ( "the count query counts the retained rows", fun () -> let plan = plan_of_fixture "count_large.delta" in check_equal_int "three nodes" 3 (List.length plan.Graph.pl_nodes); let root = match plan.Graph.pl_result with | Graph.Result_collection id -> Graph.node_of_id plan id | Graph.Result_scalar _ -> fail "plan" "expected a node result"; { Graph.n_id = -1; n_kind = Graph.Source; n_input = None; n_element = Types.TInt; n_cache = Graph.No_cache; n_span = Location.none } in check "count" (match root.Graph.n_kind with Graph.Count -> true | _ -> false) ); ( "collection nodes record their consumers", fun () -> let plan = plan_of_fixture "expensive_order.delta" in check_equal_string "source consumers" "1" (Util.join "," (List.map string_of_int plan.Graph.pl_consumers.(0))); check_equal_string "filter consumers" "2" (Util.join "," (List.map string_of_int plan.Graph.pl_consumers.(1))); check_equal_string "map has no consumers" "" (Util.join "," (List.map string_of_int plan.Graph.pl_consumers.(2))) ); ( "an identity query has a single source node", fun () -> let plan = plan "input rows : collection int\nquery q = rows\n" in check_equal_int "one node" 1 (List.length plan.Graph.pl_nodes); check "the result is the source" (match plan.Graph.pl_result with Graph.Result_collection 0 -> true | _ -> false) ); ( "a plan dump is deterministic", fun () -> Ident.reset (); Types.reset (); Anf.reset (); let first = Graph.dump (plan_of_fixture "expensive_order.delta") in Ident.reset (); Types.reset (); Anf.reset (); let second = Graph.dump (plan_of_fixture "expensive_order.delta") in check_equal_string "identical" first second ); ( "an integer query that only uses the input is a plain plan", fun () -> let plan = plan "input rows : collection int\nquery q = rows |> sum\n" in check_equal_int "two nodes" 2 (List.length plan.Graph.pl_nodes); check "integer output" (Types.repr plan.Graph.pl_output = Types.TInt) ); ( "a constant integer query produces no collection nodes", fun () -> let plan = plan "input rows : collection int\nquery q = 40 + 2\n" in check_equal_int "no nodes" 0 (List.length plan.Graph.pl_nodes); check "scalar result" (match plan.Graph.pl_result with Graph.Result_scalar _ -> true | _ -> false) ); ( "shared collection temporaries are reuse", fun () -> let plan = plan "input rows : collection int\nquery q = rows |> filter (fun r -> r > 0) |> map (fun r -> r + 1) |> sum\n" in check_equal_int "four nodes" 4 (List.length plan.Graph.pl_nodes) ); ] type fixture = { fx_program : Typed.program; fx_plan : Graph.plan; fx_counters : Delta_runtime.counters; mutable fx_state : Incremental.state; fx_entries : (int * Value.t) list; } let build_fixture text entries = let program = infer text in let plan = Graph.build (Anf.program (Specialize.program program)) in let counters = Delta_runtime.new_counters () in let state = Incremental.init plan counters entries in { fx_program = program; fx_plan = plan; fx_counters = counters; fx_state = state; fx_entries = entries } let fixture_entries plan state = Delta_runtime.Pure_map.bindings state.Incremental.s_input let reference_result fixture state = Interpret.program fixture.fx_program (fixture_entries fixture.fx_plan state) let reference_after fixture ops = match Change.validate_batch ~existing:fixture.fx_state.Incremental.s_input ops with | Change.Failure message -> Error message | Change.Success (temp, _) -> let program = fixture.fx_program in ignore program; Ok (Interpret.program fixture.fx_program (Delta_runtime.Pure_map.bindings temp)) let step fixture ops = let before = Incremental.result fixture.fx_plan fixture.fx_state in match Incremental.apply_batch fixture.fx_plan fixture.fx_counters fixture.fx_state ops with | Change.Failure message -> Error message | Change.Success (state, change) -> let applied = Change.apply before change in let cached_after = Incremental.result fixture.fx_plan state in fixture.fx_state <- state; Ok (applied, cached_after, change) let order_row customer total = Value.VRecord ("order", [ ("customer", Value.VString customer); ("total", Value.VInt total) ]) let executor_cases = let orders text entries = let fixture = build_fixture ("input orders : collection order\n" ^ text) entries in fixture in [ ( "initialization matches the reference interpreter", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500); (2, order_row "Bo" 900) ] in check_equal_string "initial result" (Value.to_string (reference_result fixture fixture.fx_state)) (Value.to_string (Incremental.result fixture.fx_plan fixture.fx_state)) ); ( "inserting a key appends a mapped contribution", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] in (match step fixture [ Change.OpInsert (2, order_row "Lin" 2200) ] with | Error message -> fail "insert" message | Ok (applied, cached, change) -> let expected = reference_result fixture fixture.fx_state in check_equal_string "change" "collection [insert 2 (tuple \"Lin\" 440)]" (Change.to_string change); check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); ( "removing a key retracts the cached contribution", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500); (2, order_row "Lin" 2200) ] in (match step fixture [ Change.OpRemove 2 ] with | Error message -> fail "remove" message | Ok (applied, cached, change) -> let expected = reference_result fixture fixture.fx_state in check_equal_string "change" "collection [remove 2 (tuple \"Lin\" 440)]" (Change.to_string change); check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); ( "replacing a retained key forwards the new contribution", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] in (match step fixture [ Change.OpReplace (1, order_row "Ada" 2500) ] with | Error message -> fail "replace" message | Ok (applied, cached, change) -> let expected = reference_result fixture fixture.fx_state in check_equal_string "change" "collection [replace 1 (tuple \"Ada\" 300) (tuple \"Ada\" 500)]" (Change.to_string change); check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); ( "replacing a key with an equal row produces no change", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] in (match step fixture [ Change.OpReplace (1, order_row "Ada" 1500) ] with | Error message -> fail "equal" message | Ok (applied, cached, change) -> check "empty change" (Change.is_empty change); check_equal_string "applied" (Value.to_string (reference_result fixture fixture.fx_state)) (Value.to_string applied); check_equal_string "cached" (Value.to_string (reference_result fixture fixture.fx_state)) (Value.to_string cached)) ); ( "a row that stops matching leaves the collection only once", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] in (match step fixture [ Change.OpReplace (1, order_row "Ada" 100) ] with | Error message -> fail "drop" message | Ok (applied, cached, change) -> let expected = reference_result fixture fixture.fx_state in check_equal_string "change" "collection [remove 1 (tuple \"Ada\" 300)]" (Change.to_string change); check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); ( "a row that starts matching is inserted once", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 100) ] in (match step fixture [ Change.OpReplace (1, order_row "Ada" 5000) ] with | Error message -> fail "gain" message | Ok (applied, cached, change) -> let expected = reference_result fixture fixture.fx_state in check_equal_string "change" "collection [insert 1 (tuple \"Ada\" 1000)]" (Change.to_string change); check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); ( "several batches keep the state consistent", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] in (match step fixture [ Change.OpInsert (2, order_row "Bo" 3000) ] with | Error message -> fail "insert" message | Ok _ -> ()); (match step fixture [ Change.OpRemove 1 ] with | Error message -> fail "remove" message | Ok _ -> ()); (match step fixture [ Change.OpReplace (2, order_row "Bo" 900) ] with | Error message -> fail "replace" message | Ok (applied, cached, _) -> let expected = reference_result fixture fixture.fx_state in check_equal_string "applied" (Value.to_string expected) (Value.to_string applied); check_equal_string "cached" (Value.to_string expected) (Value.to_string cached)) ); ( "an identity query forwards the input change", fun () -> let fixture = build_fixture "input rows : collection int\nquery q = rows" [ (1, Value.VInt 5) ] in (match step fixture [ Change.OpInsert (2, Value.VInt 9) ] with | Error message -> fail "identity" message | Ok (applied, cached, change) -> check_equal_string "change" "collection [insert 2 9]" (Change.to_string change); check_equal_string "applied" "(collection (1 5) (2 9))" (Value.to_string applied); check_equal_string "cached" "(collection (1 5) (2 9))" (Value.to_string cached)) ); ( "an integer query reports an additive change", fun () -> let fixture = build_fixture (read_fixture "revenue.delta") [ (1, order_row "Ada" 1000) ] in (match step fixture [ Change.OpInsert (2, order_row "Bo" 250) ] with | Error message -> fail "sum" message | Ok (applied, cached, change) -> check_equal_string "change" "+50" (Change.to_string change); check_equal_string "applied" "250" (Value.to_string applied); check_equal_string "cached" "250" (Value.to_string cached)) ); ( "count ignores replacements that keep a row", fun () -> let fixture = build_fixture (read_fixture "count_large.delta") [ (1, order_row "Ada" 1000) ] in (match step fixture [ Change.OpReplace (1, order_row "Ada" 2000) ] with | Error message -> fail "count" message | Ok (_, _, change) -> check "empty change" (Change.is_empty change)); (match step fixture [ Change.OpReplace (1, order_row "Ada" 10) ] with | Error message -> fail "count" message | Ok (_, cached, change) -> check_equal_string "change" "-1" (Change.to_string change); check_equal_string "cached" "0" (Value.to_string cached)) ); ( "an invalid batch leaves the state usable", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500) ] in let before = Value.to_string (Incremental.result fixture.fx_plan fixture.fx_state) in (match step fixture [ Change.OpRemove 99 ] with | Ok _ -> fail "invalid" "expected a failure" | Error _ -> ()); check_equal_string "result unchanged" before (Value.to_string (Incremental.result fixture.fx_plan fixture.fx_state)); (match step fixture [ Change.OpInsert (2, order_row "Bo" 2000) ] with | Error message -> fail "recovery" message | Ok (applied, _, _) -> check_equal_string "still incremental" (Value.to_string (reference_result fixture fixture.fx_state)) (Value.to_string applied)) ); ( "updates visit one key per node", fun () -> let fixture = build_fixture (read_fixture "expensive_order.delta") [ (1, order_row "Ada" 1500); (2, order_row "Bo" 2000); (3, order_row "Cy" 500) ] in Delta_runtime.reset_counters fixture.fx_counters; (match step fixture [ Change.OpReplace (2, order_row "Bo" 2500) ] with | Error message -> fail "visit" message | Ok _ -> let counters = fixture.fx_counters in check_equal_int "changed key visits" 3 counters.Delta_runtime.changed_key_visits; check_equal_int "predicate evaluations" 1 counters.Delta_runtime.predicate_evaluations; check_equal_int "mapping evaluations" 0 counters.Delta_runtime.mapping_evaluations; check_equal_int "scalar deltas" 1 counters.Delta_runtime.scalar_deltas; check_equal_int "full traversals" 0 counters.Delta_runtime.full_traversals) ); ( "initialization counts one traversal per node", fun () -> let plan = plan_of_fixture "expensive_order.delta" in let counters = Delta_runtime.new_counters () in let _ = Incremental.init plan counters [ (1, order_row "Ada" 1500) ] in check_equal_int "full traversals" 3 counters.Delta_runtime.full_traversals; check_equal_int "mapping evaluations" 1 counters.Delta_runtime.mapping_evaluations; check_equal_int "predicate evaluations" 1 counters.Delta_runtime.predicate_evaluations ); ] 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) ); ]