859 lines
45 KiB
OCaml
859 lines
45 KiB
OCaml
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 );
|
|
]
|