docs: @Override
  	protected List<OUT> executeOnCollections(List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		GroupReduceFunction<IN, OUT> function = this.userFunction.getUserCodeObject();
  
  		UnaryOperatorInformation<IN, OUT> operatorInfo = getOperatorInfo();
  		TypeInformation<IN> inputType = operatorInfo.getInputType();
  
  		int[] keyColumns = getKeyColumns(0);
  		int[] sortColumns = keyColumns;
  		boolean[] sortOrderings = new boolean[sortColumns.length];
  
  		if (groupOrder != null) {
  			sortColumns = ArrayUtils.addAll(sortColumns, groupOrder.getFieldPositions());
  			sortOrderings = ArrayUtils.addAll(sortOrderings, groupOrder.getFieldSortDirections());
  		}
  
  		if(sortColumns.length == 0) { // => all reduce. No comparator
  			checkArgument(sortOrderings.length == 0);
  		} else {
  			final TypeComparator<IN> sortComparator = getTypeComparator(inputType, sortColumns, sortOrderings, executionConfig);
  			Collections.sort(inputData, new Comparator<IN>() {
  				@Override
  				public int compare(IN o1, IN o2) {
  					return sortComparator.compare(o1, o2);
  				}
  			});
  		}
  
  		FunctionUtils.setFunctionRuntimeContext(function, ctx);
  		FunctionUtils.openFunction(function, this.parameters);
  		
  		ArrayList<OUT> result = new ArrayList<OUT>();
  
  		if (inputData.size() > 0) {
  			final TypeSerializer<IN> inputSerializer = inputType.createSerializer(executionConfig);
  			if (keyColumns.length == 0) {
  				TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  				List<IN> inputDataCopy = new ArrayList<IN>(inputData.size());
  				for (IN in : inputData) {
  					inputDataCopy.add(inputSerializer.copy(in));
  				}
  				CopyingListCollector<OUT> collector = new CopyingListCollector<OUT>(result, outSerializer);
  
  				function.reduce(inputDataCopy, collector);
  			} else {
  				boolean[] keyOrderings = new boolean[keyColumns.length];
  				final TypeComparator<IN> comparator = getTypeComparator(inputType, keyColumns, keyOrderings, executionConfig);
  
  				ListKeyGroupedIterator<IN> keyedIterator = new ListKeyGroupedIterator<IN>(inputData, inputSerializer, comparator);
  
  				TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  				CopyingListCollector<OUT> collector = new CopyingListCollector<OUT>(result, outSerializer);
  
  				while (keyedIterator.nextKey()) {
  					function.reduce(keyedIterator.getValues(), collector);
  				}
  			}
  		}
  
  		FunctionUtils.closeFunction(function);
  		return result;
  	}