code: @Override
  	protected List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		CrossFunction<IN1, IN2, OUT> function = this.userFunction.getUserCodeObject();
  		
  		FunctionUtils.setFunctionRuntimeContext(function, ctx);
  		FunctionUtils.openFunction(function, this.parameters);
  
  		ArrayList<OUT> result = new ArrayList<OUT>(inputData1.size() * inputData2.size());
  		
  		TypeSerializer<IN1> inSerializer1 = getOperatorInfo().getFirstInputType().createSerializer(executionConfig);
  		TypeSerializer<IN2> inSerializer2 = getOperatorInfo().getSecondInputType().createSerializer(executionConfig);
  		TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  
  		for (IN1 element1 : inputData1) {
  			for (IN2 element2 : inputData2) {
  				IN1 copy1 = inSerializer1.copy(element1);
  				IN2 copy2 = inSerializer2.copy(element2);
  				OUT o = function.cross(copy1, copy2);
  				result.add(outSerializer.copy(o));
  			}
  		}
  
  		FunctionUtils.closeFunction(function);
  		return result;
  	}
docs: --------------------------------------------------------------------------------------------