You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2015/09/08 16:58:50 UTC
[1/3] flink git commit: [FLINK-2106] [runtime] Rename
DriverStrategy.MERGE to DriverStrategy.INNER_MERGE
Repository: flink
Updated Branches:
refs/heads/master 24f7fa9ef -> 2da82f915
[FLINK-2106] [runtime] Rename DriverStrategy.MERGE to DriverStrategy.INNER_MERGE
This closes #1052
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2da82f91
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2da82f91
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2da82f91
Branch: refs/heads/master
Commit: 2da82f9154f9563e3fe8454e153d372b4bf18996
Parents: f3dee23
Author: r-pogalz <r....@campus.tu-berlin.de>
Authored: Tue Aug 11 21:40:21 2015 +0200
Committer: Fabian Hueske <fh...@apache.org>
Committed: Tue Sep 8 16:57:51 2015 +0200
----------------------------------------------------------------------
.../flink/optimizer/costs/CostEstimator.java | 2 +-
.../operators/SortMergeJoinDescriptor.java | 4 ++--
.../plandump/PlanJSONDumpGenerator.java | 2 +-
.../optimizer/java/JoinTranslationTest.java | 2 +-
.../flink/runtime/operators/DriverStrategy.java | 2 +-
.../flink/runtime/operators/JoinDriver.java | 4 ++--
.../operators/MatchTaskExternalITCase.java | 2 +-
.../flink/runtime/operators/MatchTaskTest.java | 24 ++++++++++----------
.../examples/RelationalQueryCompilerTest.java | 4 ++--
.../ConnectedComponentsCoGroupTest.java | 2 +-
10 files changed, 24 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-optimizer/src/main/java/org/apache/flink/optimizer/costs/CostEstimator.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/costs/CostEstimator.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/costs/CostEstimator.java
index 3a02735..6a3ff09 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/costs/CostEstimator.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/costs/CostEstimator.java
@@ -195,7 +195,7 @@ public abstract class CostEstimator {
// pipelined local union is for free
break;
- case MERGE:
+ case INNER_MERGE:
addLocalMergeCost(firstInput, secondInput, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_FIRST:
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/SortMergeJoinDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/SortMergeJoinDescriptor.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/SortMergeJoinDescriptor.java
index 356836a..3ab0aa7 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/SortMergeJoinDescriptor.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/SortMergeJoinDescriptor.java
@@ -49,7 +49,7 @@ public class SortMergeJoinDescriptor extends AbstractJoinDescriptor {
@Override
public DriverStrategy getStrategy() {
- return DriverStrategy.MERGE;
+ return DriverStrategy.INNER_MERGE;
}
@Override
@@ -99,7 +99,7 @@ public class SortMergeJoinDescriptor extends AbstractJoinDescriptor {
inputOrders = tmp;
}
- return new DualInputPlanNode(node, "Join("+node.getOperator().getName()+")", in1, in2, DriverStrategy.MERGE, this.keys1, this.keys2, inputOrders);
+ return new DualInputPlanNode(node, "Join(" + node.getOperator().getName() + ")", in1, in2, DriverStrategy.INNER_MERGE, this.keys1, this.keys2, inputOrders);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-optimizer/src/main/java/org/apache/flink/optimizer/plandump/PlanJSONDumpGenerator.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plandump/PlanJSONDumpGenerator.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plandump/PlanJSONDumpGenerator.java
index b04cdd8..dc99fd7 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plandump/PlanJSONDumpGenerator.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plandump/PlanJSONDumpGenerator.java
@@ -480,7 +480,7 @@ public class PlanJSONDumpGenerator {
locString = "Nested Loops (Streamed Outer: " + child2name + ")";
break;
- case MERGE:
+ case INNER_MERGE:
locString = "Merge";
break;
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-optimizer/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
index b3718b0..de1508b 100644
--- a/flink-optimizer/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
@@ -103,7 +103,7 @@ public class JoinTranslationTest extends CompilerTestBase {
DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.REPARTITION_SORT_MERGE);
assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput1().getShipStrategy());
assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput2().getShipStrategy());
- assertEquals(DriverStrategy.MERGE, node.getDriverStrategy());
+ assertEquals(DriverStrategy.INNER_MERGE, node.getDriverStrategy());
}
catch (Exception e) {
e.printStackTrace();
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/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 f42a275..74f737e 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
@@ -72,7 +72,7 @@ public enum DriverStrategy {
ALL_GROUP_COMBINE(AllGroupCombineDriver.class, null, PIPELINED, 0),
// 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),
+ INNER_MERGE(JoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
LEFT_OUTER_MERGE(LeftOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinDriver.java
index 5df715f..95e98ce 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinDriver.java
@@ -125,7 +125,7 @@ public class JoinDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT1
// create and return joining iterator according to provided local strategy.
if (objectReuseEnabled) {
switch (ls) {
- case MERGE:
+ case INNER_MERGE:
this.joinIterator = new ReusingMergeInnerJoinIterator<>(in1, in2,
serializer1, comparator1,
serializer2, comparator2,
@@ -157,7 +157,7 @@ public class JoinDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT1
}
} else {
switch (ls) {
- case MERGE:
+ case INNER_MERGE:
this.joinIterator = new NonReusingMergeInnerJoinIterator<>(in1, in2,
serializer1, comparator1,
serializer2, comparator2,
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskExternalITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskExternalITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskExternalITCase.java
index 30c1610..6f7fb21 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskExternalITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskExternalITCase.java
@@ -76,7 +76,7 @@ public class MatchTaskExternalITCase extends DriverTestBase<FlatJoinFunction<Rec
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
index 15f3d0c..6c4659d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
@@ -89,7 +89,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -123,7 +123,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -159,7 +159,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -195,7 +195,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -231,7 +231,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -267,7 +267,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -303,7 +303,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -338,7 +338,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -374,7 +374,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -404,7 +404,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -464,7 +464,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
@@ -524,7 +524,7 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Recor
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
- getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+ getTaskConfig().setDriverStrategy(DriverStrategy.INNER_MERGE);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-tests/src/test/java/org/apache/flink/test/optimizer/examples/RelationalQueryCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/optimizer/examples/RelationalQueryCompilerTest.java b/flink-tests/src/test/java/org/apache/flink/test/optimizer/examples/RelationalQueryCompilerTest.java
index f58486b..f4efb8a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/optimizer/examples/RelationalQueryCompilerTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/optimizer/examples/RelationalQueryCompilerTest.java
@@ -303,7 +303,7 @@ public class RelationalQueryCompilerTest extends CompilerTestBase {
}
private boolean checkBroadcastMergeJoin(DualInputPlanNode join, SingleInputPlanNode reducer) {
- if (DriverStrategy.MERGE == join.getDriverStrategy()) {
+ if (DriverStrategy.INNER_MERGE == join.getDriverStrategy()) {
// driver keys
Assert.assertEquals(set0, join.getKeysForInput1());
Assert.assertEquals(set0, join.getKeysForInput2());
@@ -327,7 +327,7 @@ public class RelationalQueryCompilerTest extends CompilerTestBase {
}
private boolean checkRepartitionMergeJoin(DualInputPlanNode join, SingleInputPlanNode reducer) {
- if (DriverStrategy.MERGE == join.getDriverStrategy()) {
+ if (DriverStrategy.INNER_MERGE == join.getDriverStrategy()) {
// driver keys
Assert.assertEquals(set0, join.getKeysForInput1());
Assert.assertEquals(set0, join.getKeysForInput2());
http://git-wip-us.apache.org/repos/asf/flink/blob/2da82f91/flink-tests/src/test/java/org/apache/flink/test/optimizer/iterations/ConnectedComponentsCoGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/optimizer/iterations/ConnectedComponentsCoGroupTest.java b/flink-tests/src/test/java/org/apache/flink/test/optimizer/iterations/ConnectedComponentsCoGroupTest.java
index de5fde0..99402a5 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/optimizer/iterations/ConnectedComponentsCoGroupTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/optimizer/iterations/ConnectedComponentsCoGroupTest.java
@@ -93,7 +93,7 @@ public class ConnectedComponentsCoGroupTest extends CompilerTestBase {
Assert.assertEquals(DriverStrategy.NONE, vertexSource.getDriverStrategy());
Assert.assertEquals(DriverStrategy.NONE, edgesSource.getDriverStrategy());
- Assert.assertEquals(DriverStrategy.MERGE, neighborsJoin.getDriverStrategy());
+ Assert.assertEquals(DriverStrategy.INNER_MERGE, neighborsJoin.getDriverStrategy());
Assert.assertEquals(set0, neighborsJoin.getKeysForInput1());
Assert.assertEquals(set0, neighborsJoin.getKeysForInput2());
[3/3] flink git commit: [FLINK-2106] [runtime] Add Outer Join drivers
and Outer Merge strategies to Runtime
Posted by fh...@apache.org.
[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....@campus.tu-berlin.de>
Authored: Thu Aug 6 23:13:16 2015 +0200
Committer: Fabian Hueske <fh...@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,
[2/3] flink git commit: [FLINK-2106] [runtime] Add Outer Join drivers
and Outer Merge strategies to Runtime
Posted by fh...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/f3dee23b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
new file mode 100644
index 0000000..a085eeb
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
@@ -0,0 +1,433 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
+import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+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.PactDriver;
+import org.apache.flink.runtime.operators.PactTaskContext;
+import org.apache.flink.runtime.operators.ResettablePactDriver;
+import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
+import org.apache.flink.runtime.operators.util.TaskConfig;
+import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.TestLogger;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+@RunWith(Parameterized.class)
+public class BinaryOperatorTestBase<S extends Function, IN, OUT> extends TestLogger implements PactTaskContext<S, OUT> {
+
+ protected static final int PAGE_SIZE = 32 * 1024;
+
+ private final IOManager ioManager;
+
+ private final MemoryManager memManager;
+
+ private final List<MutableObjectIterator<IN>> inputs;
+
+ private final List<TypeComparator<IN>> comparators;
+
+ private final List<UnilateralSortMerger<IN>> sorters;
+
+ private final AbstractInvokable owner;
+
+ private final TaskConfig taskConfig;
+
+ private final TaskManagerRuntimeInfo taskManageInfo;
+
+ protected final long perSortMem;
+
+ protected final double perSortFractionMem;
+
+ private Collector<OUT> output;
+
+ protected int numFileHandles;
+
+ private S stub;
+
+ private PactDriver<S, IN> driver;
+
+ private volatile boolean running = true;
+
+ private ExecutionConfig executionConfig;
+
+ private List<TypeSerializer<IN>> inputSerializers = new ArrayList<>();
+
+ protected BinaryOperatorTestBase(ExecutionConfig executionConfig, long memory, int maxNumSorters, long perSortMemory) {
+ if (memory < 0 || maxNumSorters < 0 || perSortMemory < 0) {
+ throw new IllegalArgumentException();
+ }
+
+ final long totalMem = Math.max(memory, 0) + (Math.max(maxNumSorters, 0) * perSortMemory);
+
+ this.perSortMem = perSortMemory;
+ this.perSortFractionMem = (double) perSortMemory / totalMem;
+ this.ioManager = new IOManagerAsync();
+ this.memManager = totalMem > 0 ? new DefaultMemoryManager(totalMem, 1) : null;
+
+ this.inputs = new ArrayList<>();
+ this.comparators = new ArrayList<>();
+ this.sorters = new ArrayList<>();
+
+ this.owner = new DummyInvokable();
+ this.taskConfig = new TaskConfig(new Configuration());
+ this.executionConfig = executionConfig;
+ this.taskManageInfo = new TaskManagerRuntimeInfo("localhost", new Configuration());
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> getConfigurations() throws IOException {
+ LinkedList<Object[]> configs = new LinkedList<>();
+
+ ExecutionConfig withReuse = new ExecutionConfig();
+ withReuse.enableObjectReuse();
+
+ ExecutionConfig withoutReuse = new ExecutionConfig();
+ withoutReuse.disableObjectReuse();
+
+ Object[] a = {withoutReuse};
+ configs.add(a);
+ Object[] b = {withReuse};
+ configs.add(b);
+
+ return configs;
+ }
+
+ public void addInput(MutableObjectIterator<IN> input, TypeSerializer<IN> serializer) {
+ this.inputs.add(input);
+ this.sorters.add(null);
+ this.inputSerializers.add(serializer);
+ }
+
+ @SuppressWarnings("unchecked")
+ public void addInputSorted(MutableObjectIterator<IN> input, TypeSerializer<IN> serializer, TypeComparator<IN> comp) throws Exception {
+ this.inputSerializers.add(serializer);
+ UnilateralSortMerger<IN> sorter = new UnilateralSortMerger<>(
+ this.memManager,
+ this.ioManager,
+ input,
+ this.owner,
+ new RuntimeSerializerFactory<>(serializer, (Class<IN>) serializer.createInstance().getClass()),
+ comp,
+ this.perSortFractionMem,
+ 32,
+ 0.8f
+ );
+ this.sorters.add(sorter);
+ this.inputs.add(null);
+ }
+
+ public void addDriverComparator(TypeComparator<IN> comparator) {
+ this.comparators.add(comparator);
+ }
+
+ public void setOutput(Collector<OUT> output) {
+ this.output = output;
+ }
+
+ public void setOutput(List<OUT> output, TypeSerializer<OUT> outSerializer) {
+ this.output = new ListOutputCollector<>(output, outSerializer);
+ }
+
+ public int getNumFileHandlesForSort() {
+ return numFileHandles;
+ }
+
+
+ public void setNumFileHandlesForSort(int numFileHandles) {
+ this.numFileHandles = numFileHandles;
+ }
+
+ @SuppressWarnings("rawtypes")
+ public void testDriver(PactDriver driver, Class stubClass) throws Exception {
+ testDriverInternal(driver, stubClass);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void testDriverInternal(PactDriver driver, Class stubClass) throws Exception {
+
+ this.driver = driver;
+ driver.setup(this);
+
+ this.stub = (S) stubClass.newInstance();
+
+ // regular running logic
+ this.running = true;
+ boolean stubOpen = false;
+
+ try {
+ // run the data preparation
+ try {
+ driver.prepare();
+ } catch (Throwable t) {
+ throw new Exception("The data preparation caused an error: " + t.getMessage(), t);
+ }
+
+ // open stub implementation
+ try {
+ FunctionUtils.openFunction(this.stub, getTaskConfig().getStubParameters());
+ stubOpen = true;
+ } catch (Throwable t) {
+ throw new Exception("The user defined 'open()' method caused an exception: " + t.getMessage(), t);
+ }
+
+ if (!running) {
+ return;
+ }
+
+ // run the user code
+ driver.run();
+
+ // close. We close here such that a regular close throwing an exception marks a task as failed.
+ if (this.running) {
+ FunctionUtils.closeFunction(this.stub);
+ stubOpen = false;
+ }
+
+ this.output.close();
+ } catch (Exception ex) {
+ // close the input, but do not report any exceptions, since we already have another root cause
+ if (stubOpen) {
+ try {
+ FunctionUtils.closeFunction(this.stub);
+ } catch (Throwable ignored) {
+ }
+ }
+
+ // if resettable driver invoke tear down
+ if (this.driver instanceof ResettablePactDriver) {
+ final ResettablePactDriver<?, ?> resDriver = (ResettablePactDriver<?, ?>) this.driver;
+ try {
+ resDriver.teardown();
+ } catch (Throwable t) {
+ throw new Exception("Error while shutting down an iterative operator: " + t.getMessage(), t);
+ }
+ }
+
+ // drop exception, if the task was canceled
+ if (this.running) {
+ throw ex;
+ }
+
+ } finally {
+ driver.cleanup();
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void testResettableDriver(ResettablePactDriver driver, Class stubClass, int iterations) throws Exception {
+ driver.setup(this);
+
+ for (int i = 0; i < iterations; i++) {
+
+ if (i == 0) {
+ driver.initialize();
+ } else {
+ driver.reset();
+ }
+
+ testDriver(driver, stubClass);
+
+ }
+
+ driver.teardown();
+ }
+
+ public void cancel() throws Exception {
+ this.running = false;
+
+ // compensate for races, where cancel is called before the driver is set
+ // not that this is an artifact of a bad design of this test base, where the setup
+ // of the basic properties is not separated from the invocation of the execution logic
+ while (this.driver == null) {
+ Thread.sleep(200);
+ }
+ this.driver.cancel();
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public TaskConfig getTaskConfig() {
+ return this.taskConfig;
+ }
+
+ @Override
+ public TaskManagerRuntimeInfo getTaskManagerInfo() {
+ return this.taskManageInfo;
+ }
+
+ @Override
+ public ExecutionConfig getExecutionConfig() {
+ return executionConfig;
+ }
+
+ @Override
+ public ClassLoader getUserCodeClassLoader() {
+ return getClass().getClassLoader();
+ }
+
+ @Override
+ public IOManager getIOManager() {
+ return this.ioManager;
+ }
+
+ @Override
+ public MemoryManager getMemoryManager() {
+ return this.memManager;
+ }
+
+ @Override
+ public <X> MutableObjectIterator<X> getInput(int index) {
+ MutableObjectIterator<IN> in = this.inputs.get(index);
+ if (in == null) {
+ // waiting from sorter
+ try {
+ in = this.sorters.get(index).getIterator();
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Interrupted");
+ }
+ this.inputs.set(index, in);
+ }
+
+ @SuppressWarnings("unchecked")
+ MutableObjectIterator<X> input = (MutableObjectIterator<X>) this.inputs.get(index);
+ return input;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <X> TypeSerializerFactory<X> getInputSerializer(int index) {
+ TypeSerializer<X> ser = (TypeSerializer<X>) this.inputSerializers.get(index);
+ return new RuntimeSerializerFactory<>(ser, (Class<X>) ser.createInstance().getClass());
+ }
+
+ @Override
+ public <X> TypeComparator<X> getDriverComparator(int index) {
+ @SuppressWarnings("unchecked")
+ TypeComparator<X> comparator = (TypeComparator<X>) this.comparators.get(index);
+ return comparator;
+ }
+
+ @Override
+ public S getStub() {
+ return this.stub;
+ }
+
+ @Override
+ public Collector<OUT> getOutputCollector() {
+ return this.output;
+ }
+
+ @Override
+ public AbstractInvokable getOwningNepheleTask() {
+ return this.owner;
+ }
+
+ @Override
+ public String formatLogString(String message) {
+ return "Driver Tester: " + message;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @After
+ public void shutdownAll() throws Exception {
+ // 1st, shutdown sorters
+ for (UnilateralSortMerger<?> sorter : this.sorters) {
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ this.sorters.clear();
+
+ // 2nd, shutdown I/O
+ this.ioManager.shutdown();
+ Assert.assertTrue("I/O Manager has not properly shut down.", this.ioManager.isProperlyShutDown());
+
+ // last, verify all memory is returned and shutdown mem manager
+ MemoryManager memMan = getMemoryManager();
+ if (memMan != null) {
+ Assert.assertTrue("Memory Manager managed memory was not completely freed.", memMan.verifyEmpty());
+ memMan.shutdown();
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static final class ListOutputCollector<OUT> implements Collector<OUT> {
+
+ private final List<OUT> output;
+ private final TypeSerializer<OUT> serializer;
+
+ public ListOutputCollector(List<OUT> outputList, TypeSerializer<OUT> serializer) {
+ this.output = outputList;
+ this.serializer = serializer;
+ }
+
+
+ @Override
+ public void collect(OUT record) {
+ this.output.add(serializer.copy(record));
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ public static final class CountingOutputCollector<OUT> implements Collector<OUT> {
+
+ private int num;
+
+ @Override
+ public void collect(OUT record) {
+ this.num++;
+ }
+
+ @Override
+ public void close() {
+ }
+
+ public int getNumberOfRecords() {
+ return this.num;
+ }
+ }
+}