IncrementalWindowSum extends IncrementalWindowAggregate { public Datum updateWindow(List<Datum> new_data, List<Datum> old_data) { if (new_data.isEmpty() && old_data.isEmpty()) return currWindow; RealVector results; if (currWindow != null) { results = new ArrayRealVector(currWindow.metrics()); } else { dim = new_data.get(0).metrics().getDimension(); results = new ArrayRealVector(dim); } for (Datum d : new_data) { RealVector metrics = d.metrics(); for (int i = 0; i < dim; i ++) { if (timeColumn != null && i == timeColumn) continue; results.setEntry(i, results.getEntry(i) + metrics.getEntry(i)); } } for (Datum d : old_data) { RealVector metrics = d.metrics(); for (int i = 0; i < dim; i ++) { if (timeColumn != null && i == timeColumn) continue; results.setEntry(i, results.getEntry(i) - metrics.getEntry(i)); } } currWindow = new Datum(new ArrayList<>(), results); return currWindow; } IncrementalWindowSum(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); conf.set(AggregateConf.AGGREGATE_TYPE, AggregateConf.AggregateType.SUM); IncrementalWindowSum windowSum = new IncrementalWindowSum(conf); Datum sum = windowSum.updateWindow(data.subList(0, 10), new ArrayList<>()); assert(sum.metrics().getEntry(1) == 45); sum = windowSum.updateWindow(data.subList(10, 11), data.subList(0, 1)); assert(sum.metrics().getEntry(1) == 55); sum = windowSum.updateWindow(data.subList(11, 20), data.subList(1, 5)); assert(sum.metrics().getEntry(1) == 180); sum = windowSum.updateWindow(new ArrayList<>(), data.subList(5, 10)); assert(sum.metrics().getEntry(1) == 145); }