Normalize keyed batches with atomic validation

src/change.ml

@@ -39,6 +39,57 @@ let string_of_value span value =
   | Value.VString text -> text
   | other -> Diagnostic.error span "runtime error: expected a string but got %s" (Value.to_string other)
 
+type input_op =
+  | OpInsert of int * Value.t
+  | OpRemove of int
+  | OpReplace of int * Value.t
+
+type 'a outcome = 'a Delta_runtime.outcome = Success of 'a | Failure of string
+
+let validate_batch ~existing ops =
+  let rec step temp touched = function
+    | [] -> Success (temp, touched)
+    | op :: rest -> (
+        match op with
+        | OpInsert (key, value) ->
+            if Delta_runtime.Pure_map.mem key temp then
+              Failure (Printf.sprintf "cannot insert key %d: it is already present" key)
+            else step (Delta_runtime.Pure_map.add key value temp) (key :: touched) rest
+        | OpRemove key ->
+            if Delta_runtime.Pure_map.mem key temp then
+              step (Delta_runtime.Pure_map.remove key temp) (key :: touched) rest
+            else Failure (Printf.sprintf "cannot remove key %d: it is not present" key)
+        | OpReplace (key, value) ->
+            if Delta_runtime.Pure_map.mem key temp then
+              step (Delta_runtime.Pure_map.add key value temp) (key :: touched) rest
+            else Failure (Printf.sprintf "cannot replace key %d: it is not present" key))
+  in
+  match step existing [] ops with
+  | Failure message -> Failure message
+  | Success (temp, touched) ->
+      let keys = List.sort_uniq compare touched in
+      let changes =
+        Util.filter_map
+          (fun key ->
+            match
+              ( Delta_runtime.Pure_map.find_opt key existing,
+                Delta_runtime.Pure_map.find_opt key temp )
+            with
+            | None, None -> None
+            | None, Some value -> Some (CInsert (key, value))
+            | Some value, None -> Some (CRemove (key, value))
+            | Some old_value, Some new_value ->
+                if Value.equal old_value new_value then None
+                else Some (CReplace (key, old_value, new_value)))
+          keys
+      in
+      Success (temp, changes)
+
+let apply_batch ~existing ops =
+  match validate_batch ~existing ops with
+  | Failure message -> Failure message
+  | Success (temp, changes) -> Success (temp, CCollection changes, changes)
+
 let scalar_mismatch span expected value =
   Diagnostic.error span "this change replaces a %s value but is applied to %s" expected
     (Value.to_string value)

src/change.mli

@@ -13,6 +13,23 @@ type change =
   | CRecord of string * (string * change) list
   | CCollection of keyed list
 
+type input_op =
+  | OpInsert of int * Value.t
+  | OpRemove of int
+  | OpReplace of int * Value.t
+
+type 'a outcome = 'a Delta_runtime.outcome = Success of 'a | Failure of string
+
+val validate_batch :
+  existing:Value.t Delta_runtime.Pure_map.t ->
+  input_op list ->
+  (Value.t Delta_runtime.Pure_map.t * keyed list) outcome
+
+val apply_batch :
+  existing:Value.t Delta_runtime.Pure_map.t ->
+  input_op list ->
+  (Value.t Delta_runtime.Pure_map.t * change * keyed list) outcome
+
 val empty : Types.t -> change
 val is_empty : change -> bool
 val apply : Value.t -> change -> Value.t

test/test_change.ml

@@ -192,3 +192,181 @@ let law_cases =
         expect_diagnostic "compose" (fun () -> ignore (Change.compose (Change.CInt 1) (Change.CString "x"))) );
   ]
 
+
+let collection keys element_ty rng =
+  let entries =
+    List.map
+      (fun key -> (key, value_of_type rng element_ty))
+      keys
+  in
+  Value.collection_of_list entries
+
+let batch_cases =
+  let element_ty = Types.TInt in
+  let original rng = collection [ 1; 2; 3 ] element_ty rng in
+  [
+    ( "an insert of an absent key becomes an insertion change",
+      fun () ->
+        let rng = rng 1 in
+        let existing = original rng in
+        match Change.validate_batch ~existing [ Change.OpInsert (7, Value.VInt 70) ] with
+        | Change.Failure message -> fail "insert" message
+        | Change.Success (temp, changes) ->
+            check_equal_string "changes"
+              "collection [insert 7 70]"
+              (Change.to_string (Change.CCollection changes));
+            check_equal_int "temporary map" 4 (Delta_runtime.Pure_map.cardinal temp) );
+    ( "removing a present key becomes a removal change",
+      fun () ->
+        let rng = rng 2 in
+        let existing = original rng in
+        let value = Delta_runtime.Pure_map.find 2 existing in
+        match Change.validate_batch ~existing [ Change.OpRemove 2 ] with
+        | Change.Failure message -> fail "remove" message
+        | Change.Success (temp, changes) ->
+            check_equal_string "changes"
+              (Printf.sprintf "collection [remove 2 %s]" (Value.to_string value))
+              (Change.to_string (Change.CCollection changes));
+            check "the key is gone" (not (Delta_runtime.Pure_map.mem 2 temp)) );
+    ( "replacing a present key becomes a replacement change",
+      fun () ->
+        let rng = rng 3 in
+        let existing = original rng in
+        let value = Delta_runtime.Pure_map.find 1 existing in
+        match Change.validate_batch ~existing [ Change.OpReplace (1, Value.VInt 99) ] with
+        | Change.Failure message -> fail "replace" message
+        | Change.Success (_, changes) ->
+            check_equal_string "changes"
+              (Printf.sprintf "collection [replace 1 %s 99]" (Value.to_string value))
+              (Change.to_string (Change.CCollection changes)) );
+    ( "insert followed by remove cancels",
+      fun () ->
+        let rng = rng 4 in
+        let existing = original rng in
+        let ops = [ Change.OpInsert (9, Value.VInt 1); Change.OpRemove 9 ] in
+        match Change.validate_batch ~existing ops with
+        | Change.Failure message -> fail "cancel" message
+        | Change.Success (temp, changes) ->
+            check "no effective change" (changes = []);
+            check "the map is unchanged" (Value.equal (Value.VCollection temp) (Value.VCollection existing)) );
+    ( "several replacements collapse into one",
+      fun () ->
+        let rng = rng 5 in
+        let existing = original rng in
+        let value = Delta_runtime.Pure_map.find 3 existing in
+        let ops =
+          [ Change.OpReplace (3, Value.VInt 10); Change.OpReplace (3, Value.VInt 20); Change.OpReplace (3, Value.VInt 30) ]
+        in
+        match Change.validate_batch ~existing ops with
+        | Change.Failure message -> fail "collapse" message
+        | Change.Success (_, changes) ->
+            check_equal_string "changes"
+              (Printf.sprintf "collection [replace 3 %s 30]" (Value.to_string value))
+              (Change.to_string (Change.CCollection changes)) );
+    ( "removing and reinserting an original key becomes a replacement",
+      fun () ->
+        let rng = rng 6 in
+        let existing = original rng in
+        let value = Delta_runtime.Pure_map.find 2 existing in
+        let ops = [ Change.OpRemove 2; Change.OpInsert (2, Value.VInt 200) ] in
+        match Change.validate_batch ~existing ops with
+        | Change.Failure message -> fail "reinsert" message
+        | Change.Success (_, changes) ->
+            check_equal_string "changes"
+              (Printf.sprintf "collection [replace 2 %s 200]" (Value.to_string value))
+              (Change.to_string (Change.CCollection changes)) );
+    ( "a replacement with an equal value is not an effective change",
+      fun () ->
+        let rng = rng 7 in
+        let existing = original rng in
+        let value = Delta_runtime.Pure_map.find 1 existing in
+        match Change.validate_batch ~existing [ Change.OpReplace (1, value) ] with
+        | Change.Failure message -> fail "equal" message
+        | Change.Success (_, changes) -> check "no effective change" (changes = []) );
+    ( "a duplicate insertion is rejected",
+      fun () ->
+        let rng = rng 8 in
+        let existing = original rng in
+        let ops = [ Change.OpInsert (5, Value.VInt 1); Change.OpInsert (5, Value.VInt 2) ] in
+        (match Change.validate_batch ~existing ops with
+        | Change.Success _ -> fail "duplicate" "expected a failure"
+        | Change.Failure message ->
+            check_equal_string "message" "cannot insert key 5: it is already present" message) );
+    ( "removing a missing key is rejected",
+      fun () ->
+        let rng = rng 9 in
+        let existing = original rng in
+        match Change.validate_batch ~existing [ Change.OpRemove 42 ] with
+        | Change.Success _ -> fail "missing" "expected a failure"
+        | Change.Failure message ->
+            check_equal_string "message" "cannot remove key 42: it is not present" message );
+    ( "replacing a missing key is rejected",
+      fun () ->
+        let rng = rng 10 in
+        let existing = original rng in
+        match Change.validate_batch ~existing [ Change.OpReplace (42, Value.VInt 1) ] with
+        | Change.Success _ -> fail "missing" "expected a failure"
+        | Change.Failure message ->
+            check_equal_string "message" "cannot replace key 42: it is not present" message );
+    ( "later operations see the effect of earlier ones",
+      fun () ->
+        let rng = rng 11 in
+        let existing = original rng in
+        let ops = [ Change.OpInsert (4, Value.VInt 40); Change.OpRemove 4; Change.OpInsert (4, Value.VInt 41) ] in
+        match Change.validate_batch ~existing ops with
+        | Change.Failure message -> fail "sequence" message
+        | Change.Success (temp, changes) ->
+            check_equal_string "changes" "collection [insert 4 41]"
+              (Change.to_string (Change.CCollection changes));
+            check "the temporary map holds the final value"
+              (match Delta_runtime.Pure_map.find_opt 4 temp with Some (Value.VInt 41) -> true | _ -> false) );
+    ( "a failed batch leaves the input untouched",
+      fun () ->
+        let rng = rng 12 in
+        let existing = original rng in
+        let before = Value.VCollection existing in
+        let ops = [ Change.OpInsert (8, Value.VInt 1); Change.OpRemove 42 ] in
+        (match Change.validate_batch ~existing ops with
+        | Change.Success _ -> fail "atomic" "expected a failure"
+        | Change.Failure _ -> ());
+        check_equal_string "the input is unchanged" (Value.to_string before) (Value.to_string (Value.VCollection existing));
+        check_equal_int "cardinality" 3 (Delta_runtime.Pure_map.cardinal existing) );
+    ( "the normalized change reproduces the validated map",
+      fun () ->
+        List.iter
+          (fun seed ->
+            let rng = rng seed in
+            let existing = original rng in
+            let count = range rng 6 + 1 in
+            let keys = [ 1; 2; 3; 4; 5 ] in
+            let temp = ref existing in
+            let ops = ref [] in
+            for _ = 1 to count do
+              let key = pick rng keys in
+              if Delta_runtime.Pure_map.mem key !temp then
+                if range rng 2 = 0 then (
+                  ops := Change.OpRemove key :: !ops;
+                  temp := Delta_runtime.Pure_map.remove key !temp)
+                else (
+                  let value = Value.VInt (range rng 100) in
+                  ops := Change.OpReplace (key, value) :: !ops;
+                  temp := Delta_runtime.Pure_map.add key value !temp)
+              else (
+                let value = Value.VInt (range rng 100) in
+                ops := Change.OpInsert (key, value) :: !ops;
+                temp := Delta_runtime.Pure_map.add key value !temp)
+            done;
+            let ops = List.rev !ops in
+            match Change.validate_batch ~existing ops with
+            | Change.Failure message -> fail "random batch" (Printf.sprintf "seed %d: %s" seed message)
+            | Change.Success (validated, changes) ->
+                let rebuilt = Change.apply_keyed existing changes in
+                let expected = Value.VCollection !temp in
+                let actual = Value.VCollection validated in
+                let from_changes = Value.VCollection rebuilt in
+                if not (Value.equal expected actual) then fail "validated map" (Printf.sprintf "seed %d" seed);
+                if not (Value.equal expected from_changes) then
+                  fail "change application" (Printf.sprintf "seed %d ops=%s" seed
+                    (Change.to_string (Change.CCollection changes))))
+          (Util.list_init 120 (fun index -> index + 1)) );
+  ]

test/test_main.ml

@@ -377,6 +377,7 @@ let () =
   Test_harness.run_suite "specialize" Test_type.specialize_cases;
   Test_harness.run_suite "anf" Test_type.anf_cases;
   Test_harness.run_suite "changes" Test_change.law_cases;
+  Test_harness.run_suite "batches" Test_change.batch_cases;
   Test_harness.run_suite "interpret" Test_incremental.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)