code: @Override
  	public BaseRow addInput(@Nullable BaseRow previousAcc, BaseRow input) throws Exception {
  		BaseRow currentAcc;
  		if (previousAcc == null) {
  			currentAcc = localAgg.createAccumulators();
  		} else {
  			currentAcc = previousAcc;
  		}
  
  		localAgg.setAccumulators(currentAcc);
  		localAgg.merge(input);
  		return localAgg.getAccumulators();
  	}
docs: The {@code previousAcc} is accumulator, but input is a row in &lt;key, accumulator&gt; schema,
  the specific generated {@link #localAgg} will project the {@code input} to accumulator
  in merge method.