class ListSumUdaf {

  @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(), (a,b) -> a + b);
      }
    };
  }

  private  ListSumUdaf();

}