IncrementalWindowCount extends IncrementalWindowAggregate { public Datum updateWindow(List<Datum> new_data, List<Datum> old_data) { if (new_data.isEmpty() && old_data.isEmpty()) return currWindow; if (currWindow == null) { dim = new_data.get(0).metrics().getDimension(); } count += new_data.size() - old_data.size(); RealVector results = new ArrayRealVector(dim); results.set(count); currWindow = new Datum(new ArrayList<>(), results); return currWindow; } IncrementalWindowCount(MacroBaseConf conf); Datum updateWindow(List<Datum> new_data, List<Datum> old_data);  }
@Test public void testWindowUpdate() throws Exception { List<Datum> data = new ArrayList<>(); for (int i = 0; i < 20; ++i) { Datum d = TestUtils.createTimeDatum(i, i); data.add(d); } MacroBaseConf conf = new MacroBaseConf().set(MacroBaseConf.TIME_COLUMN, 0); IncrementalWindowCount windowCount = new IncrementalWindowCount(conf); Datum count = windowCount.updateWindow(data.subList(0, 10), new ArrayList<>()); assert(count.metrics().getEntry(1) == 10); count = windowCount.updateWindow(data.subList(10, 11), data.subList(0, 1)); assert(count.metrics().getEntry(1) == 10); count = windowCount.updateWindow(data.subList(11, 20), data.subList(1, 5)); assert(count.metrics().getEntry(1) == 15); count = windowCount.updateWindow(new ArrayList<>(), data.subList(5, 10)); assert(count.metrics().getEntry(1) == 10); }