@Override
  public GenericRow apply(final K k, final GenericRow rowValue, final GenericRow aggRowValue) {
    final GenericRow result = GenericRow.fromList(aggRowValue.values());

    // copy over group-by and aggregate parameter columns into the output row
    for (int idx = 0; idx < nonAggColumnCount; idx++) {
      result.set(idx, rowValue.get(idx));
    }

    // compute the aggregation and write it into the output row. Its assumed that
    // the columns written by this statement do not overlap with those written by
    // the above statement.
    for (int idx = nonAggColumnCount; idx < columnCount; idx++) {
      final KsqlAggregateFunction<Object, Object, Object> func = aggregateFunctionForColumn(idx);
      final Object currentValue = getCurrentValue(
              rowValue,
              func.getArgIndicesInValue(),
              func::convertToInput
      );
      final Object currentAggregate = result.get(idx);
      final Object newAggregate = func.aggregate(currentValue, currentAggregate);
      result.set(idx, newAggregate);
    }

    return result;
  }
@Test
  public void shouldNotMutateParametersOnApplyVariadicParam() {
    when(func1.convertToInput(any())).thenAnswer(
            (invocation) -> {
              List<?> inputs = invocation.getArgument(0, List.class);
              return Pair.of(inputs.get(0), new VariadicArgs<>(inputs.subList(1, 4)));
            }
    );
    when(func1.getArgIndicesInValue()).thenReturn(Arrays.asList(0, 1, 2, 3));

    // Given:
    final GenericRow value = GenericRow.genericRow(1, 2L, 3L, 4L);
    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, 3L, 4L)));
    assertThat(agg, is(GenericRow.genericRow(1, 2L, 3)));
    assertThat("invalid test", result, is(not(GenericRow.genericRow(1, 2L, 3))));
  }