test_class.file: flink-core/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
test_class.identifier: InnerJoinOperatorBaseTest
test_class.interfaces: implements Serializable
test_case.body: @Test
  	public void testJoinPlain(){
  		final FlatJoinFunction<String, String, Integer> joiner = new FlatJoinFunction<String, String, Integer>() {
  
  			@Override
  			public void join(String first, String second, Collector<Integer> out) throws Exception {
  				out.collect(first.length());
  				out.collect(second.length());
  			}
  		};
  
  		@SuppressWarnings({ "rawtypes", "unchecked" })
  		InnerJoinOperatorBase<String, String, Integer,
  						FlatJoinFunction<String, String,Integer> > base = new InnerJoinOperatorBase(joiner,
  				new BinaryOperatorInformation(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO,
  						BasicTypeInfo.INT_TYPE_INFO), new int[0], new int[0], "TestJoiner");
  
  		List<String> inputData1 = new ArrayList<String>(Arrays.asList("foo", "bar", "foobar"));
  		List<String> inputData2 = new ArrayList<String>(Arrays.asList("foobar", "foo"));
  		List<Integer> expected = new ArrayList<Integer>(Arrays.asList(3, 3, 6 ,6));
  
  		try {
  			ExecutionConfig executionConfig = new ExecutionConfig();
  			executionConfig.disableObjectReuse();
  			List<Integer> resultSafe = base.executeOnCollections(inputData1, inputData2, null, executionConfig);
  			executionConfig.enableObjectReuse();
  			List<Integer> resultRegular = base.executeOnCollections(inputData1, inputData2, null, executionConfig);
  
  			assertEquals(expected, resultSafe);
  			assertEquals(expected, resultRegular);
  		}
  		catch (Exception e) {
  			e.printStackTrace();
  			fail(e.getMessage());
  		}
  	}
test_case.class_method_signature: InnerJoinOperatorBaseTest.testJoinPlain()
test_case.full_signature: @Test public void testJoinPlain()
test_case.identifier: testJoinPlain
test_case.invocations: collect length collect length asList asList asList disableObjectReuse executeOnCollections enableObjectReuse executeOnCollections assertEquals assertEquals printStackTrace fail getMessage
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testJoinPlain()
focal_class.file: flink-core/src/main/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBase.java
focal_class.identifier: InnerJoinOperatorBase
focal_class.class_method_signature: InnerJoinOperatorBase.InnerJoinOperatorBase(UserCodeWrapper<FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.full_signature: public  InnerJoinOperatorBase(UserCodeWrapper<FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.identifier: InnerJoinOperatorBase
focal_class.modifiers: public
focal_class.parameters: (UserCodeWrapper<FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.signature:  InnerJoinOperatorBase(UserCodeWrapper<FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.class_method_signature: InnerJoinOperatorBase.InnerJoinOperatorBase(FT udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo, int[] keyPositions1,
  			int[] keyPositions2, String name)
focal_class.full_signature: public  InnerJoinOperatorBase(FT udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo, int[] keyPositions1,
  			int[] keyPositions2, String name)
focal_class.identifier: InnerJoinOperatorBase
focal_class.modifiers: public
focal_class.parameters: (FT udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo, int[] keyPositions1,
  			int[] keyPositions2, String name)
focal_class.signature:  InnerJoinOperatorBase(FT udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo, int[] keyPositions1,
  			int[] keyPositions2, String name)
focal_class.class_method_signature: InnerJoinOperatorBase.InnerJoinOperatorBase(Class<? extends FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.full_signature: public  InnerJoinOperatorBase(Class<? extends FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.identifier: InnerJoinOperatorBase
focal_class.modifiers: public
focal_class.parameters: (Class<? extends FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.signature:  InnerJoinOperatorBase(Class<? extends FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
  			int[] keyPositions1, int[] keyPositions2, String name)
focal_class.class_method_signature: InnerJoinOperatorBase.executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_class.full_signature: @SuppressWarnings("unchecked") @Override protected List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_class.identifier: executeOnCollections
focal_class.modifiers: @SuppressWarnings("unchecked") @Override protected
focal_class.parameters: (List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_class.return: List<OUT>
focal_class.signature: List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_class.superclass: extends JoinOperatorBase<IN1, IN2, OUT, FT>
focal_method.body: @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;
  	}
focal_method.class_method_signature: InnerJoinOperatorBase.executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_method.full_signature: @SuppressWarnings("unchecked") @Override protected List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_method.identifier: executeOnCollections
focal_method.invocations: getUserCodeObject setFunctionRuntimeContext openFunction getFirstInputType getOperatorInfo getSecondInputType getOperatorInfo getOutputType getOperatorInfo createSerializer createSerializer createComparator getKeyColumns fill createComparator getCanonicalName getClass createComparator getKeyColumns fill createComparator getCanonicalName getClass createSerializer get hash put hash add get hash setReference equalToReference join copy copy closeFunction
focal_method.modifiers: @SuppressWarnings("unchecked") @Override protected
focal_method.parameters: (List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)
focal_method.return: List<OUT>
focal_method.signature: List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext,
  			ExecutionConfig executionConfig)