@UdafFactory(description = "Compute average of column with type Long.",
      aggregateSchema = "STRUCT<SUM bigint, COUNT bigint>")
  public static TableUdaf<Long, Struct, Double> averageLong() {
    return getAverageImplementation(
        0L,
        STRUCT_LONG,
        (sum, newValue) -> sum.getInt64(SUM) + newValue,
        (sum, count) -> sum.getInt64(SUM) / count,
        (sum1, sum2) -> sum1.getInt64(SUM) + sum2.getInt64(SUM),
        (sum, valueToUndo) -> sum.getInt64(SUM) - valueToUndo);
  }
@Test
  public void shouldMergeAverages() {
    final TableUdaf<Long, Struct, Double> udaf = AverageUdaf.averageLong();

    Struct left = udaf.initialize();
    final Long[] leftValues = new Long[]  {1L, 1L, 1L, 1L, 1L};
    for (final Long thisValue : leftValues) {
      left = udaf.aggregate(thisValue, left);
    }

    Struct right = udaf.initialize();
    final Long[] rightValues = new Long[] {2L, 2L, 1L};
    for (final Long thisValue : rightValues) {
      right = udaf.aggregate(thisValue, right);
    }


    final Struct merged = udaf.merge(left, right);
    assertThat(8L, equalTo(merged.getInt64(COUNT)));
    assertThat(10L, equalTo(merged.getInt64(SUM)));

  }