[FLINK-2106] [runtime] Rename DriverStrategy.MERGE to DriverStrategy.INNER_MERGE

This closes #1052

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:

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

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;
 

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();

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),
 

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,

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);
 		

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);
 			

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());

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());