How to use the vega-dataflow.ChangeSet function in vega-dataflow

prototype.transform = function(input, reset) {
  log.debug(input, ['folding']);

  var fold = this,
      on = this.param('fields'),
      output = df.ChangeSet.create(input);

  if (reset) this._reset(input, output);

  this._fn(input.add, on, output.add);
  this._fn(input.mod, on, reset ? output.add : output.mod);
  input.rem.forEach(function(x) {
    output.rem.push.apply(output.rem, fold._cache[x._id]);
    fold._cache[x._id] = null;
  });

  // If we're only propagating values, don't mark key/value as updated.
  if (input.add.length || input.rem.length ||
      on.field.some(function(f) { return !!input.fields[f]; })) {
    output.fields[this._output.key] = 1;
    output.fields[this._output.value] = 1;
  }

prototype.model = function(model) {
  if (!arguments.length) return this._model;
  if (this._model !== model) {
    this._model = model;
    this._streamer = new df.Node(model);
    this._streamer._rank = -1;  // HACK: To reduce re-ranking churn.
    this._changeset = df.ChangeSet.create();
    if (this._handler) this._handler.model(model);
  }
  return this;
};

prototype._on_keep = function(cell) {
  // propagate sort, signals, fields, etc.
  if (this._facet) df.ChangeSet.copy(this._input, cell.ds._input);
};

var dl = require('datalib'),
    df = require('vega-dataflow'),
    ChangeSet = df.ChangeSet,
    Base = df.Graph.prototype,
    Node  = df.Node, // jshint ignore:line
    GroupBuilder = require('../scene/GroupBuilder'),
    visit = require('../scene/visit'),
    config = require('./config');

function Model(cfg) {
  this._defs = {};
  this._predicates = {};

  this._scene  = null;  // Root scenegraph node.
  this._groups = null;  // Index of group items.

  this._node = null;
  this._builder = null; // Top-level scenegraph builder.

var dl = require('datalib'),
    log = require('vega-logging'),
    Item = require('vega-scenegraph').Item,
    df = require('vega-dataflow'),
    Node = df.Node, // jshint ignore:line
    Deps = df.Dependencies,
    Tuple = df.Tuple,
    ChangeSet = df.ChangeSet,
    Sentinel = {},
    Encoder  = require('./Encoder'),
    Bounder  = require('./Bounder'),
    parseData = require('../parse/data');

function Builder() {
  return arguments.length ? this.init.apply(this, arguments) : this;
}

var Status = Builder.STATUS = {
  ENTER:  'enter',
  UPDATE: 'update',
  EXIT:   'exit'
};

var CONNECTED = 1, DISCONNECTED = 2;

var dl = require('datalib'),
    df = require('vega-dataflow'),
    ChangeSet = df.ChangeSet,
    Base = df.Graph.prototype,
    Node  = df.Node, // jshint ignore:line
    GroupBuilder = require('../scene/GroupBuilder'),
    visit = require('../scene/visit'),
    compiler = require('../parse/expr'),
    config = require('./config');

function Model(cfg) {
  this._defs = {};
  this._predicates = {};

  this._scene  = null;  // Root scenegraph node.
  this._groups = null;  // Index of group items.

  this._node = null;
  this._builder = null; // Top-level scenegraph builder.

prototype.transform = function(input) {
  log.debug(input, ['filtering']);

  var output = df.ChangeSet.create(input),
      graph = this._graph,
      skip = this._skip,
      test = this.param('test'),
      signals = graph.values(SIGNALS, this.dependency(SIGNALS));

  input.rem.forEach(function(x) {
    if (skip[x._id] !== 1) output.rem.push(x);
    else skip[x._id] = 0;
  });

  input.add.forEach(function(x) {
    if (test(x, null, signals)) output.add.push(x);
    else skip[x._id] = 1;
  });

  input.mod.forEach(function(x) {

function fire(registry, type, datum, evt) {
    var handlers = registry.handlers[type],
        node = registry.nodes[type],
        cs = df.ChangeSet.create(null, true),
        filtered = false,
        val, i, n, h;

    function invoke(f) {
      return !f.fn(datum, evt);
    }

    for (i=0, n=handlers.length; i

var dl = require('datalib'),
    df = require('vega-dataflow'),
    log = require('vega-logging'),
    ChangeSet = df.ChangeSet,
    Tuple = df.Tuple,
    Deps = df.Dependencies,
    Transform = require('./Transform'),
    Facetor = require('./Facetor');

function Aggregate(graph) {
  Transform.prototype.init.call(this, graph);

  Transform.addParameters(this, {
    groupby: {type: 'array'},
    summarize: {
      type: 'custom',
      set: function(summarize) {
        var signalDeps = {},
            tx = this._transform,
            i, len, f, fields, name, ops;

prototype._changeset = function(input) {
  var counts = this._counts,
      tuples = this._tuples || (this._tuples = {}),
      change = df.ChangeSet.create(input),
      out = this._output, w, t, c;

  for (w in counts) {
    t = tuples[w];
    c = counts[w] || 0;
    if (!t && c) {
      tuples[w] = (t = Tuple.ingest({}));
      t[out.text] = w;
      t[out.count] = c;
      change.add.push(t);
    } else if (c === 0) {
      if (t) change.rem.push(t);
      delete counts[w];
      delete tuples[w];
    } else if (t[out.count] !== c) {
      Tuple.set(t, out.count, c);