class ListSumUdaf {

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

      @Override
      public Long initialize() {
        return 0L;
      }

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

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

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

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

      private long sumList(final List<Long> list) {
        return sum(list, initialize(), (a,b) -> a + b);
      }
    };
  }

  private  ListSumUdaf();

}