docs: @Override
  	protected List<OUT> executeOnCollections(List<IN> input, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		FlatMapFunction<IN, OUT> function = userFunction.getUserCodeObject();
  		
  		FunctionUtils.setFunctionRuntimeContext(function, ctx);
  		FunctionUtils.openFunction(function, parameters);
  
  		ArrayList<OUT> result = new ArrayList<OUT>(input.size());
  
  		TypeSerializer<IN> inSerializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
  		TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  
  		CopyingListCollector<OUT> resultCollector = new CopyingListCollector<OUT>(result, outSerializer);
  
  		for (IN element : input) {
  			IN inCopy = inSerializer.copy(element);
  			function.flatMap(inCopy, resultCollector);
  		}
  
  		FunctionUtils.closeFunction(function);
  
  		return result;
  	}