flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [3/3] flink git commit: [FLINK-2106] [runtime] Add Outer Join drivers and Outer Merge strategies to Runtime
Date Tue, 08 Sep 2015 14:58:52 GMT
[FLINK-2106] [runtime] Add Outer Join drivers and Outer Merge strategies to Runtime


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f3dee23b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f3dee23b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f3dee23b

Branch: refs/heads/master
Commit: f3dee23be8e317948a7414aa52182f0b7e2fb13e
Parents: 24f7fa9
Author: r-pogalz <r.pogalz@campus.tu-berlin.de>
Authored: Thu Aug 6 23:13:16 2015 +0200
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Tue Sep 8 16:57:51 2015 +0200

----------------------------------------------------------------------
 .../operators/AbstractOuterJoinDriver.java      | 202 +++++++++
 .../flink/runtime/operators/DriverStrategy.java |   6 +
 .../runtime/operators/FullOuterJoinDriver.java  | 107 +++++
 .../runtime/operators/LeftOuterJoinDriver.java  | 107 +++++
 .../runtime/operators/RightOuterJoinDriver.java | 107 +++++
 .../operators/sort/AbstractMergeIterator.java   |   7 +-
 .../sort/AbstractMergeOuterJoinIterator.java    |   2 +-
 .../sort/NonReusingMergeOuterJoinIterator.java  |   2 +-
 .../sort/ReusingMergeOuterJoinIterator.java     |   2 +-
 .../AbstractOuterJoinTaskExternalITCase.java    | 120 +++++
 .../operators/AbstractOuterJoinTaskTest.java    | 454 +++++++++++++++++++
 .../FullOuterJoinTaskExternalITCase.java        |  41 ++
 .../operators/FullOuterJoinTaskTest.java        |  41 ++
 .../LeftOuterJoinTaskExternalITCase.java        |  41 ++
 .../operators/LeftOuterJoinTaskTest.java        |  41 ++
 .../RightOuterJoinTaskExternalITCase.java       |  41 ++
 .../operators/RightOuterJoinTaskTest.java       |  41 ++
 ...bstractSortMergeOuterJoinIteratorITCase.java | 158 ++++---
 ...ReusingSortMergeOuterJoinIteratorITCase.java |   5 +-
 ...ReusingSortMergeOuterJoinIteratorITCase.java |   3 +-
 .../testutils/BinaryOperatorTestBase.java       | 433 ++++++++++++++++++
 21 files changed, 1881 insertions(+), 80 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/AbstractOuterJoinDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/AbstractOuterJoinDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/AbstractOuterJoinDriver.java
new file mode 100644
index 0000000..89b7709
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/AbstractOuterJoinDriver.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.util.JoinTaskIterator;
+import org.apache.flink.runtime.operators.util.TaskConfig;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The abstract outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
+ * joining pairs of records or joining records with null depending on the outer join type.
+ *
+ * @see FlatJoinFunction
+ */
+public abstract class AbstractOuterJoinDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT1, IT2, OT>, OT> {
+	
+	protected static final Logger LOG = LoggerFactory.getLogger(AbstractOuterJoinDriver.class);
+	
+	protected PactTaskContext<FlatJoinFunction<IT1, IT2, OT>, OT> taskContext;
+	
+	protected volatile JoinTaskIterator<IT1, IT2, OT> outerJoinIterator; // the iterator that does the actual outer join
+	protected volatile boolean running;
+	
+	// ------------------------------------------------------------------------
+	
+	@Override
+	public void setup(PactTaskContext<FlatJoinFunction<IT1, IT2, OT>, OT> context) {
+		this.taskContext = context;
+		this.running = true;
+	}
+	
+	@Override
+	public int getNumberOfInputs() {
+		return 2;
+	}
+	
+	@Override
+	public Class<FlatJoinFunction<IT1, IT2, OT>> getStubType() {
+		@SuppressWarnings("unchecked")
+		final Class<FlatJoinFunction<IT1, IT2, OT>> clazz = (Class<FlatJoinFunction<IT1, IT2, OT>>) (Class<?>) FlatJoinFunction.class;
+		return clazz;
+	}
+	
+	@Override
+	public int getNumberOfDriverComparators() {
+		return 2;
+	}
+	
+	@Override
+	public void prepare() throws Exception {
+		final TaskConfig config = this.taskContext.getTaskConfig();
+		
+		// obtain task manager's memory manager and I/O manager
+		final MemoryManager memoryManager = this.taskContext.getMemoryManager();
+		final IOManager ioManager = this.taskContext.getIOManager();
+		
+		// set up memory and I/O parameters
+		final double fractionAvailableMemory = config.getRelativeMemoryDriver();
+		final int numPages = memoryManager.computeNumberOfPages(fractionAvailableMemory);
+		
+		final DriverStrategy ls = config.getDriverStrategy();
+		
+		final MutableObjectIterator<IT1> in1 = this.taskContext.getInput(0);
+		final MutableObjectIterator<IT2> in2 = this.taskContext.getInput(1);
+		
+		// get serializers and comparators
+		final TypeSerializer<IT1> serializer1 = this.taskContext.<IT1>getInputSerializer(0).getSerializer();
+		final TypeSerializer<IT2> serializer2 = this.taskContext.<IT2>getInputSerializer(1).getSerializer();
+		final TypeComparator<IT1> comparator1 = this.taskContext.getDriverComparator(0);
+		final TypeComparator<IT2> comparator2 = this.taskContext.getDriverComparator(1);
+		
+		final TypePairComparatorFactory<IT1, IT2> pairComparatorFactory = config.getPairComparatorFactory(this.taskContext.getUserCodeClassLoader());
+		
+		if (pairComparatorFactory == null) {
+			throw new Exception("Missing pair comparator factory for outer join driver");
+		}
+		
+		ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+		boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+		
+		if (LOG.isDebugEnabled()) {
+			LOG.debug("Outer Join Driver object reuse: " + (objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+		}
+		
+		// create and return outer join iterator according to provided local strategy.
+		if (objectReuseEnabled) {
+			this.outerJoinIterator = getReusingOuterJoinIterator(
+					ls,
+					in1,
+					in2,
+					serializer1,
+					comparator1,
+					serializer2,
+					comparator2,
+					pairComparatorFactory,
+					memoryManager,
+					ioManager,
+					numPages
+			);
+		} else {
+			this.outerJoinIterator = getNonReusingOuterJoinIterator(
+					ls,
+					in1,
+					in2,
+					serializer1,
+					comparator1,
+					serializer2,
+					comparator2,
+					pairComparatorFactory,
+					memoryManager,
+					ioManager,
+					numPages
+			);
+		}
+		
+		this.outerJoinIterator.open();
+		
+		if (LOG.isDebugEnabled()) {
+			LOG.debug(this.taskContext.formatLogString("outer join task iterator ready."));
+		}
+	}
+	
+	@Override
+	public void run() throws Exception {
+		final FlatJoinFunction<IT1, IT2, OT> joinStub = this.taskContext.getStub();
+		final Collector<OT> collector = this.taskContext.getOutputCollector();
+		final JoinTaskIterator<IT1, IT2, OT> outerJoinIterator = this.outerJoinIterator;
+		
+		while (this.running && outerJoinIterator.callWithNextKey(joinStub, collector)) ;
+	}
+	
+	
+	@Override
+	public void cleanup() throws Exception {
+		if (this.outerJoinIterator != null) {
+			this.outerJoinIterator.close();
+			this.outerJoinIterator = null;
+		}
+	}
+	
+	@Override
+	public void cancel() {
+		this.running = false;
+		if (this.outerJoinIterator != null) {
+			this.outerJoinIterator.abort();
+		}
+	}
+	
+	protected abstract JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception;
+	
+	protected abstract JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception;
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DriverStrategy.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DriverStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DriverStrategy.java
index 3aadf2f..f42a275 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DriverStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DriverStrategy.java
@@ -74,6 +74,12 @@ public enum DriverStrategy {
 	// both inputs are merged, but materialized to the side for block-nested-loop-join among values with equal key
 	MERGE(JoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
 
+	LEFT_OUTER_MERGE(LeftOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
+
+	RIGHT_OUTER_MERGE(RightOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
+
+	FULL_OUTER_MERGE(FullOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
+
 	// co-grouping inputs
 	CO_GROUP(CoGroupDriver.class, null, PIPELINED, PIPELINED, 2),
 	// python-cogroup

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/FullOuterJoinDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/FullOuterJoinDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/FullOuterJoinDriver.java
new file mode 100644
index 0000000..998064a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/FullOuterJoinDriver.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
+import org.apache.flink.runtime.operators.sort.NonReusingMergeOuterJoinIterator;
+import org.apache.flink.runtime.operators.sort.ReusingMergeOuterJoinIterator;
+import org.apache.flink.runtime.operators.util.JoinTaskIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * The full outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
+ * joining pairs of records or joins records with null if no match is found.
+ */
+public class FullOuterJoinDriver<IT1, IT2, OT> extends AbstractOuterJoinDriver<IT1, IT2, OT> {
+	
+	@Override
+	protected JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception {
+		switch (driverStrategy) {
+			case FULL_OUTER_MERGE:
+				return new ReusingMergeOuterJoinIterator<>(
+						OuterJoinType.FULL,
+						in1,
+						in2,
+						serializer1,
+						comparator1,
+						serializer2,
+						comparator2,
+						pairComparatorFactory.createComparator12(comparator1, comparator2),
+						memoryManager,
+						ioManager,
+						numPages,
+						super.taskContext.getOwningNepheleTask()
+				);
+			default:
+				throw new Exception("Unsupported driver strategy for full outer join driver: " + driverStrategy.name());
+		}
+	}
+	
+	@Override
+	protected JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception {
+		switch (driverStrategy) {
+			case FULL_OUTER_MERGE:
+				return new NonReusingMergeOuterJoinIterator<>(
+						OuterJoinType.FULL,
+						in1,
+						in2,
+						serializer1,
+						comparator1,
+						serializer2,
+						comparator2,
+						pairComparatorFactory.createComparator12(comparator1, comparator2),
+						memoryManager,
+						ioManager,
+						numPages,
+						super.taskContext.getOwningNepheleTask()
+				);
+			default:
+				throw new Exception("Unsupported driver strategy for full outer join driver: " + driverStrategy.name());
+		}
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/LeftOuterJoinDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/LeftOuterJoinDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/LeftOuterJoinDriver.java
new file mode 100644
index 0000000..b8932af
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/LeftOuterJoinDriver.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
+import org.apache.flink.runtime.operators.sort.NonReusingMergeOuterJoinIterator;
+import org.apache.flink.runtime.operators.sort.ReusingMergeOuterJoinIterator;
+import org.apache.flink.runtime.operators.util.JoinTaskIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * The left outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
+ * joining pairs of records or joins records from the left side with null if no match is found.
+ */
+public class LeftOuterJoinDriver<IT1, IT2, OT> extends AbstractOuterJoinDriver<IT1, IT2, OT> {
+	
+	@Override
+	protected JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception {
+		switch (driverStrategy) {
+			case LEFT_OUTER_MERGE:
+				return new ReusingMergeOuterJoinIterator<>(
+						OuterJoinType.LEFT,
+						in1,
+						in2,
+						serializer1,
+						comparator1,
+						serializer2,
+						comparator2,
+						pairComparatorFactory.createComparator12(comparator1, comparator2),
+						memoryManager,
+						ioManager,
+						numPages,
+						super.taskContext.getOwningNepheleTask()
+				);
+			default:
+				throw new Exception("Unsupported driver strategy for left outer join driver: " + driverStrategy.name());
+		}
+	}
+	
+	@Override
+	protected JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception {
+		switch (driverStrategy) {
+			case LEFT_OUTER_MERGE:
+				return new NonReusingMergeOuterJoinIterator<>(
+						OuterJoinType.LEFT,
+						in1,
+						in2,
+						serializer1,
+						comparator1,
+						serializer2,
+						comparator2,
+						pairComparatorFactory.createComparator12(comparator1, comparator2),
+						memoryManager,
+						ioManager,
+						numPages,
+						super.taskContext.getOwningNepheleTask()
+				);
+			default:
+				throw new Exception("Unsupported driver strategy for left outer join driver: " + driverStrategy.name());
+		}
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RightOuterJoinDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RightOuterJoinDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RightOuterJoinDriver.java
new file mode 100644
index 0000000..c455dc4
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RightOuterJoinDriver.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
+import org.apache.flink.runtime.operators.sort.NonReusingMergeOuterJoinIterator;
+import org.apache.flink.runtime.operators.sort.ReusingMergeOuterJoinIterator;
+import org.apache.flink.runtime.operators.util.JoinTaskIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * The right outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
+ * joining pairs of records or joins records from the right side with null if no match is found.
+ */
+public class RightOuterJoinDriver<IT1, IT2, OT> extends AbstractOuterJoinDriver<IT1, IT2, OT> {
+	
+	@Override
+	protected JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception {
+		switch (driverStrategy) {
+			case RIGHT_OUTER_MERGE:
+				return new ReusingMergeOuterJoinIterator<>(
+						OuterJoinType.RIGHT,
+						in1,
+						in2,
+						serializer1,
+						comparator1,
+						serializer2,
+						comparator2,
+						pairComparatorFactory.createComparator12(comparator1, comparator2),
+						memoryManager,
+						ioManager,
+						numPages,
+						super.taskContext.getOwningNepheleTask()
+				);
+			default:
+				throw new Exception("Unsupported driver strategy for right outer join driver: " + driverStrategy.name());
+		}
+	}
+	
+	@Override
+	protected JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
+			DriverStrategy driverStrategy,
+			MutableObjectIterator<IT1> in1,
+			MutableObjectIterator<IT2> in2,
+			TypeSerializer<IT1> serializer1,
+			TypeComparator<IT1> comparator1,
+			TypeSerializer<IT2> serializer2,
+			TypeComparator<IT2> comparator2,
+			TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
+			MemoryManager memoryManager,
+			IOManager ioManager,
+			int numPages
+	) throws Exception {
+		switch (driverStrategy) {
+			case RIGHT_OUTER_MERGE:
+				return new NonReusingMergeOuterJoinIterator<>(
+						OuterJoinType.RIGHT,
+						in1,
+						in2,
+						serializer1,
+						comparator1,
+						serializer2,
+						comparator2,
+						pairComparatorFactory.createComparator12(comparator1, comparator2),
+						memoryManager,
+						ioManager,
+						numPages,
+						super.taskContext.getOwningNepheleTask()
+				);
+			default:
+				throw new Exception("Unsupported driver strategy for right outer join driver: " + driverStrategy.name());
+		}
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeIterator.java
index c01afc7..68e0d47 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeIterator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeIterator.java
@@ -87,7 +87,7 @@ public abstract class AbstractMergeIterator<T1, T2, O> implements JoinTaskIterat
 		this.iterator2 = createKeyGroupedIterator(input2, serializer2, comparator2.duplicate());
 
 		final int numPagesForSpiller = numMemoryPages > 20 ? 2 : 1;
-		this.blockIt = new NonReusingBlockResettableIterator<T2>(this.memoryManager, this.serializer2,
+		this.blockIt = new NonReusingBlockResettableIterator<>(this.memoryManager, this.serializer2,
 				(numMemoryPages - numPagesForSpiller), parentTask);
 		this.memoryForSpillingIterator = memoryManager.allocatePages(parentTask, numPagesForSpiller);
 	}
@@ -267,8 +267,9 @@ public abstract class AbstractMergeIterator<T1, T2, O> implements JoinTaskIterat
 			if (spillingRequired) {
 				// more data than would fit into one block. we need to wrap the other side in a spilling iterator
 				// create spilling iterator on first input
-				spillIt = new SpillingResettableIterator<T1>(spillVals, this.serializer1,
-						this.memoryManager, this.ioManager, this.memoryForSpillingIterator);
+				spillIt = new SpillingResettableIterator<>(
+						spillVals, this.serializer1, this.memoryManager, this.ioManager, this.memoryForSpillingIterator
+				);
 				leftSideIter = spillIt;
 				spillIt.open();
 

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeOuterJoinIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeOuterJoinIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeOuterJoinIterator.java
index 01b371e..f34639f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeOuterJoinIterator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeOuterJoinIterator.java
@@ -37,7 +37,7 @@ import java.util.Iterator;
  */
 public abstract class AbstractMergeOuterJoinIterator<T1, T2, O> extends AbstractMergeIterator<T1, T2, O> {
 
-	public static enum OuterJoinType {LEFT, RIGHT, FULL}
+	public enum OuterJoinType {LEFT, RIGHT, FULL}
 
 	private final OuterJoinType outerJoinType;
 

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NonReusingMergeOuterJoinIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NonReusingMergeOuterJoinIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NonReusingMergeOuterJoinIterator.java
index ac49ece..33510d3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NonReusingMergeOuterJoinIterator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NonReusingMergeOuterJoinIterator.java
@@ -48,7 +48,7 @@ public class NonReusingMergeOuterJoinIterator<T1, T2, O> extends AbstractMergeOu
 
 	@Override
 	protected <T> KeyGroupedIterator<T> createKeyGroupedIterator(MutableObjectIterator<T> input, TypeSerializer<T> serializer, TypeComparator<T> comparator) {
-		return new NonReusingKeyGroupedIterator<T>(input, comparator);
+		return new NonReusingKeyGroupedIterator<>(input, comparator);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReusingMergeOuterJoinIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReusingMergeOuterJoinIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReusingMergeOuterJoinIterator.java
index 0cefbc5..ce7bab4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReusingMergeOuterJoinIterator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReusingMergeOuterJoinIterator.java
@@ -53,7 +53,7 @@ public class ReusingMergeOuterJoinIterator<T1, T2, O> extends AbstractMergeOuter
 
 	@Override
 	protected <T> KeyGroupedIterator<T> createKeyGroupedIterator(MutableObjectIterator<T> input, TypeSerializer<T> serializer, TypeComparator<T> comparator) {
-		return new ReusingKeyGroupedIterator<T>(input, serializer, comparator);
+		return new ReusingKeyGroupedIterator<>(input, serializer, comparator);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskExternalITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskExternalITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskExternalITCase.java
new file mode 100644
index 0000000..7c8d04e
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskExternalITCase.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntComparator;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.RuntimePairComparatorFactory;
+import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.runtime.operators.testutils.BinaryOperatorTestBase;
+import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
+import org.apache.flink.util.Collector;
+import org.junit.Assert;
+import org.junit.Test;
+
+public abstract class AbstractOuterJoinTaskExternalITCase extends BinaryOperatorTestBase<FlatJoinFunction<Tuple2<Integer, Integer>,
+		Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
+	
+	private static final long HASH_MEM = 4 * 1024 * 1024;
+	
+	private static final long SORT_MEM = 3 * 1024 * 1024;
+	
+	private static final long BNLJN_MEM = 10 * PAGE_SIZE;
+	
+	private final double bnljn_frac;
+	
+	@SuppressWarnings("unchecked")
+	private final TypeComparator<Tuple2<Integer, Integer>> comparator1 = new TupleComparator<>(
+			new int[]{0},
+			new TypeComparator<?>[]{new IntComparator(true)},
+			new TypeSerializer<?>[]{IntSerializer.INSTANCE}
+	);
+	
+	@SuppressWarnings("unchecked")
+	private final TypeComparator<Tuple2<Integer, Integer>> comparator2 = new TupleComparator<>(
+			new int[]{0},
+			new TypeComparator<?>[]{new IntComparator(true)},
+			new TypeSerializer<?>[]{IntSerializer.INSTANCE}
+	);
+	
+	@SuppressWarnings("unchecked")
+	private final TypeSerializer<Tuple2<Integer, Integer>> serializer = new TupleSerializer<>(
+			(Class<Tuple2<Integer, Integer>>) (Class<?>) Tuple2.class,
+			new TypeSerializer<?>[]{IntSerializer.INSTANCE, IntSerializer.INSTANCE}
+	);
+	
+	private final CountingOutputCollector<Tuple2<Integer, Integer>> output = new CountingOutputCollector<>();
+	
+	private final DriverStrategy driverStrategy;
+	
+	public AbstractOuterJoinTaskExternalITCase(ExecutionConfig config, DriverStrategy driverStrategy) {
+		super(config, HASH_MEM, 2, SORT_MEM);
+		bnljn_frac = (double) BNLJN_MEM / this.getMemoryManager().getMemorySize();
+		this.driverStrategy = driverStrategy;
+	}
+	
+	@Test
+	public void testExternalSortOuterJoinTask() throws Exception {
+		final int keyCnt1 = 16384 * 4;
+		final int valCnt1 = 2;
+		
+		final int keyCnt2 = 8192;
+		final int valCnt2 = 4 * 2;
+		
+		final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
+		
+		setOutput(this.output);
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInputSorted(new UniformIntTupleGenerator(keyCnt1, valCnt1, false), serializer, this.comparator1.duplicate());
+		addInputSorted(new UniformIntTupleGenerator(keyCnt2, valCnt2, false), serializer, this.comparator2.duplicate());
+		testDriver(testTask, MockJoinStub.class);
+		
+		Assert.assertEquals("Wrong result set size.", expCnt, this.output.getNumberOfRecords());
+	}
+	
+	protected abstract int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2);
+	
+	protected abstract AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver();
+	
+	// =================================================================================================
+
+	@SuppressWarnings("serial")
+	public static final class MockJoinStub implements FlatJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
+		
+		@Override
+		public void join(Tuple2<Integer, Integer> first, Tuple2<Integer, Integer> second, Collector<Tuple2<Integer, Integer>> out) throws Exception {
+			out.collect(first != null ? first : second);
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskTest.java
new file mode 100644
index 0000000..ad784b5
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskTest.java
@@ -0,0 +1,454 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import com.google.common.base.Throwables;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntComparator;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.RuntimePairComparatorFactory;
+import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.runtime.operators.testutils.BinaryOperatorTestBase;
+import org.apache.flink.runtime.operators.testutils.DelayingIterator;
+import org.apache.flink.runtime.operators.testutils.DiscardingOutputCollector;
+import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
+import org.apache.flink.runtime.operators.testutils.InfiniteIntTupleIterator;
+import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
+import org.apache.flink.util.Collector;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
+
+public abstract class AbstractOuterJoinTaskTest extends BinaryOperatorTestBase<FlatJoinFunction<Tuple2<Integer, Integer>,
+		Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
+	
+	private static final long HASH_MEM = 6 * 1024 * 1024;
+	
+	private static final long SORT_MEM = 3 * 1024 * 1024;
+	
+	private static final int NUM_SORTER = 2;
+	
+	private static final long BNLJN_MEM = 10 * PAGE_SIZE;
+	
+	private final double bnljn_frac;
+	
+	private final DriverStrategy driverStrategy;
+	
+	@SuppressWarnings("unchecked")
+	private final TypeComparator<Tuple2<Integer, Integer>> comparator1 = new TupleComparator<>(
+			new int[]{0},
+			new TypeComparator<?>[]{new IntComparator(true)},
+			new TypeSerializer<?>[]{IntSerializer.INSTANCE}
+	);
+	
+	@SuppressWarnings("unchecked")
+	private final TypeComparator<Tuple2<Integer, Integer>> comparator2 = new TupleComparator<>(
+			new int[]{0},
+			new TypeComparator<?>[]{new IntComparator(true)},
+			new TypeSerializer<?>[]{IntSerializer.INSTANCE}
+	);
+	
+	private final List<Tuple2<Integer, Integer>> outList = new ArrayList<>();
+	
+	@SuppressWarnings("unchecked")
+	private final TypeSerializer<Tuple2<Integer, Integer>> serializer = new TupleSerializer<>(
+			(Class<Tuple2<Integer, Integer>>) (Class<?>) Tuple2.class,
+			new TypeSerializer<?>[]{IntSerializer.INSTANCE, IntSerializer.INSTANCE}
+	);
+	
+	
+	public AbstractOuterJoinTaskTest(ExecutionConfig config, DriverStrategy driverStrategy) {
+		super(config, HASH_MEM, NUM_SORTER, SORT_MEM);
+		bnljn_frac = (double) BNLJN_MEM / this.getMemoryManager().getMemorySize();
+		this.driverStrategy = driverStrategy;
+	}
+	
+	@Test
+	public void testSortBoth1OuterJoinTask() throws Exception {
+		final int keyCnt1 = 20;
+		final int valCnt1 = 1;
+		
+		final int keyCnt2 = 10;
+		final int valCnt2 = 2;
+		
+		testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
+	}
+	
+	@Test
+	public void testSortBoth2OuterJoinTask() throws Exception {
+		final int keyCnt1 = 20;
+		final int valCnt1 = 1;
+		
+		final int keyCnt2 = 20;
+		final int valCnt2 = 1;
+		
+		testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
+	}
+	
+	@Test
+	public void testSortBoth3OuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 1;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 20;
+		
+		testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
+	}
+	
+	@Test
+	public void testSortBoth4OuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 20;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 1;
+		
+		testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
+	}
+	
+	@Test
+	public void testSortBoth5OuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 20;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 20;
+		
+		testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
+	}
+	
+	@Test
+	public void testSortBoth6OuterJoinTask() throws Exception {
+		int keyCnt1 = 10;
+		int valCnt1 = 1;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 2;
+		
+		testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
+	}
+	
+	private void testSortBothOuterJoinTask(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) throws Exception {
+		setOutput(this.outList, this.serializer);
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInputSorted(new UniformIntTupleGenerator(keyCnt1, valCnt1, false), this.serializer, this.comparator1.duplicate());
+		addInputSorted(new UniformIntTupleGenerator(keyCnt2, valCnt2, false), this.serializer, this.comparator2.duplicate());
+		testDriver(testTask, MockJoinStub.class);
+		
+		final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
+		
+		Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
+		
+		this.outList.clear();
+	}
+	
+	@Test
+	public void testSortFirstOuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 20;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 20;
+		
+		setOutput(this.outList, this.serializer);
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInputSorted(new UniformIntTupleGenerator(keyCnt1, valCnt1, false), this.serializer, this.comparator1.duplicate());
+		addInput(new UniformIntTupleGenerator(keyCnt2, valCnt2, true), this.serializer);
+		testDriver(testTask, MockJoinStub.class);
+		
+		final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
+		
+		Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
+		
+		this.outList.clear();
+	}
+	
+	@Test
+	public void testSortSecondOuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 20;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 20;
+		
+		setOutput(this.outList, this.serializer);
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInput(new UniformIntTupleGenerator(keyCnt1, valCnt1, true), this.serializer);
+		addInputSorted(new UniformIntTupleGenerator(keyCnt2, valCnt2, false), this.serializer, this.comparator2.duplicate());
+		testDriver(testTask, MockJoinStub.class);
+		
+		final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
+		
+		Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
+		
+		this.outList.clear();
+	}
+	
+	@Test
+	public void testMergeOuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 20;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 20;
+		
+		setOutput(this.outList, this.serializer);
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInput(new UniformIntTupleGenerator(keyCnt1, valCnt1, true), this.serializer);
+		addInput(new UniformIntTupleGenerator(keyCnt2, valCnt2, true), this.serializer);
+		
+		testDriver(testTask, MockJoinStub.class);
+		
+		final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
+		
+		Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
+		
+		this.outList.clear();
+	}
+	
+	@Test(expected = ExpectedTestException.class)
+	public void testFailingOuterJoinTask() throws Exception {
+		int keyCnt1 = 20;
+		int valCnt1 = 20;
+		
+		int keyCnt2 = 20;
+		int valCnt2 = 20;
+		
+		setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInput(new UniformIntTupleGenerator(keyCnt1, valCnt1, true), this.serializer);
+		addInput(new UniformIntTupleGenerator(keyCnt2, valCnt2, true), this.serializer);
+		
+		testDriver(testTask, MockFailingJoinStub.class);
+	}
+	
+	@Test
+	public void testCancelOuterJoinTaskWhileSort1() throws Exception {
+		setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInputSorted(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer, this.comparator1.duplicate());
+		addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
+		
+		final AtomicReference<Throwable> error = new AtomicReference<>();
+		
+		final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileSort1()") {
+			@Override
+			public void run() {
+				try {
+					testDriver(testTask, MockJoinStub.class);
+				} catch (Throwable t) {
+					error.set(t);
+				}
+			}
+		};
+		taskRunner.start();
+		
+		Thread.sleep(1000);
+		
+		cancel();
+		taskRunner.interrupt();
+		
+		taskRunner.join(60000);
+		
+		assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
+		
+		final Throwable taskError = error.get();
+		if (taskError != null) {
+			fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
+		}
+	}
+	
+	@Test
+	public void testCancelOuterJoinTaskWhileSort2() throws Exception {
+		setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(this.driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 1), this.serializer);
+		addInputSorted(new DelayingIterator<>(new InfiniteIntTupleIterator(), 1), this.serializer, this.comparator2.duplicate());
+		
+		final AtomicReference<Throwable> error = new AtomicReference<>();
+		
+		final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileSort2()") {
+			@Override
+			public void run() {
+				try {
+					testDriver(testTask, MockJoinStub.class);
+				} catch (Throwable t) {
+					error.set(t);
+				}
+			}
+		};
+		taskRunner.start();
+		
+		Thread.sleep(1000);
+		
+		cancel();
+		taskRunner.interrupt();
+		
+		taskRunner.join(60000);
+		
+		assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
+		
+		final Throwable taskError = error.get();
+		if (taskError != null) {
+			fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
+		}
+	}
+	
+	@Test
+	public void testCancelOuterJoinTaskWhileRunning() throws Exception {
+		setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
+		addDriverComparator(this.comparator1);
+		addDriverComparator(this.comparator2);
+		getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
+		getTaskConfig().setDriverStrategy(driverStrategy);
+		getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
+		setNumFileHandlesForSort(4);
+		
+		final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
+		
+		addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
+		addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
+		
+		final AtomicReference<Throwable> error = new AtomicReference<>();
+		
+		final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileRunning()") {
+			@Override
+			public void run() {
+				try {
+					testDriver(testTask, MockJoinStub.class);
+				} catch (Throwable t) {
+					error.set(t);
+				}
+			}
+		};
+		taskRunner.start();
+		
+		Thread.sleep(1000);
+		
+		cancel();
+		taskRunner.interrupt();
+		
+		taskRunner.join(60000);
+		
+		assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
+		
+		final Throwable taskError = error.get();
+		if (taskError != null) {
+			fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
+		}
+	}
+	
+	protected abstract AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver();
+	
+	protected abstract int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2);
+	
+	// =================================================================================================
+
+	@SuppressWarnings("serial")
+	public static final class MockJoinStub implements FlatJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
+		
+		@Override
+		public void join(Tuple2<Integer, Integer> first, Tuple2<Integer, Integer> second, Collector<Tuple2<Integer, Integer>> out) throws Exception {
+			out.collect(first != null ? first : second);
+		}
+	}
+
+	@SuppressWarnings("serial")
+	public static final class MockFailingJoinStub implements FlatJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
+		
+		private int cnt = 0;
+		
+		@Override
+		public void join(Tuple2<Integer, Integer> first, Tuple2<Integer, Integer> second, Collector<Tuple2<Integer, Integer>> out) throws Exception {
+			if (++this.cnt >= 10) {
+				throw new ExpectedTestException();
+			}
+			out.collect(first != null ? first : second);
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskExternalITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskExternalITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskExternalITCase.java
new file mode 100644
index 0000000..d52f5fb
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskExternalITCase.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+public class FullOuterJoinTaskExternalITCase extends AbstractOuterJoinTaskExternalITCase {
+	
+	
+	public FullOuterJoinTaskExternalITCase(ExecutionConfig config) {
+		super(config, DriverStrategy.FULL_OUTER_MERGE);
+	}
+	
+	@Override
+	protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
+		return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : (keyCnt1 - keyCnt2) * valCnt1);
+	}
+	
+	@Override
+	protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
+		return new FullOuterJoinDriver<>();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskTest.java
new file mode 100644
index 0000000..d3296f6
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+public class FullOuterJoinTaskTest extends AbstractOuterJoinTaskTest {
+	
+	
+	public FullOuterJoinTaskTest(ExecutionConfig config) {
+		super(config, DriverStrategy.FULL_OUTER_MERGE);
+	}
+	
+	@Override
+	protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
+		return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : (keyCnt1 - keyCnt2) * valCnt1);
+	}
+	
+	@Override
+	protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
+		return new FullOuterJoinDriver<>();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskExternalITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskExternalITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskExternalITCase.java
new file mode 100644
index 0000000..89d68f2
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskExternalITCase.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+public class LeftOuterJoinTaskExternalITCase extends AbstractOuterJoinTaskExternalITCase {
+	
+	
+	public LeftOuterJoinTaskExternalITCase(ExecutionConfig config) {
+		super(config, DriverStrategy.LEFT_OUTER_MERGE);
+	}
+	
+	@Override
+	protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
+		return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt1 > keyCnt2 ? (keyCnt1 - keyCnt2) * valCnt1 : 0);
+	}
+	
+	@Override
+	protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
+		return new LeftOuterJoinDriver<>();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskTest.java
new file mode 100644
index 0000000..9a1ec8f
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+public class LeftOuterJoinTaskTest extends AbstractOuterJoinTaskTest {
+	
+	
+	public LeftOuterJoinTaskTest(ExecutionConfig config) {
+		super(config, DriverStrategy.LEFT_OUTER_MERGE);
+	}
+	
+	@Override
+	protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
+		return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt1 > keyCnt2 ? (keyCnt1 - keyCnt2) * valCnt1 : 0);
+	}
+	
+	@Override
+	protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
+		return new LeftOuterJoinDriver<>();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskExternalITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskExternalITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskExternalITCase.java
new file mode 100644
index 0000000..4e7df4b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskExternalITCase.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+public class RightOuterJoinTaskExternalITCase extends AbstractOuterJoinTaskExternalITCase {
+	
+	
+	public RightOuterJoinTaskExternalITCase(ExecutionConfig config) {
+		super(config, DriverStrategy.RIGHT_OUTER_MERGE);
+	}
+	
+	@Override
+	protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
+		return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : 0);
+	}
+	
+	@Override
+	protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
+		return new RightOuterJoinDriver<>();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskTest.java
new file mode 100644
index 0000000..506e95b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+public class RightOuterJoinTaskTest extends AbstractOuterJoinTaskTest {
+	
+	
+	public RightOuterJoinTaskTest(ExecutionConfig config) {
+		super(config, DriverStrategy.RIGHT_OUTER_MERGE);
+	}
+	
+	@Override
+	protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
+		return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : 0);
+	}
+	
+	@Override
+	protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
+		return new RightOuterJoinDriver<>();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/AbstractSortMergeOuterJoinIteratorITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/AbstractSortMergeOuterJoinIteratorITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/AbstractSortMergeOuterJoinIteratorITCase.java
index 1fbe025..d4da9d3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/AbstractSortMergeOuterJoinIteratorITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/AbstractSortMergeOuterJoinIteratorITCase.java
@@ -39,7 +39,12 @@ import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
 import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
 import org.apache.flink.runtime.memorymanager.MemoryManager;
 import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
-import org.apache.flink.runtime.operators.testutils.*;
+import org.apache.flink.runtime.operators.testutils.CollectionIterator;
+import org.apache.flink.runtime.operators.testutils.DiscardingOutputCollector;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.operators.testutils.Match;
+import org.apache.flink.runtime.operators.testutils.MatchRemovingJoiner;
+import org.apache.flink.runtime.operators.testutils.SimpleTupleJoinFunction;
 import org.apache.flink.runtime.operators.testutils.TestData.TupleConstantValueIterator;
 import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator;
 import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.KeyMode;
@@ -52,7 +57,13 @@ import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
 
 public abstract class AbstractSortMergeOuterJoinIteratorITCase {
@@ -61,11 +72,6 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 	private static final int MEMORY_SIZE = 1024 * 1024 * 16;
 	private static final int PAGES_FOR_BNLJN = 2;
 
-	// the size of the left and right inputs
-	private static final int INPUT_1_SIZE = 20000;
-
-	private static final int INPUT_2_SIZE = 1000;
-
 	// random seeds for the left and right input data generators
 	private static final long SEED1 = 561349061987311L;
 
@@ -76,9 +82,7 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 
 	private IOManager ioManager;
 	private MemoryManager memoryManager;
-
-	private TupleTypeInfo<Tuple2<String, String>> typeInfo1;
-	private TupleTypeInfo<Tuple2<String, Integer>> typeInfo2;
+	
 	private TupleSerializer<Tuple2<String, String>> serializer1;
 	private TupleSerializer<Tuple2<String, Integer>> serializer2;
 	private TypeComparator<Tuple2<String, String>> comparator1;
@@ -90,14 +94,14 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 	public void beforeTest() {
 		ExecutionConfig config = new ExecutionConfig();
 		config.disableObjectReuse();
-
-		typeInfo1 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
-		typeInfo2 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class);
+		
+		TupleTypeInfo<Tuple2<String, String>> typeInfo1 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
+		TupleTypeInfo<Tuple2<String, Integer>> typeInfo2 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class);
 		serializer1 = typeInfo1.createSerializer(config);
 		serializer2 = typeInfo2.createSerializer(config);
 		comparator1 = typeInfo1.createComparator(new int[]{0}, new boolean[]{true}, 0, config);
 		comparator2 = typeInfo2.createComparator(new int[]{0}, new boolean[]{true}, 0, config);
-		pairComp = new GenericPairComparator<Tuple2<String, String>, Tuple2<String, Integer>>(comparator1, comparator2);
+		pairComp = new GenericPairComparator<>(comparator1, comparator2);
 
 		this.memoryManager = new DefaultMemoryManager(MEMORY_SIZE, 1);
 		this.ioManager = new IOManagerAsync();
@@ -121,17 +125,18 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	protected void testFullOuterWithSample() throws Exception {
 		CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
-				new Tuple2<String, String>("Jack", "Engineering"),
-				new Tuple2<String, String>("Tim", "Sales"),
-				new Tuple2<String, String>("Zed", "HR")
+				new Tuple2<>("Jack", "Engineering"),
+				new Tuple2<>("Tim", "Sales"),
+				new Tuple2<>("Zed", "HR")
 		);
 		CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
-				new Tuple2<String, Integer>("Allison", 100),
-				new Tuple2<String, Integer>("Jack", 200),
-				new Tuple2<String, Integer>("Zed", 150),
-				new Tuple2<String, Integer>("Zed", 250)
+				new Tuple2<>("Allison", 100),
+				new Tuple2<>("Jack", 200),
+				new Tuple2<>("Zed", 150),
+				new Tuple2<>("Zed", 250)
 		);
 
 		OuterJoinType outerJoinType = OuterJoinType.FULL;
@@ -148,17 +153,18 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		Assert.assertEquals(expected, actual);
 	}
 
+	@SuppressWarnings("unchecked")
 	protected void testLeftOuterWithSample() throws Exception {
 		CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
-				new Tuple2<String, String>("Jack", "Engineering"),
-				new Tuple2<String, String>("Tim", "Sales"),
-				new Tuple2<String, String>("Zed", "HR")
+				new Tuple2<>("Jack", "Engineering"),
+				new Tuple2<>("Tim", "Sales"),
+				new Tuple2<>("Zed", "HR")
 		);
 		CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
-				new Tuple2<String, Integer>("Allison", 100),
-				new Tuple2<String, Integer>("Jack", 200),
-				new Tuple2<String, Integer>("Zed", 150),
-				new Tuple2<String, Integer>("Zed", 250)
+				new Tuple2<>("Allison", 100),
+				new Tuple2<>("Jack", 200),
+				new Tuple2<>("Zed", 150),
+				new Tuple2<>("Zed", 250)
 		);
 
 		List<Tuple4<String, String, String, Object>> actual = computeOuterJoin(input1, input2, OuterJoinType.LEFT);
@@ -173,17 +179,18 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		Assert.assertEquals(expected, actual);
 	}
 
+	@SuppressWarnings("unchecked")
 	protected void testRightOuterWithSample() throws Exception {
 		CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
-				new Tuple2<String, String>("Jack", "Engineering"),
-				new Tuple2<String, String>("Tim", "Sales"),
-				new Tuple2<String, String>("Zed", "HR")
+				new Tuple2<>("Jack", "Engineering"),
+				new Tuple2<>("Tim", "Sales"),
+				new Tuple2<>("Zed", "HR")
 		);
 		CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
-				new Tuple2<String, Integer>("Allison", 100),
-				new Tuple2<String, Integer>("Jack", 200),
-				new Tuple2<String, Integer>("Zed", 150),
-				new Tuple2<String, Integer>("Zed", 250)
+				new Tuple2<>("Allison", 100),
+				new Tuple2<>("Jack", 200),
+				new Tuple2<>("Zed", 150),
+				new Tuple2<>("Zed", 250)
 		);
 
 		List<Tuple4<String, String, String, Object>> actual = computeOuterJoin(input1, input2, OuterJoinType.RIGHT);
@@ -198,11 +205,12 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		Assert.assertEquals(expected, actual);
 	}
 
+	@SuppressWarnings("unchecked")
 	protected void testRightSideEmpty() throws Exception {
 		CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
-				new Tuple2<String, String>("Jack", "Engineering"),
-				new Tuple2<String, String>("Tim", "Sales"),
-				new Tuple2<String, String>("Zed", "HR")
+				new Tuple2<>("Jack", "Engineering"),
+				new Tuple2<>("Tim", "Sales"),
+				new Tuple2<>("Zed", "HR")
 		);
 		CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of();
 
@@ -221,13 +229,14 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		Assert.assertEquals(Collections.<Tuple4<String,String,String,Object>>emptyList(), actualRight);
 	}
 
+	@SuppressWarnings("unchecked")
 	protected void testLeftSideEmpty() throws Exception {
 		CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of();
 		CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
-				new Tuple2<String, Integer>("Allison", 100),
-				new Tuple2<String, Integer>("Jack", 200),
-				new Tuple2<String, Integer>("Zed", 150),
-				new Tuple2<String, Integer>("Zed", 250)
+				new Tuple2<>("Allison", 100),
+				new Tuple2<>("Jack", 200),
+				new Tuple2<>("Zed", 150),
+				new Tuple2<>("Zed", 250)
 		);
 
 		List<Tuple4<String, String, String, Object>> actualLeft = computeOuterJoin(input1, input2, OuterJoinType.LEFT);
@@ -246,42 +255,49 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		Assert.assertEquals(expected, actualFull);
 	}
 
+	@SuppressWarnings("unchecked, rawtypes")
 	private List<Tuple4<String, String, String, Object>> computeOuterJoin(ResettableMutableObjectIterator<Tuple2<String, String>> input1,
 																		  ResettableMutableObjectIterator<Tuple2<String, Integer>> input2,
 																		  OuterJoinType outerJoinType) throws Exception {
 		input1.reset();
 		input2.reset();
-		AbstractMergeOuterJoinIterator<Tuple2<String, String>, Tuple2<String, Integer>, Tuple4<String, String, String, Object>> iterator =
-				createOuterJoinIterator(outerJoinType, input1, input2, serializer1, comparator1, serializer2, comparator2,
-						pairComp, this.memoryManager, this.ioManager, PAGES_FOR_BNLJN, this.parentTask);
-
-		List<Tuple4<String, String, String, Object>> actual = new ArrayList<Tuple4<String, String, String, Object>>();
-		ListCollector<Tuple4<String, String, String, Object>> collector = new ListCollector<Tuple4<String, String, String, Object>>(actual);
+		AbstractMergeOuterJoinIterator iterator =
+				createOuterJoinIterator(
+						outerJoinType, input1, input2, serializer1, comparator1, serializer2, comparator2,
+						pairComp, this.memoryManager, this.ioManager, PAGES_FOR_BNLJN, this.parentTask
+				);
+
+		List<Tuple4<String, String, String, Object>> actual = new ArrayList<>();
+		ListCollector<Tuple4<String, String, String, Object>> collector = new ListCollector<>(actual);
 		while (iterator.callWithNextKey(new SimpleTupleJoinFunction(), collector)) ;
 		iterator.close();
 
 		return actual;
 	}
 
+	@SuppressWarnings("unchecked, rawtypes")
 	protected void testOuterJoinWithHighNumberOfCommonKeys(OuterJoinType outerJoinType, int input1Size, int input1Duplicates, int input1ValueLength,
 														float input1KeyDensity, int input2Size, int input2Duplicates, int input2ValueLength, float input2KeyDensity) {
-		TypeSerializer<Tuple2<Integer, String>> serializer1 = new TupleSerializer<Tuple2<Integer, String>>(
+		TypeSerializer<Tuple2<Integer, String>> serializer1 = new TupleSerializer<>(
 				(Class<Tuple2<Integer, String>>) (Class<?>) Tuple2.class,
-				new TypeSerializer<?>[] { IntSerializer.INSTANCE, StringSerializer.INSTANCE });
-		TypeSerializer<Tuple2<Integer, String>> serializer2 = new TupleSerializer<Tuple2<Integer, String>>(
+				new TypeSerializer<?>[]{IntSerializer.INSTANCE, StringSerializer.INSTANCE}
+		);
+		TypeSerializer<Tuple2<Integer, String>> serializer2 = new TupleSerializer<>(
 				(Class<Tuple2<Integer, String>>) (Class<?>) Tuple2.class,
-				new TypeSerializer<?>[] { IntSerializer.INSTANCE, StringSerializer.INSTANCE });
-		TypeComparator<Tuple2<Integer, String>> comparator1 =  new TupleComparator<Tuple2<Integer, String>>(
+				new TypeSerializer<?>[]{IntSerializer.INSTANCE, StringSerializer.INSTANCE}
+		);
+		TypeComparator<Tuple2<Integer, String>> comparator1 = new TupleComparator<>(
 				new int[]{0},
-				new TypeComparator<?>[] { new IntComparator(true) },
-				new TypeSerializer<?>[] { IntSerializer.INSTANCE });
-		TypeComparator<Tuple2<Integer, String>> comparator2 =  new TupleComparator<Tuple2<Integer, String>>(
+				new TypeComparator<?>[]{new IntComparator(true)},
+				new TypeSerializer<?>[]{IntSerializer.INSTANCE}
+		);
+		TypeComparator<Tuple2<Integer, String>> comparator2 = new TupleComparator<>(
 				new int[]{0},
-				new TypeComparator<?>[] { new IntComparator(true) },
-				new TypeSerializer<?>[] { IntSerializer.INSTANCE });
+				new TypeComparator<?>[]{new IntComparator(true)},
+				new TypeSerializer<?>[]{IntSerializer.INSTANCE}
+		);
 
-		TypePairComparator<Tuple2<Integer, String>, Tuple2<Integer, String>> pairComparator =
-				new GenericPairComparator<Tuple2<Integer, String>, Tuple2<Integer, String>>(comparator1, comparator2);
+		TypePairComparator<Tuple2<Integer, String>, Tuple2<Integer, String>> pairComparator = new GenericPairComparator<>(comparator1, comparator2);
 
 		this.memoryManager = new DefaultMemoryManager(MEMORY_SIZE, 1);
 		this.ioManager = new IOManagerAsync();
@@ -298,16 +314,16 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 			final TupleConstantValueIterator const1Iter = new TupleConstantValueIterator(DUPLICATE_KEY, "LEFT String for Duplicate Keys", input1Duplicates);
 			final TupleConstantValueIterator const2Iter = new TupleConstantValueIterator(DUPLICATE_KEY, "RIGHT String for Duplicate Keys", input2Duplicates);
 
-			final List<MutableObjectIterator<Tuple2<Integer, String>>> inList1 = new ArrayList<MutableObjectIterator<Tuple2<Integer, String>>>();
+			final List<MutableObjectIterator<Tuple2<Integer, String>>> inList1 = new ArrayList<>();
 			inList1.add(gen1Iter);
 			inList1.add(const1Iter);
 
-			final List<MutableObjectIterator<Tuple2<Integer, String>>> inList2 = new ArrayList<MutableObjectIterator<Tuple2<Integer, String>>>();
+			final List<MutableObjectIterator<Tuple2<Integer, String>>> inList2 = new ArrayList<>();
 			inList2.add(gen2Iter);
 			inList2.add(const2Iter);
 
-			MutableObjectIterator<Tuple2<Integer, String>> input1 = new MergeIterator<Tuple2<Integer, String>>(inList1, comparator1.duplicate());
-			MutableObjectIterator<Tuple2<Integer, String>> input2 = new MergeIterator<Tuple2<Integer, String>>(inList2, comparator2.duplicate());
+			MutableObjectIterator<Tuple2<Integer, String>> input1 = new MergeIterator<>(inList1, comparator1.duplicate());
+			MutableObjectIterator<Tuple2<Integer, String>> input2 = new MergeIterator<>(inList2, comparator2.duplicate());
 
 			// collect expected data
 			final Map<Integer, Collection<Match>> expectedMatchesMap = joinValues(
@@ -333,13 +349,13 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 			inList2.add(gen2Iter);
 			inList2.add(const2Iter);
 
-			input1 = new MergeIterator<Tuple2<Integer, String>>(inList1, comparator1.duplicate());
-			input2 = new MergeIterator<Tuple2<Integer, String>>(inList2, comparator2.duplicate());
+			input1 = new MergeIterator<>(inList1, comparator1.duplicate());
+			input2 = new MergeIterator<>(inList2, comparator2.duplicate());
 
 			final FlatJoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple2<Integer, String>> joinFunction =
 					new MatchRemovingJoiner(expectedMatchesMap);
 
-			final Collector<Tuple2<Integer, String>> collector = new DiscardingOutputCollector<Tuple2<Integer, String>>();
+			final Collector<Tuple2<Integer, String>> collector = new DiscardingOutputCollector<>();
 
 
 			// we create this sort-merge iterator with little memory for the block-nested-loops fall-back to make sure it
@@ -367,7 +383,7 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 		}
 	}
 
-	protected abstract <T1, T2> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType,
+	protected abstract <T1, T2, T3> AbstractMergeOuterJoinIterator<T1, T2, T3> createOuterJoinIterator(OuterJoinType outerJoinType,
 																			  MutableObjectIterator<T1> input1,
 																			  MutableObjectIterator<T2> input2,
 																			  TypeSerializer<T1> serializer1, TypeComparator<T1> comparator1,
@@ -387,7 +403,7 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 			Map<Integer, Collection<String>> leftMap,
 			Map<Integer, Collection<String>> rightMap,
 			OuterJoinType outerJoinType) {
-		Map<Integer, Collection<Match>> map = new HashMap<Integer, Collection<Match>>();
+		Map<Integer, Collection<Match>> map = new HashMap<>();
 
 		for (Integer key : leftMap.keySet()) {
 			Collection<String> leftValues = leftMap.get(key);
@@ -441,8 +457,8 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
 
 	private Map<Integer, Collection<String>> collectData(MutableObjectIterator<Tuple2<Integer, String>> iter)
 			throws Exception {
-		final Map<Integer, Collection<String>> map = new HashMap<Integer, Collection<String>>();
-		Tuple2<Integer, String> pair = new Tuple2<Integer, String>();
+		final Map<Integer, Collection<String>> map = new HashMap<>();
+		Tuple2<Integer, String> pair = new Tuple2<>();
 
 		while ((pair = iter.next(pair)) != null) {
 			final Integer key = pair.getField(0);

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java
index 1205bc1..91609bb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java
@@ -28,10 +28,11 @@ import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.Ou
 import org.apache.flink.util.MutableObjectIterator;
 import org.junit.Test;
 
-public class NonReusingSortMergeOuterJoinIteratorITCase  extends AbstractSortMergeOuterJoinIteratorITCase {
+public class NonReusingSortMergeOuterJoinIteratorITCase extends AbstractSortMergeOuterJoinIteratorITCase {
 
 	@Override
-	protected <T1, T2> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	protected <T1, T2, T3> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
 																			  MutableObjectIterator<T2> input2, TypeSerializer<T1> serializer1,
 																			  TypeComparator<T1> comparator1, TypeSerializer<T2> serializer2, TypeComparator<T2> comparator2,
 																			  TypePairComparator<T1, T2> pairComparator, MemoryManager memoryManager, IOManager ioManager,

http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/ReusingSortMergeOuterJoinIteratorITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/ReusingSortMergeOuterJoinIteratorITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/ReusingSortMergeOuterJoinIteratorITCase.java
index b4fbd80..779cf37 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/ReusingSortMergeOuterJoinIteratorITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/ReusingSortMergeOuterJoinIteratorITCase.java
@@ -31,7 +31,8 @@ import org.junit.Test;
 public class ReusingSortMergeOuterJoinIteratorITCase extends AbstractSortMergeOuterJoinIteratorITCase {
 
 	@Override
-	protected <T1, T2> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	protected <T1, T2, T3> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
 																			  MutableObjectIterator<T2> input2, TypeSerializer<T1> serializer1,
 																			  TypeComparator<T1> comparator1, TypeSerializer<T2> serializer2, TypeComparator<T2> comparator2,
 																			  TypePairComparator<T1, T2> pairComparator, MemoryManager memoryManager, IOManager ioManager,


Mime
View raw message