@SuppressWarnings({"unchecked", "rawtypes"})
  @Override
  public GenericRow apply(
      final GenericKey k,
      final GenericRow rowValue,
      final GenericRow aggRowValue
  ) {
    final GenericRow result = GenericRow.fromList(aggRowValue.values());

    for (int idx = 0; idx < nonAggColumnCount; idx++) {
      result.set(idx, rowValue.get(idx));
    }

    for (int idx = nonAggColumnCount; idx < columnCount; idx++) {
      final TableAggregationFunction function = aggregateFunctions.get(idx - nonAggColumnCount);
      final Object argument = getCurrentValue(
              rowValue,
              function.getArgIndicesInValue(),
              function::convertToInput
      );
      final Object previous = result.get(idx);
      result.set(idx, function.undo(argument, previous));
    }

    return result;
  }
@Test
  public void shouldNotMutateParametersOnApply() {
    // Given:
    final GenericRow value = GenericRow.genericRow(1, 2L);
    final GenericRow agg = GenericRow.genericRow(1, 2L, 3);

    // When:
    final GenericRow result = aggregator.apply(key, value, agg);

    // Then:
    assertThat(value, is(GenericRow.genericRow(1, 2L)));
    assertThat(agg, is(GenericRow.genericRow(1, 2L, 3)));
    assertThat("invalid test", result, is(not(GenericRow.genericRow(1, 2L, 3))));
  }