InnerJoinOperatorBase extends JoinOperatorBase<IN1, IN2, OUT, FT> { @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>>(); 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); } 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; } InnerJoinOperatorBase(UserCodeWrapper<FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
			int[] keyPositions1, int[] keyPositions2, String name);  InnerJoinOperatorBase(FT udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo, int[] keyPositions1,
			int[] keyPositions2, String name);  InnerJoinOperatorBase(Class<? extends FT> udf, BinaryOperatorInformation<IN1, IN2, OUT> operatorInfo,
			int[] keyPositions1, int[] keyPositions2, String name);   }
@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 public void testJoinRich(){ final AtomicBoolean opened = new AtomicBoolean(false); final AtomicBoolean closed = new AtomicBoolean(false); final String taskName = "Test rich join function"; final RichFlatJoinFunction<String, String, Integer> joiner = new RichFlatJoinFunction<String, String, Integer>() { @Override public void open(Configuration parameters) throws Exception { opened.compareAndSet(false, true); assertEquals(0, getRuntimeContext().getIndexOfThisSubtask()); assertEquals(1, getRuntimeContext().getNumberOfParallelSubtasks()); } @Override public void close() throws Exception{ closed.compareAndSet(false, true); } @Override public void join(String first, String second, Collector<Integer> out) throws Exception { out.collect(first.length()); out.collect(second.length()); } }; InnerJoinOperatorBase<String, String, Integer, RichFlatJoinFunction<String, String, Integer>> base = new InnerJoinOperatorBase<String, String, Integer, RichFlatJoinFunction<String, String, Integer>>(joiner, new BinaryOperatorInformation<String, String, Integer>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO), new int[0], new int[0], taskName); final List<String> inputData1 = new ArrayList<String>(Arrays.asList("foo", "bar", "foobar")); final List<String> inputData2 = new ArrayList<String>(Arrays.asList("foobar", "foo")); final List<Integer> expected = new ArrayList<Integer>(Arrays.asList(3, 3, 6, 6)); try { final TaskInfo taskInfo = new TaskInfo(taskName, 0, 1, 0); final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<String, Accumulator<?, ?>>(); final HashMap<String, Future<Path>> cpTasks = new HashMap<>(); ExecutionConfig executionConfig = new ExecutionConfig(); executionConfig.disableObjectReuse(); List<Integer> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap), executionConfig); executionConfig.enableObjectReuse(); List<Integer> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap), executionConfig); assertEquals(expected, resultSafe); assertEquals(expected, resultRegular); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } assertTrue(opened.get()); assertTrue(closed.get()); }