docs: @Override
  	protected List<T> executeOnCollections(List<T> inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		// make sure we can handle empty inputs
  		if (inputData.isEmpty()) {
  			return Collections.emptyList();
  		}
  
  		ReduceFunction<T> function = this.userFunction.getUserCodeObject();
  
  		UnaryOperatorInformation<T, T> operatorInfo = getOperatorInfo();
  		TypeInformation<T> inputType = operatorInfo.getInputType();
  
  		int[] inputColumns = getKeyColumns(0);
  
  		if (!(inputType instanceof CompositeType) && inputColumns.length > 1) {
  			throw new InvalidProgramException("Grouping is only possible on composite types.");
  		}
  
  		FunctionUtils.setFunctionRuntimeContext(function, ctx);
  		FunctionUtils.openFunction(function, this.parameters);
  
  		TypeSerializer<T> serializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
  
  		if (inputColumns.length > 0) {
  			boolean[] inputOrderings = new boolean[inputColumns.length];
  			TypeComparator<T> inputComparator = inputType instanceof AtomicType
  					? ((AtomicType<T>) inputType).createComparator(false, executionConfig)
  					: ((CompositeType<T>) inputType).createComparator(inputColumns, inputOrderings, 0, executionConfig);
  
  			Map<TypeComparable<T>, T> aggregateMap = new HashMap<TypeComparable<T>, T>(inputData.size() / 10);
  
  			for (T next : inputData) {
  				TypeComparable<T> wrapper = new TypeComparable<T>(next, inputComparator);
  
  				T existing = aggregateMap.get(wrapper);
  				T result;
  
  				if (existing != null) {
  					result = function.reduce(existing, serializer.copy(next));
  				} else {
  					result = next;
  				}
  
  				result = serializer.copy(result);
  
  				aggregateMap.put(wrapper, result);
  
  			}
  
  			FunctionUtils.closeFunction(function);
  			return new ArrayList<T>(aggregateMap.values());
  		}
  		else {
  			T aggregate = inputData.get(0);
  			
  			aggregate = serializer.copy(aggregate);
  
  			for (int i = 1; i < inputData.size(); i++) {
  				T next = function.reduce(aggregate, serializer.copy(inputData.get(i)));
  				aggregate = serializer.copy(next);
  			}
  
  			FunctionUtils.setFunctionRuntimeContext(function, ctx);
  			
  			return Collections.singletonList(aggregate);
  		}
  	}