Fix optimizer error in channel instantiation (wrong parameterization)

stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java

@@ -42,7 +42,7 @@ public class DefaultCostEstimator extends CostEstimator {
 	
 	private static final float HASHING_CPU_FACTOR = 4;
 	
-	private static final float SORTING_CPU_FACTOR = 7;
+	private static final float SORTING_CPU_FACTOR = 9;
 	
 	
 	// --------------------------------------------------------------------------------------------

stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TwoInputNode.java

@@ -474,12 +474,16 @@ public abstract class TwoInputNode extends OptimizerNode {
 							if (dps.areCoFulfilled(lpp.getProperties1(), lpp.getProperties2(), 
 								in1.getLocalProperties(), in2.getLocalProperties()))
 							{
+								// copy, because setting required properties and instantiation may
+								// change the channels and should not affect prior candidates
 								Channel in1Copy = in1.clone();
 								in1Copy.setRequiredLocalProps(lpp.getProperties1());
-								in2.setRequiredLocalProps(lpp.getProperties2());
+								
+								Channel in2Copy = in2.clone();
+								in2Copy.setRequiredLocalProps(lpp.getProperties2());
 								
 								// all right, co compatible
-								instantiate(dps, in1Copy, in2, broadcastPlanChannels, target, estimator, rgps1, rgps2, ilp1, ilp2);
+								instantiate(dps, in1Copy, in2Copy, broadcastPlanChannels, target, estimator, rgps1, rgps2, ilp1, ilp2);
 								break;
 							} else {
 								// meet, but not co-compatible

stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/examples/RelationalQueryCompilerTest.java

@@ -60,6 +60,7 @@ public class RelationalQueryCompilerTest extends CompilerTestBase {
 			
 			// compile
 			final OptimizedPlan plan = compileNoStats(p);
+			
 			final OptimizerPlanNodeResolver or = getOptimizerPlanNodeResolver(plan);
 			
 			// get the nodes from the final plan

stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/iterations/ConnectedComponentsTest.java

@@ -102,7 +102,9 @@ public class ConnectedComponentsTest extends CompilerTestBase {
 		Assert.assertEquals(DriverStrategy.NONE, vertexSource.getDriverStrategy());
 		Assert.assertEquals(DriverStrategy.NONE, edgesSource.getDriverStrategy());
 		
-		Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, neighborsJoin.getDriverStrategy());
+		Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, neighborsJoin.getDriverStrategy());
+		Assert.assertTrue(!neighborsJoin.getInput1().getTempMode().isCached());
+		Assert.assertTrue(!neighborsJoin.getInput2().getTempMode().isCached());
 		Assert.assertEquals(set0, neighborsJoin.getKeysForInput1());
 		Assert.assertEquals(set0, neighborsJoin.getKeysForInput2());
 		
@@ -120,7 +122,6 @@ public class ConnectedComponentsTest extends CompilerTestBase {
 		Assert.assertEquals(ShipStrategyType.FORWARD, neighborsJoin.getInput1().getShipStrategy()); // workset
 		Assert.assertEquals(ShipStrategyType.PARTITION_HASH, neighborsJoin.getInput2().getShipStrategy()); // edges
 		Assert.assertEquals(set0, neighborsJoin.getInput2().getShipStrategyKeys());
-		Assert.assertTrue(neighborsJoin.getInput2().getTempMode().isCached());
 		
 		Assert.assertEquals(ShipStrategyType.PARTITION_HASH, minIdReducer.getInput().getShipStrategy());
 		Assert.assertEquals(set0, minIdReducer.getInput().getShipStrategyKeys());
@@ -182,7 +183,9 @@ public class ConnectedComponentsTest extends CompilerTestBase {
 		Assert.assertEquals(DriverStrategy.NONE, vertexSource.getDriverStrategy());
 		Assert.assertEquals(DriverStrategy.NONE, edgesSource.getDriverStrategy());
 		
-		Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, neighborsJoin.getDriverStrategy());
+		Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, neighborsJoin.getDriverStrategy());
+		Assert.assertTrue(!neighborsJoin.getInput1().getTempMode().isCached());
+		Assert.assertTrue(!neighborsJoin.getInput2().getTempMode().isCached());
 		Assert.assertEquals(set0, neighborsJoin.getKeysForInput1());
 		Assert.assertEquals(set0, neighborsJoin.getKeysForInput2());
 		
@@ -200,7 +203,6 @@ public class ConnectedComponentsTest extends CompilerTestBase {
 		Assert.assertEquals(ShipStrategyType.FORWARD, neighborsJoin.getInput1().getShipStrategy()); // workset
 		Assert.assertEquals(ShipStrategyType.PARTITION_HASH, neighborsJoin.getInput2().getShipStrategy()); // edges
 		Assert.assertEquals(set0, neighborsJoin.getInput2().getShipStrategyKeys());
-		Assert.assertTrue(neighborsJoin.getInput2().getTempMode().isCached());
 		
 		Assert.assertEquals(ShipStrategyType.PARTITION_HASH, minIdReducer.getInput().getShipStrategy());
 		Assert.assertEquals(set0, minIdReducer.getInput().getShipStrategyKeys());