@UdafFactory(description = "sum int values in a list into a single int")
  public static TableUdaf<List<Integer>, Integer, Integer> sumIntList() {
    return new TableUdaf<List<Integer>, Integer, Integer>() {

      @Override
      public Integer initialize() {
        return 0;
      }

      @Override
      public Integer aggregate(final List<Integer> valueToAdd, final Integer aggregateValue) {
        if (valueToAdd == null) {
          return aggregateValue;
        }
        return aggregateValue + sumList(valueToAdd);
      }

      @Override
      public Integer merge(final Integer aggOne, final Integer aggTwo) {
        return aggOne + aggTwo;
      }

      @Override
      public Integer map(final Integer agg) {
        return agg;
      }

      @Override
      public Integer undo(final List<Integer> valueToUndo, final Integer aggregateValue) {
        if (valueToUndo == null) {
          return aggregateValue;
        }
        return aggregateValue - sumList(valueToUndo);
      }

      private int sumList(final List<Integer> list) {
        return sum(list, initialize(), Integer::sum);
      }
    };
  }
@Test
  public void shouldMergeSums() {
    final TableUdaf<List<Integer>, Integer, Integer>  udaf = ListSumUdaf.sumIntList();

    final Integer[] leftValues = new Integer[] {1, 1, 1, 1};
    final List<Integer> leftList = Arrays.asList(leftValues);
    final Integer sumLeft = udaf.aggregate(leftList, 0);

    final Integer[] rightValues = new Integer[] {2, 2, 2};
    final List<Integer> rightList = Arrays.asList(rightValues);
    final Integer sumRight = udaf.aggregate(rightList, 0);


    final Integer merged = udaf.merge(sumLeft, sumRight);
    assertThat(10, equalTo(merged));
  }