code: @SuppressWarnings("unchecked")
  	private static <IN, OUT, K> PlanUnwrappingGroupCombineOperator<IN, OUT, K> translateSelectorFunctionReducer(
  			SelectorFunctionKeys<IN, ?> rawKeys,
  			GroupCombineFunction<IN, OUT> function,
  			TypeInformation<OUT> outputType,
  			String name,
  			Operator<IN> input) {
  		final SelectorFunctionKeys<IN, K> keys = (SelectorFunctionKeys<IN, K>) rawKeys;
  
  		TypeInformation<Tuple2<K, IN>> typeInfoWithKey = KeyFunctions.createTypeWithKey(keys);
  		Operator<Tuple2<K, IN>> keyedInput = KeyFunctions.appendKeyExtractor(input, keys);
  
  		PlanUnwrappingGroupCombineOperator<IN, OUT, K> reducer =
  			new PlanUnwrappingGroupCombineOperator<>(function, keys, name, outputType, typeInfoWithKey);
  		reducer.setInput(keyedInput);
  
  		return reducer;
  	}
docs: --------------------------------------------------------------------------------------------