docs: @SuppressWarnings("unchecked")
  	@Override
  	protected List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig) throws Exception {
  		FlatJoinFunction<IN1, IN2, OUT> function = userFunction.getUserCodeObject();
  
  		FunctionUtils.setFunctionRuntimeContext(function, runtimeContext);
  		FunctionUtils.openFunction(function, this.parameters);
  
  		TypeInformation<IN1> leftInformation = getOperatorInfo().getFirstInputType();
  		TypeInformation<IN2> rightInformation = getOperatorInfo().getSecondInputType();
  		TypeInformation<OUT> outInformation = getOperatorInfo().getOutputType();
  
  		TypeSerializer<IN1> leftSerializer = leftInformation.createSerializer(executionConfig);
  		TypeSerializer<IN2> rightSerializer = rightInformation.createSerializer(executionConfig);
  
  		TypeComparator<IN1> leftComparator;
  		TypeComparator<IN2> rightComparator;
  
  		if (leftInformation instanceof AtomicType) {
  			leftComparator = ((AtomicType<IN1>) leftInformation).createComparator(true, executionConfig);
  		} else if (leftInformation instanceof CompositeType) {
  			int[] keyPositions = getKeyColumns(0);
  			boolean[] orders = new boolean[keyPositions.length];
  			Arrays.fill(orders, true);
  
  			leftComparator = ((CompositeType<IN1>) leftInformation).createComparator(keyPositions, orders, 0, executionConfig);
  		} else {
  			throw new RuntimeException("Type information for left input of type " + leftInformation.getClass()
  					.getCanonicalName() + " is not supported. Could not generate a comparator.");
  		}
  
  		if (rightInformation instanceof AtomicType) {
  			rightComparator = ((AtomicType<IN2>) rightInformation).createComparator(true, executionConfig);
  		} else if (rightInformation instanceof CompositeType) {
  			int[] keyPositions = getKeyColumns(1);
  			boolean[] orders = new boolean[keyPositions.length];
  			Arrays.fill(orders, true);
  
  			rightComparator = ((CompositeType<IN2>) rightInformation).createComparator(keyPositions, orders, 0, executionConfig);
  		} else {
  			throw new RuntimeException("Type information for right input of type " + rightInformation.getClass()
  					.getCanonicalName() + " is not supported. Could not generate a comparator.");
  		}
  
  		TypePairComparator<IN1, IN2> pairComparator = new GenericPairComparator<IN1, IN2>(leftComparator, rightComparator);
  
  		List<OUT> result = new ArrayList<OUT>();
  		Collector<OUT> collector = new CopyingListCollector<OUT>(result, outInformation.createSerializer(executionConfig));
  
  		Map<Integer, List<IN2>> probeTable = new HashMap<Integer, List<IN2>>();
  
  		//Build hash table
  		for (IN2 element : inputData2) {
  			List<IN2> list = probeTable.get(rightComparator.hash(element));
  			if (list == null) {
  				list = new ArrayList<IN2>();
  				probeTable.put(rightComparator.hash(element), list);
  			}
  
  			list.add(element);
  		}
  
  		//Probing
  		for (IN1 left : inputData1) {
  			List<IN2> matchingHashes = probeTable.get(leftComparator.hash(left));
  
  			if (matchingHashes != null) {
  				pairComparator.setReference(left);
  				for (IN2 right : matchingHashes) {
  					if (pairComparator.equalToReference(right)) {
  						function.join(leftSerializer.copy(left), rightSerializer.copy(right), collector);
  					}
  				}
  			}
  		}
  
  		FunctionUtils.closeFunction(function);
  
  		return result;
  	}