@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(), Long::sum);
      }
    };
  }
@Test
  public void shouldSumLongList() {
    final TableUdaf<List<Long>, Long, Long>  udaf = ListSumUdaf.sumLongList();
    final Long[] values = new Long[] {1L, 1L, 1L, 1L, 1L};
    final List<Long> list = Arrays.asList(values);
    final Long sum = udaf.aggregate(list, 0L);

    assertThat(5L, equalTo(sum));
  }