@UdafFactory(description = "Compute sample standard deviation of column with type Long.",
      aggregateSchema = "STRUCT<SUM bigint, COUNT bigint, M2 double>")
  public static TableUdaf<Long, Struct, Double> stdDevLong() {
    return getStdDevImplementation(
        0L,
        STRUCT_LONG,
        (agg, newValue) -> newValue + agg.getInt64(SUM),
        (agg, newValue) ->
            Double.valueOf(newValue * (agg.getInt64(COUNT) + 1) - (agg.getInt64(SUM) + newValue)),
        (agg1, agg2) ->
            Double.valueOf(
                agg1.getInt64(SUM) / agg1.getInt64(COUNT)
                    - agg2.getInt64(SUM) / agg2.getInt64(COUNT)),
        (agg1, agg2) -> agg1.getInt64(SUM) + agg2.getInt64(SUM),
        (agg, valueToRemove) -> agg.getInt64(SUM) - valueToRemove);
  }
@Test
  public void shouldMergeLongs() {
    final TableUdaf<Long, Struct, Double> udaf = stdDevLong();

    Struct left = udaf.initialize();
    final Long[] leftValues = new Long[] {1L, 2L, 3L, 4L, 5L};
    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(merged.getInt64(COUNT), equalTo(8L));
    assertThat(merged.getInt64(SUM), equalTo(20L));
    assertThat(merged.getFloat64(M2), equalTo(18.166666666666664));

    final double standardDev = udaf.map(merged);
    assertThat(standardDev, equalTo(2.595238095238095));
  }