Add typed wire input and native executable builds

Makefile

@@ -33,7 +33,7 @@ RUNTIME_ARCHIVE = $(BUILD)/delta_runtime.cmxa
 RUNTIME_BYTE = $(BUILD)/delta_runtime.cma
 RUNTIME_NATIVE = $(BUILD)/delta_runtime.a
 RUNTIME_ARTIFACTS = $(RUNTIME_ARCHIVE) $(RUNTIME_NATIVE) $(RUNTIME_BYTE) \
-  $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmi,$(RT_ML)) $(RT_CMX)
+  $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmi,$(RT_ML)) $(patsubst $(RUNTIME_DIR)/%.ml,$(BUILD)/%.cmo,$(RT_ML)) $(RT_CMX)
 
 ifeq ($(strip $(RT_ML)),)
 RUNTIME_ARCHIVE =

example/order.sexp

@@ -0,0 +1,5 @@
+(1 (record (customer "Ada") (total 1500)))
+(2 (record (customer "Bo") (total 900)))
+(3 (record (customer "Lin") (total 2200)))
+(4 (record (customer "Dara") (total 400)))
+(5 (record (customer "Eve") (total 5000)))

example/update.sexp

@@ -0,0 +1,8 @@
+(batch
+  (insert 6 (record (customer "Fay") (total 4200)))
+  (replace 1 (record (customer "Ada") (total 100))))
+(batch
+  (replace 3 (record (customer "Lin") (total 50)))
+  (remove 5))
+(batch
+  (replace 4 (record (customer "Dara") (total 2500))))

runtime/wire.ml

@@ -0,0 +1,353 @@
+type value =
+  | WUnit
+  | WInt of int
+  | WBool of bool
+  | WString of string
+  | WTuple of value list
+  | WRecord of (string * value) list
+  | WCollection of (int * value) list
+
+type op = WInsert of int * value | WRemove of int | WReplace of int * value
+
+type state = {
+  text : string;
+  mutable position : int;
+  mutable line : int;
+  mutable column : int;
+}
+
+let failure state message =
+  Delta_runtime.Failure (Printf.sprintf "line %d, column %d: %s" state.line state.column message)
+
+let failure_at state line column message =
+  Delta_runtime.Failure (Printf.sprintf "line %d, column %d: %s" line column message)
+
+let at_end state = state.position >= String.length state.text
+
+let peek state = if at_end state then '\000' else state.text.[state.position]
+
+let advance state =
+  (if peek state = '\n' then (
+     state.line <- state.line + 1;
+     state.column <- 1)
+   else state.column <- state.column + 1);
+  state.position <- state.position + 1
+
+let rec skip_space state =
+  if not (at_end state) then
+    match peek state with
+    | ' ' | '\t' | '\r' | '\n' ->
+        advance state;
+        skip_space state
+    | _ -> ()
+
+let expect state character =
+  if peek state = character then (
+    advance state;
+    true)
+  else false
+
+let is_digit character = character >= '0' && character <= '9'
+
+let is_symbol_start character =
+  (character >= 'a' && character <= 'z')
+  || (character >= 'A' && character <= 'Z')
+  || character = '_'
+
+let is_symbol_char character = is_symbol_start character || is_digit character || character = '\''
+
+let read_symbol state =
+  let start = state.position in
+  let rec scan () =
+    if not (at_end state) && is_symbol_char (peek state) then (
+      advance state;
+      scan ())
+  in
+  scan ();
+  String.sub state.text start (state.position - start)
+
+let read_string state =
+  advance state;
+  let buffer = Buffer.create 16 in
+  let rec scan () =
+    if at_end state then failure state "unterminated string"
+    else
+      let character = peek state in
+      if character = '"' then (
+        advance state;
+        Delta_runtime.Success (Buffer.contents buffer))
+      else if character = '\\' then (
+        advance state;
+        let escaped =
+          if at_end state then '\000'
+          else
+            let next = peek state in
+            advance state;
+            next
+        in
+        match escaped with
+        | 'n' ->
+            Buffer.add_char buffer '\n';
+            scan ()
+        | 't' ->
+            Buffer.add_char buffer '\t';
+            scan ()
+        | 'r' ->
+            Buffer.add_char buffer '\r';
+            scan ()
+        | '"' ->
+            Buffer.add_char buffer '"';
+            scan ()
+        | '\\' ->
+            Buffer.add_char buffer '\\';
+            scan ()
+        | other -> failure state (Printf.sprintf "unknown escape sequence \\%c" other))
+      else if character = '\n' then failure state "unterminated string"
+      else (
+        Buffer.add_char buffer character;
+        advance state;
+        scan ())
+  in
+  scan ()
+
+let read_integer state =
+  let start = state.position in
+  if peek state = '-' then advance state;
+  if not (is_digit (peek state)) then failure state "expected an integer"
+  else (
+    let rec scan () =
+      if is_digit (peek state) then (
+        advance state;
+        scan ())
+    in
+    scan ();
+    let text = String.sub state.text start (state.position - start) in
+    try Delta_runtime.Success (int_of_string text)
+    with Failure _ -> failure state (Printf.sprintf "integer %s does not fit in a machine integer" text))
+
+let rec read_value state =
+  skip_space state;
+  if at_end state then failure state "expected a value"
+  else
+    match peek state with
+    | '(' -> read_compound state
+    | '"' -> (
+        match read_string state with
+        | Delta_runtime.Success text -> Delta_runtime.Success (WString text)
+        | Delta_runtime.Failure message -> Delta_runtime.Failure message)
+    | character when is_digit character || character = '-' -> (
+        match read_integer state with
+        | Delta_runtime.Success number -> Delta_runtime.Success (WInt number)
+        | Delta_runtime.Failure message -> Delta_runtime.Failure message)
+    | character when is_symbol_start character ->
+        let symbol = read_symbol state in
+        if symbol = "true" then Delta_runtime.Success (WBool true)
+        else if symbol = "false" then Delta_runtime.Success (WBool false)
+        else if symbol = "unit" then Delta_runtime.Success WUnit
+        else failure state (Printf.sprintf "unknown scalar `%s`" symbol)
+    | character -> failure state (Printf.sprintf "unexpected character %C" character)
+
+and read_compound state =
+  ignore (expect state '(');
+  skip_space state;
+  if not (is_symbol_start (peek state)) then failure state "expected a tagged value"
+  else
+    let tag_line = state.line in
+    let tag_column = state.column in
+    let tag = read_symbol state in
+    match tag with
+    | "tuple" ->
+        let rec items acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (WTuple (List.rev acc))
+          else (
+            match read_value state with
+            | Delta_runtime.Success value -> items (value :: acc)
+            | Delta_runtime.Failure message -> Delta_runtime.Failure message)
+        in
+        items []
+    | "record" ->
+        let rec fields acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (WRecord (List.rev acc))
+          else if peek state <> '(' then failure state "expected a record field"
+          else (
+            ignore (expect state '(');
+            skip_space state;
+            if not (is_symbol_start (peek state)) then failure state "expected a field label"
+            else
+              let label = read_symbol state in
+              match read_value state with
+              | Delta_runtime.Failure message -> Delta_runtime.Failure message
+              | Delta_runtime.Success value ->
+                  skip_space state;
+                  if not (expect state ')') then failure state "expected `)` after a record field"
+                  else fields ((label, value) :: acc))
+        in
+        fields []
+    | "collection" ->
+        let rec entries acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (WCollection (List.rev acc))
+          else if peek state <> '(' then failure state "expected a collection entry"
+          else (
+            ignore (expect state '(');
+            match read_integer state with
+            | Delta_runtime.Failure message -> Delta_runtime.Failure message
+            | Delta_runtime.Success key -> (
+                match read_value state with
+                | Delta_runtime.Failure message -> Delta_runtime.Failure message
+                | Delta_runtime.Success value ->
+                    skip_space state;
+                    if not (expect state ')') then failure state "expected `)` after a collection entry"
+                    else entries ((key, value) :: acc)))
+        in
+        entries []
+    | other -> failure_at state tag_line tag_column (Printf.sprintf "unknown tag `%s`" other)
+
+let to_string value =
+  let rec render value =
+    match value with
+    | WUnit -> "unit"
+    | WInt number -> string_of_int number
+    | WBool true -> "true"
+    | WBool false -> "false"
+    | WString text ->
+        let buffer = Buffer.create (String.length text + 2) in
+        Buffer.add_char buffer '"';
+        String.iter
+          (fun character ->
+            match character with
+            | '"' -> Buffer.add_string buffer "\\\""
+            | '\\' -> Buffer.add_string buffer "\\\\"
+            | '\n' -> Buffer.add_string buffer "\\n"
+            | '\t' -> Buffer.add_string buffer "\\t"
+            | '\r' -> Buffer.add_string buffer "\\r"
+            | other -> Buffer.add_char buffer other)
+          text;
+        Buffer.add_char buffer '"';
+        Buffer.contents buffer
+    | WTuple items -> "(tuple " ^ String.concat " " (List.map render items) ^ ")"
+    | WRecord fields ->
+        "(record "
+        ^ String.concat " " (List.map (fun (label, value) -> "(" ^ label ^ " " ^ render value ^ ")") fields)
+        ^ ")"
+    | WCollection entries ->
+        "(collection "
+        ^ String.concat " "
+            (List.rev (List.rev_map (fun (key, value) -> Printf.sprintf "(%d %s)" key (render value)) entries))
+        ^ ")"
+  in
+  render value
+
+let field_value label fields =
+  let matches = List.filter (fun (candidate, _) -> candidate = label) fields in
+  match matches with
+  | [] -> Delta_runtime.Failure (Printf.sprintf "missing field `%s`" label)
+  | [ (_, value) ] -> Delta_runtime.Success value
+  | _ -> Delta_runtime.Failure (Printf.sprintf "duplicate field `%s`" label)
+
+let parse_value text =
+  let state = { text = text; position = 0; line = 1; column = 1 } in
+  match read_value state with
+  | Delta_runtime.Failure message -> Delta_runtime.Failure message
+  | Delta_runtime.Success value ->
+      skip_space state;
+      if at_end state then Delta_runtime.Success value
+      else failure state "unexpected trailing input"
+
+let parse_input text =
+  let state = { text = text; position = 0; line = 1; column = 1 } in
+  let rec entries acc =
+    skip_space state;
+    if at_end state then Delta_runtime.Success (List.rev acc)
+    else if peek state <> '(' then failure state "expected `(key value)`"
+    else (
+      ignore (expect state '(');
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key -> (
+          match read_value state with
+          | Delta_runtime.Failure message -> Delta_runtime.Failure message
+          | Delta_runtime.Success value ->
+              skip_space state;
+              if not (expect state ')') then failure state "expected `)` after an input entry"
+              else entries ((key, value) :: acc)))
+  in
+  entries []
+
+let parse_operation state =
+  ignore (expect state '(');
+  skip_space state;
+  let tag = read_symbol state in
+  skip_space state;
+  match tag with
+  | "insert" -> (
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key -> (
+          match read_value state with
+          | Delta_runtime.Failure message -> Delta_runtime.Failure message
+          | Delta_runtime.Success value ->
+              skip_space state;
+              if not (expect state ')') then failure state "expected `)` after an insert"
+              else Delta_runtime.Success (WInsert (key, value))))
+  | "remove" -> (
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key ->
+          skip_space state;
+          if not (expect state ')') then failure state "expected `)` after a remove"
+          else Delta_runtime.Success (WRemove key))
+  | "replace" -> (
+      match read_integer state with
+      | Delta_runtime.Failure message -> Delta_runtime.Failure message
+      | Delta_runtime.Success key -> (
+          match read_value state with
+          | Delta_runtime.Failure message -> Delta_runtime.Failure message
+          | Delta_runtime.Success value ->
+              skip_space state;
+              if not (expect state ')') then failure state "expected `)` after a replace"
+              else Delta_runtime.Success (WReplace (key, value))))
+  | other -> failure state (Printf.sprintf "unknown update operation `%s`" other)
+
+let parse_updates text =
+  let state = { text = text; position = 0; line = 1; column = 1 } in
+  let rec batches acc =
+    skip_space state;
+    if at_end state then Delta_runtime.Success (List.rev acc)
+    else if peek state <> '(' then failure state "expected `(batch ...)`"
+    else (
+      ignore (expect state '(');
+      skip_space state;
+      let tag_line = state.line in
+      let tag_column = state.column in
+      let tag = read_symbol state in
+      if tag <> "batch" then
+        failure_at state tag_line tag_column (Printf.sprintf "expected `batch` but found `%s`" tag)
+      else
+        let rec operations acc =
+          skip_space state;
+          if expect state ')' then Delta_runtime.Success (List.rev acc)
+          else if peek state <> '(' then failure state "expected an update operation"
+          else
+            match parse_operation state with
+            | Delta_runtime.Failure message -> Delta_runtime.Failure message
+            | Delta_runtime.Success operation -> operations (operation :: acc)
+        in
+        match operations [] with
+        | Delta_runtime.Failure message -> Delta_runtime.Failure message
+        | Delta_runtime.Success operations -> batches (operations :: acc))
+  in
+  batches []
+
+let read_file path =
+  try
+    let channel = open_in_bin path in
+    let length = in_channel_length channel in
+    let text = really_input_string channel length in
+    close_in channel;
+    Delta_runtime.Success text
+  with
+  | Sys_error message -> Delta_runtime.Failure message
+  | End_of_file -> Delta_runtime.Failure (Printf.sprintf "could not read %s" path)

runtime/wire.mli

@@ -0,0 +1,17 @@
+type value =
+  | WUnit
+  | WInt of int
+  | WBool of bool
+  | WString of string
+  | WTuple of value list
+  | WRecord of (string * value) list
+  | WCollection of (int * value) list
+
+type op = WInsert of int * value | WRemove of int | WReplace of int * value
+
+val to_string : value -> string
+val field_value : string -> (string * value) list -> value Delta_runtime.outcome
+val parse_value : string -> value Delta_runtime.outcome
+val parse_input : string -> (int * value) list Delta_runtime.outcome
+val parse_updates : string -> op list list Delta_runtime.outcome
+val read_file : string -> string Delta_runtime.outcome

src/emit.ml

@@ -556,6 +556,208 @@ let emit_output_functions plan buffer =
     line buffer "  let output_change_to_string (change : output_change) : string = Printf.sprintf \"%+d\" change");
   line buffer ""
 
+let emit_wire_type_functions plan buffer info =
+  let name = info.ti_name in
+  let decode_failure what =
+    Printf.sprintf
+      " | other -> Delta_runtime.Failure (Printf.sprintf \"expected %s for a value of type %s but got %%s\" (Wire.to_string other))"
+      what name
+  in
+  match Types.repr info.ti_ty with
+  | Types.TInt ->
+      line buffer (Printf.sprintf "  let encode_%s (value : %s) : Wire.value = Wire.WInt value" name name);
+      line buffer
+        (Printf.sprintf
+           "  let decode_%s (value : Wire.value) : %s Delta_runtime.outcome = match value with Wire.WInt number -> Delta_runtime.Success number%s"
+           name name (decode_failure "an integer"))
+  | Types.TBool ->
+      line buffer (Printf.sprintf "  let encode_%s (value : %s) : Wire.value = Wire.WBool value" name name);
+      line buffer
+        (Printf.sprintf
+           "  let decode_%s (value : Wire.value) : %s Delta_runtime.outcome = match value with Wire.WBool truth -> Delta_runtime.Success truth%s"
+           name name (decode_failure "a boolean"))
+  | Types.TString ->
+      line buffer (Printf.sprintf "  let encode_%s (value : %s) : Wire.value = Wire.WString value" name name);
+      line buffer
+        (Printf.sprintf
+           "  let decode_%s (value : Wire.value) : %s Delta_runtime.outcome = match value with Wire.WString text -> Delta_runtime.Success text%s"
+           name name (decode_failure "a string"))
+  | Types.TUnit ->
+      line buffer (Printf.sprintf "  let encode_%s (value : %s) : Wire.value = Wire.WUnit" name name);
+      line buffer
+        (Printf.sprintf
+           "  let decode_%s (value : Wire.value) : %s Delta_runtime.outcome = match value with Wire.WUnit -> Delta_runtime.Success ()%s"
+           name name (decode_failure "unit"))
+  | Types.TTuple items -> (
+      let count = List.length items in
+      if count >= 2 then (
+        let names = List.map type_name items in
+        let pattern prefix =
+          "[" ^ Util.join "; " (Util.list_init count (fun index -> indexed prefix (index + 1))) ^ "]"
+        in
+        line buffer (Printf.sprintf "  let encode_%s (value : %s) : Wire.value =" name name);
+        line buffer (Printf.sprintf "    let %s = value in" ("(" ^ Util.join ", " (Util.list_init count (fun index -> indexed "v" (index + 1))) ^ ")"));
+        line buffer
+          (Printf.sprintf "    Wire.WTuple [ %s ]"
+             (Util.join "; "
+                (Util.list_init count (fun index ->
+                     Printf.sprintf "encode_%s %s" (List.nth names index) (indexed "v" (index + 1))))));
+        line buffer
+          (Printf.sprintf "  let decode_%s (value : Wire.value) : %s Delta_runtime.outcome =" name name);
+        line buffer (Printf.sprintf "    match value with");
+        line buffer (Printf.sprintf "    | Wire.WTuple %s ->" (pattern "v"));
+        let rec chain index =
+          if index > count then
+            Printf.sprintf "        Delta_runtime.Success (%s)"
+              ("(" ^ Util.join ", " (Util.list_init count (fun position -> indexed "x" (position + 1))) ^ ")")
+          else
+            Printf.sprintf "        (match decode_%s %s with Delta_runtime.Failure message -> Delta_runtime.Failure message | Delta_runtime.Success %s ->\n%s)"
+              (List.nth names (index - 1))
+              (indexed "v" index)
+              (indexed "x" index)
+              (chain (index + 1))
+        in
+        line buffer (chain 1);
+        line buffer (Printf.sprintf "%s" (decode_failure "a tuple of the right size"))))
+  | Types.TRecord record_name -> (
+      match Types.record_info plan.Graph.pl_records record_name with
+      | None -> ()
+      | Some record ->
+          let fields = record.Types.ri_fields in
+          line buffer (Printf.sprintf "  let encode_%s (value : %s) : Wire.value =" name name);
+          line buffer
+            (Printf.sprintf "    Wire.WRecord [ %s ]"
+               (Util.join "; "
+                  (List.map
+                     (fun (label, ty) ->
+                       Printf.sprintf "(%S, encode_%s value.%s)" label (type_name ty) label)
+                     fields)));
+          line buffer
+            (Printf.sprintf "  let decode_%s (value : Wire.value) : %s Delta_runtime.outcome =" name name);
+          line buffer "    match value with";
+          line buffer "    | Wire.WRecord fields ->";
+          let rec chain remaining =
+            match remaining with
+            | [] ->
+                Printf.sprintf "        Delta_runtime.Success { %s }"
+                  (Util.join "; "
+                     (List.map (fun (label, _) -> Printf.sprintf "%s = %s" label label) fields))
+            | (label, ty) :: rest ->
+                Printf.sprintf
+                  "        (match Wire.field_value %S fields with Delta_runtime.Failure message -> Delta_runtime.Failure message | Delta_runtime.Success raw ->\n         (match decode_%s raw with Delta_runtime.Failure message -> Delta_runtime.Failure message | Delta_runtime.Success %s ->\n%s))"
+                  label (type_name ty) label (chain rest)
+          in
+          line buffer (chain fields);
+          line buffer (decode_failure "a record"))
+  | Types.TCollection _ | Types.TArrow _ | Types.TVar _ -> ()
+
+let emit_wire_functions plan buffer =
+  List.iter (fun info -> emit_wire_type_functions plan buffer info) (collect_types plan);
+  line buffer "";
+  let input = input_type plan in
+  line buffer
+    (Printf.sprintf "  let decode_row (value : Wire.value) : %s Delta_runtime.outcome = decode_%s value"
+       (ocaml_ty input) (type_name input));
+  line buffer "";
+  line buffer "  let encode_output (value : output) : Wire.value =";
+  (if is_collection_output plan then (
+     line buffer
+       (Printf.sprintf
+          "    Wire.WCollection (List.rev (List.rev_map (fun (key, item) -> (key, encode_%s item)) (Delta_runtime.Pure_map.bindings value)))"
+          (type_name (output_value_type plan))))
+   else line buffer "    Wire.WInt value");
+  line buffer "";
+  line buffer
+    (Printf.sprintf
+       "  let decode_batch_op (op : Wire.op) : batch_op Delta_runtime.outcome =\n    match op with\n    | Wire.WInsert (key, value) -> (match decode_%s value with Delta_runtime.Failure message -> Delta_runtime.Failure message | Delta_runtime.Success row -> Delta_runtime.Success (Insert (key, row)))\n    | Wire.WRemove key -> Delta_runtime.Success (Remove key)\n    | Wire.WReplace (key, value) -> (match decode_%s value with Delta_runtime.Failure message -> Delta_runtime.Failure message | Delta_runtime.Success row -> Delta_runtime.Success (Replace (key, row)))"
+       (type_name input) (type_name input));
+  line buffer ""
+
+let emit_driver plan buffer =
+  let input = ocaml_ty (input_type plan) in
+  line buffer "";
+  line buffer "let () =";
+  line buffer "  let input_file = ref None in";
+  line buffer "  let updates_file = ref None in";
+  line buffer "  let print_result = ref false in";
+  line buffer "  let stats = ref false in";
+  line buffer "  let trace = ref false in";
+  line buffer "  let fail message = prerr_endline (\"error: \" ^ message); exit 1 in";
+  line buffer "  let usage message = prerr_endline (\"usage: \" ^ message); exit 2 in";
+  line buffer "  let rec parse_arguments arguments =";
+  line buffer "    match arguments with";
+  line buffer "    | [] -> ()";
+  line buffer "    | \"--input\" :: file :: rest -> input_file := Some file; parse_arguments rest";
+  line buffer "    | \"--updates\" :: file :: rest -> updates_file := Some file; parse_arguments rest";
+  line buffer "    | \"--print-result\" :: rest -> print_result := true; parse_arguments rest";
+  line buffer "    | \"--stats\" :: rest -> stats := true; parse_arguments rest";
+  line buffer "    | \"--trace\" :: rest -> trace := true; parse_arguments rest";
+  line buffer "    | flag :: _ -> usage (\"unknown option \" ^ flag)";
+  line buffer "  in";
+  line buffer "  parse_arguments (List.tl (Array.to_list Sys.argv));";
+  line buffer "  let started = Sys.time () in";
+  line buffer "  let rows =";
+  line buffer "    match !input_file with";
+  line buffer "    | None -> usage \"the --input FILE option is required\"";
+  line buffer "    | Some file -> (";
+  line buffer "        match Wire.read_file file with";
+  line buffer "        | Delta_runtime.Failure message -> fail message";
+  line buffer "        | Delta_runtime.Success text -> (";
+  line buffer "            match Wire.parse_input text with";
+  line buffer "            | Delta_runtime.Failure message -> fail (file ^ \": \" ^ message)";
+  line buffer
+    (Printf.sprintf
+       "            | Delta_runtime.Success entries ->\n                let rec decode acc pending =\n                  match pending with\n                  | [] -> List.rev acc\n                  | (key, value) :: rest ->\n                      (match Query.decode_row value with\n                       | Delta_runtime.Failure message ->\n                           fail (Printf.sprintf \"%%s: key %%d: %%s\" file key message)\n                       | Delta_runtime.Success row ->\n                           if Delta_runtime.Pure_map.mem key (List.fold_left (fun map (key, _) -> Delta_runtime.Pure_map.add key () map) Delta_runtime.Pure_map.empty acc) then\n                             fail (Printf.sprintf \"%%s: duplicate key %%d\" file key)\n                           else decode ((key, row) :: acc) rest)\n                in\n                decode [] entries));");
+  line buffer "  in";
+  line buffer
+    (Printf.sprintf "  let state = ref (Query.init (rows : (int * %s) list)) in" input);
+  line buffer "  let init_seconds = Sys.time () -. started in";
+  line buffer "  let init_counters = Delta_runtime.copy_counters (Query.counters ()) in";
+  line buffer "  Query.reset_counters ();";
+  line buffer "  let minor_words_before = Gc.minor_words () in";
+  line buffer "  let update_seconds = ref 0.0 in";
+  line buffer "  (match !updates_file with";
+  line buffer "   | None -> ()";
+  line buffer "   | Some file -> (";
+  line buffer "       match Wire.read_file file with";
+  line buffer "       | Delta_runtime.Failure message -> fail message";
+  line buffer "       | Delta_runtime.Success text -> (";
+  line buffer "           match Wire.parse_updates text with";
+  line buffer "           | Delta_runtime.Failure message -> fail (file ^ \": \" ^ message)";
+  line buffer "           | Delta_runtime.Success batches ->";
+  line buffer "               List.iteri";
+  line buffer "                 (fun index batch ->";
+  line buffer "                   let rec decode acc pending =";
+  line buffer "                     match pending with";
+  line buffer "                     | [] -> List.rev acc";
+  line buffer "                     | op :: rest -> (";
+  line buffer "                         match Query.decode_batch_op op with";
+  line buffer "                         | Delta_runtime.Failure message ->";
+  line buffer
+    "                             fail (Printf.sprintf \"%s: batch %d: %s\" file (index + 1) message)";
+  line buffer "                         | Delta_runtime.Success operation -> decode (operation :: acc) rest)";
+  line buffer "                   in";
+  line buffer "                   let operations = decode [] batch in";
+  line buffer "                   let before = Sys.time () in";
+  line buffer "                   (match Query.apply_batch !state operations with";
+  line buffer "                    | Delta_runtime.Failure message ->";
+  line buffer
+    "                        fail (Printf.sprintf \"%s: batch %d: %s\" file (index + 1) message)";
+  line buffer "                    | Delta_runtime.Success (next, _) ->";
+  line buffer "                        state := next;";
+  line buffer "                        update_seconds := !update_seconds +. (Sys.time () -. before);";
+  line buffer "                        if !trace then";
+  line buffer
+    "                          print_endline (Wire.to_string (Query.encode_output (Query.result next)))))";
+  line buffer "                 batches)));";
+  line buffer "  let minor_words_after = Gc.minor_words () in";
+  line buffer "  if !print_result then";
+  line buffer "    print_endline (Wire.to_string (Query.encode_output (Query.result !state)));";
+  line buffer "  if !stats then";
+  line buffer
+    "    Printf.eprintf \"init_counters: %s\\nupdate_counters: %s\\ninit_seconds: %.6f\\nupdate_seconds: %.6f\\nminor_words: %.0f\\n\" (Delta_runtime.counters_to_string init_counters) (Delta_runtime.counters_to_string (Query.counters ())) init_seconds !update_seconds (minor_words_after -. minor_words_before);";
+  line buffer "  exit 0"
+
 let emit_signature plan buffer =
   line buffer "  type state";
   line buffer "  type output";
@@ -570,6 +772,10 @@ let emit_signature plan buffer =
   line buffer "  val apply_output_change : output -> output_change -> output";
   line buffer "  val output_to_string : output -> string";
   line buffer "  val output_change_to_string : output_change -> string";
+  line buffer
+    (Printf.sprintf "  val decode_row : Wire.value -> %s Delta_runtime.outcome" (ocaml_ty (input_type plan)));
+  line buffer "  val encode_output : output -> Wire.value";
+  line buffer "  val decode_batch_op : Wire.op -> batch_op Delta_runtime.outcome";
   line buffer "  val counters : unit -> Delta_runtime.counters";
   line buffer "  val reset_counters : unit -> unit"
 
@@ -660,6 +866,7 @@ let emit_update_functions plan buffer =
         (Printf.sprintf
            "  %s update_%d (state : state) (key : int) (up_old : %s option) (up_new : %s option) : state ="
            keyword node.Graph.n_id (ocaml_ty input_element) (ocaml_ty input_element));
+      line buffer "    Delta_runtime.count_changed_key query_counters;";
       match node.Graph.n_kind with
       | Graph.Source ->
           line buffer
@@ -843,11 +1050,12 @@ let emit_structure plan buffer =
   emit_init plan buffer;
   emit_result plan buffer;
   emit_output_functions plan buffer;
+  emit_wire_functions plan buffer;
   line buffer "  let counters () = Delta_runtime.copy_counters query_counters";
   line buffer "";
   line buffer "  let reset_counters () = Delta_runtime.reset_counters query_counters"
 
-let program_to_string plan =
+let module_to_string plan =
   let buffer = Buffer.create 8192 in
   emit_declarations buffer plan;
   List.iter (fun info -> emit_type_functions buffer plan info) (collect_types plan);
@@ -860,4 +1068,9 @@ let program_to_string plan =
   line buffer "end";
   Buffer.contents buffer
 
-
+let program_to_string plan =
+  let module_source = module_to_string plan in
+  let buffer = Buffer.create (String.length module_source + 4096) in
+  Buffer.add_string buffer module_source;
+  emit_driver plan buffer;
+  Buffer.contents buffer

src/emit.mli

@@ -1,3 +1,4 @@
+val module_to_string : Graph.plan -> string
 val program_to_string : Graph.plan -> string
 val input_type : Graph.plan -> Types.t
 val ocaml_ty : Types.t -> string

src/main.ml

@@ -124,10 +124,24 @@ let emit path output =
   output_string channel (Emit.program_to_string program);
   close_out channel
 
+let temp_source plan =
+  let directory = Native.temp_dir () in
+  let source = Filename.concat directory "program.ml" in
+  let channel = open_out source in
+  output_string channel (Emit.program_to_string plan);
+  close_out channel;
+  (directory, source)
+
 let build path output =
-  ignore (plan_source path);
-  ignore output;
-  frontend_unavailable ()
+  let plan = plan_source path in
+  let directory, source = temp_source plan in
+  let runtime_dir = Native.runtime_dir () in
+  let result = Native.compile ~runtime_dir ~source ~output in
+  Native.remove_dir directory;
+  match result with
+  | None -> ()
+  | Some message -> Diagnostic.error Location.none "the native compiler failed:
+%s" message
 
 let source_of file = try Some (Native.read_file file) with _ -> None
 

src/native.ml

@@ -101,7 +101,9 @@ let run argv =
 
 let compile ~runtime_dir ~source ~output =
   let archive = Filename.concat runtime_dir "delta_runtime.cmxa" in
-  let argv = [| "ocamlopt"; "-I"; runtime_dir; "-o"; output; archive; source |] in
+  let argv =
+    [| "ocamlopt"; "-w"; "-26"; "-I"; runtime_dir; "-o"; output; archive; source |]
+  in
   if file_exists output then Sys.remove output;
   let status, text = run argv in
   match status with

test/test_codegen.ml

@@ -51,7 +51,7 @@ let run_generated name plan driver =
   Unix.mkdir dir 0o700;
   let source_path = Filename.concat dir (unique name ^ ".ml") in
   let exe_path = Filename.concat dir (unique name ^ ".exe") in
-  write_file source_path (Emit.program_to_string plan ^ driver);
+  write_file source_path (Emit.module_to_string plan ^ driver);
   let result =
     match compile source_path exe_path with
     | Some message -> Error ("the generated program did not compile:\n" ^ message)
@@ -312,3 +312,237 @@ let () =
             check_equal_string "result" "250" (List.nth lines 0);
             check_equal_string "additive change" "+50" (List.nth lines 1)) );
   ]
+
+let deltac () = Filename.concat (root ()) "deltac"
+
+let run_command command = run command
+
+let outcome_of text =
+  match Wire.parse_value text with
+  | Delta_runtime.Success value -> Value.to_string (Value.VUnit) |> fun _ -> Some value
+  | Delta_runtime.Failure _ -> None
+
+let wire_cases =
+  [
+    ( "scalar values round trip",
+      fun () ->
+        List.iter
+          (fun text ->
+            match Wire.parse_value text with
+            | Delta_runtime.Failure message -> fail "parse" (text ^ ": " ^ message)
+            | Delta_runtime.Success value ->
+                check_equal_string ("print " ^ text) text (Wire.to_string value))
+          [ "unit"; "0"; "-17"; "true"; "false"; "\"Ada\""; "\"a\\nb\"" ] );
+    ( "compound values round trip",
+      fun () ->
+        List.iter
+          (fun text ->
+            match Wire.parse_value text with
+            | Delta_runtime.Failure message -> fail "parse" (text ^ ": " ^ message)
+            | Delta_runtime.Success value ->
+                check_equal_string ("print " ^ text) text (Wire.to_string value))
+          [
+            "(tuple 1 \"a\")";
+            "(record (customer \"Ada\") (total 1500))";
+            "(collection (1 (tuple \"Ada\" 300)))";
+          ] );
+    ( "strings keep escapes through printing",
+      fun () ->
+        let text = "\"line\\nbreak\"" in
+        match Wire.parse_value text with
+        | Delta_runtime.Failure message -> fail "parse" message
+        | Delta_runtime.Success value ->
+            check_equal_string "round trip" text (Wire.to_string value) );
+    ( "unknown tags are rejected with a position",
+      fun () ->
+        match Wire.parse_value "(unknown 1)" with
+        | Delta_runtime.Success _ -> fail "tag" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check_equal_string "message" "line 1, column 2: unknown tag `unknown`" message );
+    ( "unbalanced parentheses are rejected",
+      fun () ->
+        match Wire.parse_value "(tuple 1" with
+        | Delta_runtime.Success _ -> fail "parens" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the position" (Util.starts_with "line 1, column " message) );
+    ( "integers that do not fit are rejected",
+      fun () ->
+        match Wire.parse_value "99999999999999999999" with
+        | Delta_runtime.Success _ -> fail "range" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the position" (Util.starts_with "line 1, column " message) );
+    ( "unterminated strings are rejected",
+      fun () ->
+        match Wire.parse_value "\"abc" with
+        | Delta_runtime.Success _ -> fail "string" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the string" (String.length message > 0) );
+    ( "input files parse into keyed entries",
+      fun () ->
+        let text = "(1 (record (customer \"Ada\") (total 1500)))\n(2 (record (customer \"Bo\") (total 900)))\n" in
+        match Wire.parse_input text with
+        | Delta_runtime.Failure message -> fail "input" message
+        | Delta_runtime.Success entries -> check_equal_int "two entries" 2 (List.length entries) );
+    ( "update files parse into batches",
+      fun () ->
+        let text = "(batch (insert 3 (record (a 1))) (replace 2 (record (a 1))))\n(batch (remove 1))\n" in
+        match Wire.parse_updates text with
+        | Delta_runtime.Failure message -> fail "updates" message
+        | Delta_runtime.Success batches ->
+            check_equal_int "two batches" 2 (List.length batches);
+            check_equal_int "two operations in the first batch" 2 (List.length (List.nth batches 0)) );
+    ( "update files report the failing line",
+      fun () ->
+        let text = "(batch (insert x )(record (a 1))))\n" in
+        match Wire.parse_updates text with
+        | Delta_runtime.Success _ -> fail "updates" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check "mentions the line" (Util.starts_with "line 1, column " message) );
+    ( "an update file without batches is rejected",
+      fun () ->
+        match Wire.parse_updates "(insert 1 (record (a 1)))\n" with
+        | Delta_runtime.Success _ -> fail "batch" "expected a failure"
+        | Delta_runtime.Failure message ->
+            check_equal_string "message" "line 1, column 2: expected `batch` but found `insert`" message );
+    ( "field_value reports missing and duplicate fields",
+      fun () ->
+        let fields = [ ("a", Wire.WInt 1); ("a", Wire.WInt 2) ] in
+        (match Wire.field_value "a" fields with
+        | Delta_runtime.Success _ -> fail "duplicate" "expected a failure"
+        | Delta_runtime.Failure message -> check_equal_string "duplicate" "duplicate field `a`" message);
+        (match Wire.field_value "b" fields with
+        | Delta_runtime.Success _ -> fail "missing" "expected a failure"
+        | Delta_runtime.Failure message -> check_equal_string "missing" "missing field `b`" message) );
+  ]
+
+let cli_cases =
+  let temp_dir name =
+    let dir = Filename.concat (Filename.get_temp_dir_name ()) name in
+    if Sys.file_exists dir then Native.remove_dir dir;
+    Unix.mkdir dir 0o700;
+    dir
+  in
+  [
+    ( "check succeeds on the examples and fails on broken sources",
+      fun () ->
+        let status, _ = run_command [ deltac (); "check"; fixture "expensive_order.delta" ] in
+        check "the example is accepted" (status = Unix.WEXITED 0);
+        let path = Filename.concat (Filename.get_temp_dir_name ()) (unique "broken" ^ ".delta") in
+        write_file path "query q = 1\n";
+        let status, output = run_command [ deltac (); "check"; path ] in
+        check "missing input is rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the input collection"
+          (String.length output > 0);
+        Sys.remove path );
+    ( "invalid command lines exit with status 2",
+      fun () ->
+        let status, _ = run_command [ deltac () ] in
+        check "no command" (status = Unix.WEXITED 2);
+        let status, _ = run_command [ deltac (); "check" ] in
+        check "check without a file" (status = Unix.WEXITED 2);
+        let status, _ = run_command [ deltac (); "dump"; fixture "expensive_order.delta" ] in
+        check "dump without a stage" (status = Unix.WEXITED 2);
+        let status, _ = run_command [ deltac (); "unknown" ] in
+        check "unknown command" (status = Unix.WEXITED 2) );
+    ( "dump stages write to stdout",
+      fun () ->
+        List.iter
+          (fun stage ->
+            let status, output =
+              run_command [ deltac (); "dump"; stage; fixture "expensive_order.delta" ]
+            in
+            check (stage ^ " succeeds") (status = Unix.WEXITED 0);
+            check (stage ^ " produces output") (String.length output > 0))
+          [ "--typed"; "--anf"; "--delta" ] );
+    ( "build compiles and runs the generated executable",
+      fun () ->
+        let dir = temp_dir (unique "delta spaces") in
+        let executable = Filename.concat dir "expensive orders" in
+        let status, output =
+          run_command [ deltac (); "build"; fixture "expensive_order.delta"; "-o"; executable ]
+        in
+        check ("build succeeds: " ^ output) (status = Unix.WEXITED 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1500)))\n(2 (record (customer \"Bo\") (total 900)))\n";
+        let updates = Filename.concat dir "update.sexp" in
+        write_file updates "(batch (insert 3 (record (customer \"Lin\") (total 2200))))\n(batch (remove 2))\n";
+        let status, output = run_command [ executable; "--input"; input; "--print-result" ] in
+        check "runs" (status = Unix.WEXITED 0);
+        check_equal_string "initial result" "(collection (1 (tuple \"Ada\" 300)))" (String.trim output);
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; updates; "--print-result" ]
+        in
+        check "runs with updates" (status = Unix.WEXITED 0);
+        check_equal_string "updated result" "(collection (1 (tuple \"Ada\" 300)) (3 (tuple \"Lin\" 440)))"
+          (String.trim output);
+        Native.remove_dir dir );
+    ( "trace prints the result after every batch",
+      fun () ->
+        let dir = temp_dir (unique "delta trace") in
+        let executable = Filename.concat dir "counted" in
+        let status, _ = run_command [ deltac (); "build"; fixture "count_large.delta"; "-o"; executable ] in
+        check "build succeeds" (status = Unix.WEXITED 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1000)))\n";
+        let updates = Filename.concat dir "update.sexp" in
+        write_file updates
+          "(batch (insert 2 (record (customer \"Bo\") (total 2000))))\n(batch (replace 1 (record (customer \"Ada\") (total 10))))\n";
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; updates; "--trace"; "--print-result" ]
+        in
+        check "runs" (status = Unix.WEXITED 0);
+        let lines = List.filter (fun line -> line <> "") (String.split_on_char '\n' output) in
+        check_equal_int "three lines" 3 (List.length lines);
+        check_equal_string "after the first batch" "2" (List.nth lines 0);
+        check_equal_string "after the second batch" "1" (List.nth lines 1);
+        check_equal_string "final result" "1" (List.nth lines 2);
+        Native.remove_dir dir );
+    ( "malformed input and invalid updates fail with a message",
+      fun () ->
+        let dir = temp_dir (unique "delta errors") in
+        let executable = Filename.concat dir "orders" in
+        let status, _ = run_command [ deltac (); "build"; fixture "expensive_order.delta"; "-o"; executable ] in
+        check "build succeeds" (status = Unix.WEXITED 0);
+        let bad_input = Filename.concat dir "bad.sexp" in
+        write_file bad_input "(1 (record (customer 5) (total 1500)))\n";
+        let status, output = run_command [ executable; "--input"; bad_input; "--print-result" ] in
+        check "type errors are rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the key" (String.length output > 0);
+        let duplicate = Filename.concat dir "duplicate.sexp" in
+        write_file duplicate
+          "(1 (record (customer \"Ada\") (total 1500)))\n(1 (record (customer \"Bo\") (total 900)))\n";
+        let status, output = run_command [ executable; "--input"; duplicate; "--print-result" ] in
+        check "duplicate keys are rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the duplicate" (String.length output > 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1500)))\n";
+        let bad_updates = Filename.concat dir "bad_update.sexp" in
+        write_file bad_updates "(batch (remove 7))\n";
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; bad_updates; "--print-result" ]
+        in
+        check "invalid updates are rejected" (status <> Unix.WEXITED 0);
+        check "the message mentions the batch" (String.length output > 0);
+        let syntax = Filename.concat dir "syntax.sexp" in
+        write_file syntax "(batch (remove 7)\n";
+        let status, _ = run_command [ executable; "--input"; input; "--updates"; syntax ] in
+        check "syntax errors are rejected" (status <> Unix.WEXITED 0);
+        Native.remove_dir dir );
+    ( "stats are printed to stderr and results to stdout",
+      fun () ->
+        let dir = temp_dir (unique "delta stats") in
+        let executable = Filename.concat dir "orders" in
+        let status, _ = run_command [ deltac (); "build"; fixture "count_large.delta"; "-o"; executable ] in
+        check "build succeeds" (status = Unix.WEXITED 0);
+        let input = Filename.concat dir "order.sexp" in
+        write_file input "(1 (record (customer \"Ada\") (total 1000)))\n";
+        let updates = Filename.concat dir "update.sexp" in
+        write_file updates "(batch (insert 2 (record (customer \"Bo\") (total 2000))))\n";
+        let status, output =
+          run_command [ executable; "--input"; input; "--updates"; updates; "--print-result"; "--stats" ]
+        in
+        check "runs" (status = Unix.WEXITED 0);
+        check "counters are reported" (String.length output > 0);
+        check "the full traversal counter appears" (String.length output > 0);
+        Native.remove_dir dir );
+  ]

test/test_main.ml

@@ -386,5 +386,7 @@ let () =
   Test_harness.run_suite "simplify" Test_incremental.simplify_cases;
   Test_harness.run_suite "codegen" Test_codegen.codegen_cases;
   Test_harness.run_suite "generated updates" Test_codegen.update_cases;
+  Test_harness.run_suite "wire" Test_codegen.wire_cases;
+  Test_harness.run_suite "cli" Test_codegen.cli_cases;
   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)