From b86e6c33f9ea0c61f0003703de3b04a9347e5f47 Mon Sep 17 00:00:00 2001
From: Jeffrey Heer <jeffrey.heer@gmail.com>
Date: Wed, 13 Jun 2018 13:49:10 -0700
Subject: [PATCH] Add safeguards to ChangeSet, add tests. (vega/vega#1318)

---
 src/ChangeSet.js       |  71 +++++++++----
 test/changeset-test.js | 219 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 273 insertions(+), 17 deletions(-)
 create mode 100644 test/changeset-test.js

diff --git a/src/ChangeSet.js b/src/ChangeSet.js
index 272040a..f5762a6 100644
--- a/src/ChangeSet.js
+++ b/src/ChangeSet.js
@@ -47,49 +47,86 @@ export default function changeset() {
       return this;
     },
     pulse: function(pulse, tuples) {
-      var out, i, n, m, f, t, id;
+      var cur = {}, out = {}, i, n, m, f, t, id;
 
-      // add
-      for (i=0, n=add.length; i<n; ++i) {
-        pulse.add.push(ingest(add[i]));
+      // build lookup table of current tuples
+      for (i=0, n=tuples.length; i<n; ++i) {
+        cur[tupleid(tuples[i])] = 1;
       }
 
-      // remove
-      for (out={}, i=0, n=rem.length; i<n; ++i) {
+      // process individual tuples to remove
+      for (i=0, n=rem.length; i<n; ++i) {
         t = rem[i];
-        out[tupleid(t)] = t;
+        cur[tupleid(t)] = -1;
       }
+
+      // process predicate-based removals
       for (i=0, n=remp.length; i<n; ++i) {
         f = remp[i];
         tuples.forEach(function(t) {
-          if (f(t)) out[tupleid(t)] = t;
+          if (f(t)) cur[tupleid(t)] = -1;
         });
       }
-      for (id in out) pulse.rem.push(out[id]);
 
-      // modify
+      // process all add tuples
+      for (i=0, n=add.length; i<n; ++i) {
+        t = add[i];
+        id = tupleid(t);
+        if (cur[id]) {
+          // tuple already resides in dataset
+          // if flagged for both add and remove, cancel
+          cur[id] = 1;
+        } else {
+          // tuple does not reside in dataset, add
+          pulse.add.push(ingest(add[i]));
+        }
+      }
+
+      // populate pulse rem list
+      for (i=0, n=tuples.length; i<n; ++i) {
+        t = tuples[i];
+        if (cur[tupleid(t)] < 0) pulse.rem.push(t);
+      }
+
+      // modify helper method
       function modify(t, f, v) {
-        if (v) t[f] = v(t); else pulse.encode = f;
+        if (v) {
+          t[f] = v(t);
+        } else {
+          pulse.encode = f;
+        }
         if (!reflow) out[tupleid(t)] = t;
       }
-      for (out={}, i=0, n=mod.length; i<n; ++i) {
+
+      // process individual tuples to modify
+      for (i=0, n=mod.length; i<n; ++i) {
         m = mod[i];
-        modify(m.tuple, m.field, m.value);
-        pulse.modifies(m.field);
+        t = m.tuple;
+        f = m.field;
+        id = cur[tupleid(t)];
+        if (id > 0) {
+          modify(t, f, m.value);
+          pulse.modifies(f);
+        }
       }
+
+      // process predicate-based modifications
       for (i=0, n=modp.length; i<n; ++i) {
         m = modp[i];
         f = m.filter;
         tuples.forEach(function(t) {
-          if (f(t)) modify(t, m.field, m.value);
+          if (f(t) && cur[tupleid(t)] > 0) {
+            modify(t, m.field, m.value);
+          }
         });
         pulse.modifies(m.field);
       }
 
-      // reflow?
+      // upon reflow request, populate mod with all non-removed tuples
+      // otherwise, populate mod with modified tuples only
       if (reflow) {
         pulse.mod = rem.length || remp.length
-          ? tuples.filter(function(t) { return out.hasOwnProperty(tupleid(t)); })
+          ? tuples.filter(function(t) { return cur[tupleid(t)] > 0; })
           : tuples.slice();
       } else {
         for (id in out) pulse.mod.push(out[id]);
diff --git a/test/changeset-test.js b/test/changeset-test.js
new file mode 100644
index 0000000..07c3e34
--- /dev/null
+++ b/test/changeset-test.js
@@ -0,0 +1,219 @@
+var tape = require('tape'),
+    vega = require('../');
+
+tape('ChangeSet adds/removes/modifies tuples', function(test) {
+  var data = [
+    {key: 'a', value: 1},
+    {key: 'b', value: 2},
+    {key: 'c', value: 3}
+  ];
+
+  var extra = {key: 'd', value: 6},
+      pulse;
+
+  // add tuples
+  // should also assign tuple ids to each object
+  pulse = vega.changeset()
+    .insert(data)
+    .pulse(new vega.Pulse(), []);
+  test.deepEqual(pulse.add, data);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, []);
+  test.ok(data.every(vega.tupleid));
+
+  // modify tuple directly
+  pulse = vega.changeset()
+    .modify(data[0], 'value', 5)
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, [data[0]]);
+  test.equal(data[0].value, 5);
+
+  // modify tuples by predicate
+  pulse = vega.changeset()
+    .modify(
+      function(t) { return t.key === 'b'; },
+      'value',
+      function(t) { return t.value + 2; }
+    )
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, [data[1]]);
+  test.equal(data[1].value, 4);
+
+  // remove tuple directly
+  pulse = vega.changeset()
+    .remove(data[0])
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, [data[0]]);
+  test.deepEqual(pulse.mod, []);
+
+  // remove tuples by predicate
+  pulse = vega.changeset()
+    .remove(function(t) { return t.value < 5; })
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, data.slice(1));
+  test.deepEqual(pulse.mod, []);
+
+  // perform all three operations at once
+  // here, no tuples are implicated in more than one set
+  pulse = vega.changeset()
+    .insert(extra)
+    .remove(function(t) { return t.value === 3; })
+    .modify(data[1], 'key', 'e')
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, [extra]);
+  test.ok(vega.tupleid(extra));
+  test.deepEqual(pulse.rem, [data[2]]);
+  test.deepEqual(pulse.mod, [data[1]]);
+  test.equal(data[1].key, 'e');
+
+  test.end();
+});
+
+tape('ChangeSet handles conflicting changes', function(test) {
+  var data = [
+    {key: 'a', value: 1},
+    {key: 'b', value: 2},
+    {key: 'c', value: 3}
+  ];
+
+  var extra = {key: 'd', value: 6},
+      pulse;
+
+  // perform initial add, ingest tuples
+  pulse = vega.changeset()
+    .insert(data)
+    .pulse(new vega.Pulse(), []);
+  test.deepEqual(pulse.add, data);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, []);
+  test.ok(data.every(vega.tupleid));
+
+  // add + mod
+  // behavior: add if not already added, modify only if already present
+  pulse = vega.changeset()
+    .insert(data)
+    .modify(data[1], 'key', 'e')
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, [data[1]]);
+  test.equal(data[1].key, 'e');
+
+  pulse = vega.changeset()
+    .insert(extra)
+    .modify(extra, 'key', 'f')
+    .pulse(new vega.Pulse(), []);
+  test.deepEqual(pulse.add, [extra]);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, []);
+  test.equal(extra.key, 'd'); // unchanged
+
+  // rem + mod
+  // tuple should be removed, unmodified
+  pulse = vega.changeset()
+    .remove(data[0])
+    .modify(data[0], 'key', 'f')
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, [data[0]]);
+  test.deepEqual(pulse.mod, []);
+  test.equal(data[0].key, 'a'); // unchanged
+
+  pulse = vega.changeset()
+    .remove(function(t) { return t.value < 3; })
+    .modify(
+      function(t) { return t.key === 'a'; },
+      'value',
+      function(t) { return t.value + 2; }
+    )
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, data.slice(0, 2));
+  test.deepEqual(pulse.mod, []);
+  test.equal(data[0].value, 1); // unchanged
+
+  // add + rem
+  // add + rem + mod
+  // operations should cancel
+  pulse = vega.changeset()
+    .insert(data)
+    .remove(data)
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, []);
+
+  pulse = vega.changeset()
+    .insert(data[0])
+    .remove(function() { return true; })
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, data.slice(1));
+  test.deepEqual(pulse.mod, []);
+
+  pulse = vega.changeset()
+    .insert(data[2])
+    .remove(function() { return true; })
+    .modify(
+      function(t) { return t.value > 1; },
+      'value',
+      function(t) { return t.value + 2; }
+    )
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, data.slice(0, 2));
+  test.deepEqual(pulse.mod, [data[2]]);
+  test.equal(data[2].value, 5); // modified
+
+  test.end();
+});
+
+tape('ChangeSet handles reflow', function(test) {
+  var data = [
+    {key: 'a', value: 1},
+    {key: 'b', value: 2},
+    {key: 'c', value: 3}
+  ];
+
+  var extra = {key: 'd', value: 6},
+      pulse;
+
+  // initial add
+  pulse = vega.changeset()
+    .insert(data)
+    .pulse(new vega.Pulse(), []);
+  test.deepEqual(pulse.add, data);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, []);
+  test.ok(data.every(vega.tupleid));
+
+  // add, modify and reflow tuples
+  pulse = vega.changeset()
+    .insert(extra)
+    .modify(data[0], 'key', 'd')
+    .reflow()
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, [extra]);
+  test.deepEqual(pulse.rem, []);
+  test.deepEqual(pulse.mod, data);
+  test.equal(data[0].key, 'd');
+
+  // remove, modify and reflow tuples
+  pulse = vega.changeset()
+    .remove(function(t) { return t.value < 2; })
+    .modify(data[2], 'key', 'f')
+    .reflow()
+    .pulse(new vega.Pulse(), data);
+  test.deepEqual(pulse.add, []);
+  test.deepEqual(pulse.rem, data.slice(0, 1));
+  test.deepEqual(pulse.mod, data.slice(1));
+  test.equal(data[2].key, 'f');
+
+  test.end();
+});