Introduce insert-or-update command.

The new command is useful when working with DDlog input relations with
primary keys.  Operations available on such relations so far are:

- insert - fails if a record with the same key exists
- delete-by-value, delete-by-key - fail if a record with the specified
  key does not exist
- modify - fails if a record with the specified key does not exist.

The new command is similar to insert, except that, if a record with the
same key as the one being inserted already exists, it replaces this
record with the new one instead of failing.  This is convenient in
scenarios where the client does not know if the key exists or cannot
easily issue a delete command.

The insert-or-update command is only applicable to input relations with
a primary key.

The CLI syntax for the new command is the same as for the `insert`
command, but using the `insert_ord_update` keyword:

```
insert_or_update WithKey(1, "bar3");
```

C API:

```
extern ddlog_cmd* ddlog_insert_or_update_cmd(table_id table, ddlog_record *rec);
```

rust/template/cmd_parser/parse.rs

@@ -196,6 +196,19 @@ fn test_command() {
             )
         ))
     );
+    assert_eq!(
+        parse_command(br"insert_or_update Rel1(true);"),
+        Ok((
+            &br""[..],
+            Command::Update(
+                UpdCmd::InsertOrUpdate(
+                    RelIdentifier::RelName(Cow::from("Rel1")),
+                    Record::PosStruct(Cow::from("Rel1"), vec![Record::Bool(true)])
+                ),
+                true
+            )
+        ))
+    );
     assert_eq!(
         parse_command(br" insert Rel1[true];"),
         Ok((
@@ -275,9 +288,10 @@ fn test_command() {
 }
 
 named!(update<&[u8], UpdCmd>,
-    alt!(do_parse!(apply!(sym,"insert")     >> rec: rel_record >> (UpdCmd::Insert(RelIdentifier::RelName(rec.0), rec.1)))    |
-         do_parse!(apply!(sym,"delete")     >> rec: rel_record >> (UpdCmd::Delete(RelIdentifier::RelName(rec.0), rec.1)))    |
-         do_parse!(apply!(sym,"delete_key") >> rec: rel_key    >> (UpdCmd::DeleteKey(RelIdentifier::RelName(rec.0), rec.1))) |
+    alt!(do_parse!(apply!(sym,"insert")     >> rec: rel_record >> (UpdCmd::Insert(RelIdentifier::RelName(rec.0), rec.1)))               |
+         do_parse!(apply!(sym,"insert_or_update") >> rec: rel_record >> (UpdCmd::InsertOrUpdate(RelIdentifier::RelName(rec.0), rec.1))) |
+         do_parse!(apply!(sym,"delete")     >> rec: rel_record >> (UpdCmd::Delete(RelIdentifier::RelName(rec.0), rec.1)))               |
+         do_parse!(apply!(sym,"delete_key") >> rec: rel_key    >> (UpdCmd::DeleteKey(RelIdentifier::RelName(rec.0), rec.1)))            |
          do_parse!(apply!(sym,"modify") >>
                    rec: rel_key         >>
                    apply!(sym, "<-")    >>

rust/template/ddlog.h

@@ -1061,6 +1061,17 @@ extern const ddlog_record* ddlog_get_struct_field(const ddlog_record* rec,
  */
 extern ddlog_cmd* ddlog_insert_cmd(table_id table, ddlog_record *rec);
 
+/*
+ * Create an insert-or-update command.
+ *
+ * `table` is the table to insert to.  The table must have a primary key.
+ * `rec` is the record to insert.  The function takes ownership of this record.
+ *
+ * Returns pointer to a new command, which can be sent to DDlog by calling
+ * `ddlog_apply_updates()`.
+ */
+extern ddlog_cmd* ddlog_insert_or_update_cmd(table_id table, ddlog_record *rec);
+
 /*
  * Create delete-by-value command.
  *

rust/template/differential_datalog/program.rs

@@ -888,6 +888,10 @@ pub enum Update<V> {
         relid: RelId,
         v: V,
     },
+    InsertOrUpdate {
+        relid: RelId,
+        v: V,
+    },
     DeleteValue {
         relid: RelId,
         v: V,
@@ -917,6 +921,12 @@ where
                 let _ = builder.field("v", v);
                 builder.finish()
             }
+            Update::InsertOrUpdate { relid, v } => {
+                let mut builder = f.debug_struct("InsertOrUpdate");
+                let _ = builder.field("relid", relid);
+                let _ = builder.field("v", v);
+                builder.finish()
+            }
             Update::DeleteValue { relid, v } => {
                 let mut builder = f.debug_struct("DeleteValue");
                 let _ = builder.field("relid", relid);
@@ -948,7 +958,7 @@ where
         let upd = match self {
             Update::Insert { relid, v } => (true, relid, v),
             Update::DeleteValue { relid, v } => (false, relid, v),
-            _ => panic!("Cannot serialize Modify/DeleteKey update"),
+            _ => panic!("Cannot serialize InsertOrUpdate/Modify/DeleteKey update"),
         };
 
         upd.serialize(serializer)
@@ -973,6 +983,7 @@ impl<V> Update<V> {
     pub fn relid(&self) -> RelId {
         match self {
             Update::Insert { relid, .. } => *relid,
+            Update::InsertOrUpdate { relid, .. } => *relid,
             Update::DeleteValue { relid, .. } => *relid,
             Update::DeleteKey { relid, .. } => *relid,
             Update::Modify { relid, .. } => *relid,
@@ -1366,6 +1377,9 @@ impl Program {
                                                   .ok_or_else(|| format!("no session found for relation ID {}", relid))?
                                                   .update(v, -1);
                                             },
+                                            Update::InsertOrUpdate{..} => {
+                                                return Err("InsertOrUpdate command received by worker thread".to_string());
+                                            },
                                             Update::DeleteKey{..} => {
                                                 // workers don't know about keys
                                                 return Err("DeleteKey command received by worker thread".to_string());
@@ -2105,6 +2119,12 @@ impl RunningProgram {
         self.apply_updates(vec![Update::Insert { relid: relid, v: v }].into_iter())
     }
 
+    /// Insert one record into input relation or replace existing record with the same
+    /// key.
+    pub fn insert_or_update(&mut self, relid: RelId, v: DDValue) -> Response<()> {
+        self.apply_updates(vec![Update::InsertOrUpdate { relid: relid, v: v }].into_iter())
+    }
+
     /// Remove a record if it exists in the relation.
     pub fn delete_value(&mut self, relid: RelId, v: DDValue) -> Response<()> {
         self.apply_updates(vec![Update::DeleteValue { relid: relid, v: v }].into_iter())
@@ -2326,6 +2346,12 @@ impl RunningProgram {
                 };
                 present
             }
+            Update::InsertOrUpdate { relid, .. } => {
+                return Err(format!(
+                    "Cannot perform insert_or_update operation on relation {} that does not have a primary key",
+                    relid
+                ));
+            }
             Update::DeleteKey { relid, .. } => {
                 return Err(format!(
                     "Cannot delete by key from relation {} that does not have a primary key",
@@ -2379,6 +2405,31 @@ impl RunningProgram {
                     Ok(())
                 }
             },
+            Update::InsertOrUpdate { relid, v } => match s.entry(key_func(&v)) {
+                hash_map::Entry::Occupied(mut oe) => {
+                    // Delete old value.
+                    let old = oe.get().clone();
+                    Self::delta_dec(ds, oe.get());
+                    updates.push(Update::DeleteValue { relid, v: old });
+
+                    // Insert new value.
+                    Self::delta_inc(ds, &v);
+                    updates.push(Update::Insert {
+                        relid,
+                        v: v.clone(),
+                    });
+
+                    // Update store
+                    *oe.get_mut() = v;
+                    Ok(())
+                }
+                hash_map::Entry::Vacant(ve) => {
+                    ve.insert(v.clone());
+                    Self::delta_inc(ds, &v);
+                    updates.push(Update::Insert { relid, v });
+                    Ok(())
+                }
+            },
             Update::DeleteValue { relid, v } => match s.entry(key_func(&v)) {
                 hash_map::Entry::Occupied(oe) => {
                     if *oe.get() != v {

rust/template/differential_datalog/record.rs

@@ -146,6 +146,7 @@ impl fmt::Display for RelIdentifier {
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum UpdCmd {
     Insert(RelIdentifier, Record),
+    InsertOrUpdate(RelIdentifier, Record),
     Delete(RelIdentifier, Record),
     DeleteKey(RelIdentifier, Record),
     Modify(RelIdentifier, Record, Record),
@@ -719,6 +720,18 @@ pub unsafe extern "C" fn ddlog_insert_cmd(table: libc::size_t, rec: *mut Record)
     Box::into_raw(Box::new(UpdCmd::Insert(RelIdentifier::RelId(table), *rec)))
 }
 
+#[no_mangle]
+pub unsafe extern "C" fn ddlog_insert_or_update_cmd(
+    table: libc::size_t,
+    rec: *mut Record,
+) -> *mut UpdCmd {
+    let rec = Box::from_raw(rec);
+    Box::into_raw(Box::new(UpdCmd::InsertOrUpdate(
+        RelIdentifier::RelId(table),
+        *rec,
+    )))
+}
+
 #[no_mangle]
 pub unsafe extern "C" fn ddlog_delete_val_cmd(
     table: libc::size_t,

rust/template/differential_datalog/replay.rs

@@ -139,6 +139,13 @@ pub trait RecordReplay: Write {
         write!(self, "insert {}[{}]", name, value)
     }
 
+    fn record_insert_or_update<V>(&mut self, name: &str, value: V) -> Result<()>
+    where
+        V: Display,
+    {
+        write!(self, "insert_or_update {}[{}]", name, value)
+    }
+
     /// Record an `UpdCmd`.
     fn record_upd_cmd<C>(&mut self, upd: &UpdCmd) -> Result<()>
     where
@@ -148,6 +155,9 @@ pub trait RecordReplay: Write {
             UpdCmd::Insert(rel, record) => {
                 self.record_insert(relident2name::<C>(rel).unwrap_or(&"???"), record)
             }
+            UpdCmd::InsertOrUpdate(rel, record) => {
+                self.record_insert_or_update(relident2name::<C>(rel).unwrap_or(&"???"), record)
+            }
             UpdCmd::Delete(rel, record) => write!(
                 self,
                 "delete {}[{}]",
@@ -179,6 +189,9 @@ pub trait RecordReplay: Write {
             Update::Insert { relid, v } => {
                 self.record_insert(C::relid2name(*relid).unwrap_or(&"???"), v)
             }
+            Update::InsertOrUpdate { relid, v } => {
+                self.record_insert_or_update(C::relid2name(*relid).unwrap_or(&"???"), v)
+            }
             Update::DeleteValue { relid, v } => write!(
                 self,
                 "delete {}[{}]",

rust/template/src/api.rs

@@ -474,6 +474,15 @@ pub fn updcmd2upd(c: &record::UpdCmd) -> Result<Update<DDValue>, String> {
                 v: val,
             })
         }
+        record::UpdCmd::InsertOrUpdate(rident, rec) => {
+            let relid =
+                Relations::try_from(rident).map_err(|_| format!("Unknown relation {}", rident))?;
+            let val = relval_from_record(relid, rec)?;
+            Ok(Update::InsertOrUpdate {
+                relid: relid as RelId,
+                v: val,
+            })
+        }
         record::UpdCmd::Delete(rident, rec) => {
             let relid =
                 Relations::try_from(rident).map_err(|()| format!("Unknown relation {}", rident))?;

test/datalog_tests/simple.dat

@@ -156,6 +156,12 @@ dump WithKeyDbg;
 
 start;
 
+insert_or_update WithKey(1, "bar3");
+
+commit dump_changes;
+
+start;
+
 insert Gangster("Billy Jack", "Vito Giacalone"),
 insert Gangster("Black Leo", "Leonardo Cellura"),
 insert Gangster("Black Sam", "Sam Todaro"),

test/datalog_tests/simple.dump.expected

@@ -595,6 +595,9 @@ WithKeyDbg{.key = 4, .val = "hello"}: +1
 WithKeyDbg: +3, -2
 WithKeyDbg{1,"bar2"}
 WithKeyDbg{4,"hello"}
+WithKeyDbg:
+WithKeyDbg{.key = 1, .val = "bar2"}: -1
+WithKeyDbg{.key = 1, .val = "bar3"}: +1
 Innocent:
 Innocent{.name = "Bill Smith"}: +1
 Innocent{.name = "John Doe"}: +1