code: @Override
  	protected List<OUT> executeOnCollections(List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		MapFunction<IN, OUT> function = this.userFunction.getUserCodeObject();
  		
  		FunctionUtils.setFunctionRuntimeContext(function, ctx);
  		FunctionUtils.openFunction(function, this.parameters);
  		
  		ArrayList<OUT> result = new ArrayList<OUT>(inputData.size());
  
  		TypeSerializer<IN> inSerializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
  		TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  
  		for (IN element : inputData) {
  			IN inCopy = inSerializer.copy(element);
  			OUT out = function.map(inCopy);
  			result.add(outSerializer.copy(out));
  		}
  
  		FunctionUtils.closeFunction(function);
  		
  		return result;
  	}
docs: --------------------------------------------------------------------------------------------