code: @Override
  	protected List<OUT> executeOnCollections(List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		MapPartitionFunction<IN, OUT> function = this.userFunction.getUserCodeObject();
  		
  		FunctionUtils.setFunctionRuntimeContext(function, ctx);
  		FunctionUtils.openFunction(function, this.parameters);
  		
  		ArrayList<OUT> result = new ArrayList<OUT>(inputData.size() / 4);
  
  		TypeSerializer<IN> inSerializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
  		TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  
  		CopyingIterator<IN> source = new CopyingIterator<IN>(inputData.iterator(), inSerializer);
  		CopyingListCollector<OUT> resultCollector = new CopyingListCollector<OUT>(result, outSerializer);
  
  		function.mapPartition(source, resultCollector);
  
  		result.trimToSize();
  		FunctionUtils.closeFunction(function);
  		return result;
  	}
docs: --------------------------------------------------------------------------------------------