flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [47/53] [abbrv] flink git commit: [FLINK-441] [optimizer] Rename Pact* and Nephele* classes
Date Fri, 20 Mar 2015 10:07:26 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/BranchingPlansCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/BranchingPlansCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/BranchingPlansCompilerTest.java
index 0dd7e93..ff0e004 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/BranchingPlansCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/BranchingPlansCompilerTest.java
@@ -47,7 +47,7 @@ import org.apache.flink.api.java.record.operators.ReduceOperator;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.SinkPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.testfunctions.DummyFlatJoinFunction;
 import org.apache.flink.optimizer.testfunctions.IdentityGroupReducer;
 import org.apache.flink.optimizer.testfunctions.IdentityKeyExtractor;
@@ -101,7 +101,7 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 	
 			// ---------- compile plan to nephele job graph to verify that no error is thrown ----------
 	
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			jobGen.compileJobGraph(oPlan);
 		} catch (Exception e) {
 			e.printStackTrace();
@@ -162,13 +162,13 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 			allSinks.add(out3Path);
 			
 			for (SinkPlanNode n : oPlan.getDataSinks()) {
-				String path = ((FileDataSink) n.getSinkNode().getPactContract()).getFilePath();
+				String path = ((FileDataSink) n.getSinkNode().getOperator()).getFilePath();
 				Assert.assertTrue("Invalid data sink.", allSinks.remove(path));
 			}
 			
 			// ---------- compile plan to nephele job graph to verify that no error is thrown ----------
 			
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			jobGen.compileJobGraph(oPlan);
 		} catch (Exception e) {
 			e.printStackTrace();
@@ -262,7 +262,7 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 			
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			
 			//Compile plan to verify that no error is thrown
 			jobGen.compileJobGraph(oPlan);
@@ -350,13 +350,13 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 			allSinks.add(out3Path);
 			
 			for (SinkPlanNode n : oPlan.getDataSinks()) {
-				String path = ((FileDataSink) n.getSinkNode().getPactContract()).getFilePath();
+				String path = ((FileDataSink) n.getSinkNode().getOperator()).getFilePath();
 				Assert.assertTrue("Invalid data sink.", allSinks.remove(path));
 			}
 			
 			// ---------- compile plan to nephele job graph to verify that no error is thrown ----------
 			
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			jobGen.compileJobGraph(oPlan);
 		} catch (Exception e) {
 			e.printStackTrace();
@@ -449,7 +449,7 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 			
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			
 			//Compile plan to verify that no error is thrown
 			jobGen.compileJobGraph(oPlan);
@@ -495,7 +495,7 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 				.input2(ma2)
 				.name("Match 2")
 				.build();
-			mat2.setParameter(PactCompiler.HINT_LOCAL_STRATEGY, PactCompiler.HINT_LOCAL_STRATEGY_MERGE);
+			mat2.setParameter(Optimizer.HINT_LOCAL_STRATEGY, Optimizer.HINT_LOCAL_STRATEGY_MERGE);
 			
 			FileDataSink sink = new FileDataSink(new DummyOutputFormat(), OUT_FILE, mat2);
 			
@@ -505,7 +505,7 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 			
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			
 			//Compile plan to verify that no error is thrown
 			jobGen.compileJobGraph(oPlan);
@@ -555,13 +555,13 @@ public class BranchingPlansCompilerTest extends CompilerTestBase {
 			allSinks.add(out2Path);
 			
 			for (SinkPlanNode n : oPlan.getDataSinks()) {
-				String path = ((FileDataSink) n.getSinkNode().getPactContract()).getFilePath();
+				String path = ((FileDataSink) n.getSinkNode().getOperator()).getFilePath();
 				Assert.assertTrue("Invalid data sink.", allSinks.remove(path));
 			}
 			
 			// ---------- compile plan to nephele job graph to verify that no error is thrown ----------
 			
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			jobGen.compileJobGraph(oPlan);
 		} catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/CachedMatchStrategyCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/CachedMatchStrategyCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/CachedMatchStrategyCompilerTest.java
index c17ebe8..3e7da6c 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/CachedMatchStrategyCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/CachedMatchStrategyCompilerTest.java
@@ -33,7 +33,7 @@ import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.optimizer.dag.TempMode;
 import org.apache.flink.optimizer.plan.DualInputPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.operators.DriverStrategy;
 
@@ -51,7 +51,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 	public void testRightSide() {
 		try {
 			
-			Plan plan = getTestPlanRightStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_SECOND);
+			Plan plan = getTestPlanRightStatic(Optimizer.HINT_LOCAL_STRATEGY_HASH_BUILD_SECOND);
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 	
@@ -63,7 +63,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 			assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
 			assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
 		
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -79,7 +79,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 	public void testRightSideCountercheck() {
 		try {
 			
-			Plan plan = getTestPlanRightStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_FIRST);
+			Plan plan = getTestPlanRightStatic(Optimizer.HINT_LOCAL_STRATEGY_HASH_BUILD_FIRST);
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 	
@@ -91,7 +91,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 			assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
 			assertEquals(TempMode.CACHED, innerJoin.getInput2().getTempMode());
 		
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -108,7 +108,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 	public void testLeftSide() {
 		try {
 			
-			Plan plan = getTestPlanLeftStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_FIRST);
+			Plan plan = getTestPlanLeftStatic(Optimizer.HINT_LOCAL_STRATEGY_HASH_BUILD_FIRST);
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 	
@@ -120,7 +120,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 			assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
 			assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
 		
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -136,7 +136,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 	public void testLeftSideCountercheck() {
 		try {
 			
-			Plan plan = getTestPlanLeftStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_SECOND);
+			Plan plan = getTestPlanLeftStatic(Optimizer.HINT_LOCAL_STRATEGY_HASH_BUILD_SECOND);
 			
 			OptimizedPlan oPlan = compileNoStats(plan);
 	
@@ -148,7 +148,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 			assertEquals(TempMode.CACHED, innerJoin.getInput1().getTempMode());
 			assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
 		
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -191,7 +191,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 			assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
 			assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
 		
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -212,10 +212,10 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 		IterativeDataSet<Tuple3<Long, Long, Long>> iteration = bigInput.iterate(10);
 		
 		Configuration joinStrategy = new Configuration();
-		joinStrategy.setString(PactCompiler.HINT_SHIP_STRATEGY, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION_HASH);
+		joinStrategy.setString(Optimizer.HINT_SHIP_STRATEGY, Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH);
 		
 		if(strategy != "") {
-			joinStrategy.setString(PactCompiler.HINT_LOCAL_STRATEGY, strategy);
+			joinStrategy.setString(Optimizer.HINT_LOCAL_STRATEGY, strategy);
 		}
 		
 		DataSet<Tuple3<Long, Long, Long>> inner = iteration.join(smallInput).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner").withParameters(joinStrategy);
@@ -243,7 +243,7 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
 		IterativeDataSet<Tuple3<Long, Long, Long>> iteration = bigInput.iterate(10);
 		
 		Configuration joinStrategy = new Configuration();
-		joinStrategy.setString(PactCompiler.HINT_LOCAL_STRATEGY, strategy);
+		joinStrategy.setString(Optimizer.HINT_LOCAL_STRATEGY, strategy);
 		
 		DataSet<Tuple3<Long, Long, Long>> inner = smallInput.join(iteration).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner").withParameters(joinStrategy);
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/CompilerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/CompilerTestBase.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/CompilerTestBase.java
index 5265e3a..565d992 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/CompilerTestBase.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/CompilerTestBase.java
@@ -59,9 +59,9 @@ public abstract class CompilerTestBase implements java.io.Serializable {
 	
 	protected transient DataStatistics dataStats;
 	
-	protected transient PactCompiler withStatsCompiler;
+	protected transient Optimizer withStatsCompiler;
 	
-	protected transient PactCompiler noStatsCompiler;
+	protected transient Optimizer noStatsCompiler;
 	
 	private transient int statCounter;
 	
@@ -70,10 +70,10 @@ public abstract class CompilerTestBase implements java.io.Serializable {
 	@Before
 	public void setup() {
 		this.dataStats = new DataStatistics();
-		this.withStatsCompiler = new PactCompiler(this.dataStats, new DefaultCostEstimator());
+		this.withStatsCompiler = new Optimizer(this.dataStats, new DefaultCostEstimator());
 		this.withStatsCompiler.setDefaultDegreeOfParallelism(DEFAULT_PARALLELISM);
 		
-		this.noStatsCompiler = new PactCompiler(null, new DefaultCostEstimator());
+		this.noStatsCompiler = new Optimizer(null, new DefaultCostEstimator());
 		this.noStatsCompiler.setDefaultDegreeOfParallelism(DEFAULT_PARALLELISM);
 	}
 	
@@ -111,7 +111,7 @@ public abstract class CompilerTestBase implements java.io.Serializable {
 			HashMap<String, ArrayList<PlanNode>> map = new HashMap<String, ArrayList<PlanNode>>();
 			
 			for (PlanNode n : p.getAllNodes()) {
-				Operator<?> c = n.getOriginalOptimizerNode().getPactContract();
+				Operator<?> c = n.getOriginalOptimizerNode().getOperator();
 				String name = c.getName();
 				
 				ArrayList<PlanNode> list = map.get(name);
@@ -124,7 +124,7 @@ public abstract class CompilerTestBase implements java.io.Serializable {
 				boolean shouldAdd = true;
 				for (Iterator<PlanNode> iter = list.iterator(); iter.hasNext();) {
 					PlanNode in = iter.next();
-					if (in.getOriginalOptimizerNode().getPactContract() == c) {
+					if (in.getOriginalOptimizerNode().getOperator() == c) {
 						// is this the child or is our node the child
 						if (in instanceof SingleInputPlanNode && n instanceof SingleInputPlanNode) {
 							SingleInputPlanNode thisNode = (SingleInputPlanNode) n;

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/DOPChangeTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/DOPChangeTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/DOPChangeTest.java
index 7fa331a..b17e777 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/DOPChangeTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/DOPChangeTest.java
@@ -30,7 +30,7 @@ import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.PlanNode;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
 import org.apache.flink.optimizer.plan.SinkPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.util.DummyInputFormat;
 import org.apache.flink.optimizer.util.DummyMatchStub;
 import org.apache.flink.optimizer.util.DummyOutputFormat;
@@ -317,7 +317,7 @@ public class DOPChangeTest extends CompilerTestBase {
 		
 		OptimizedPlan oPlan = compileNoStats(plan);
 		
-		NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+		JobGraphGenerator jobGen = new JobGraphGenerator();
 		
 		//Compile plan to verify that no error is thrown
 		jobGen.compileJobGraph(oPlan);

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/DisjointDataFlowsTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/DisjointDataFlowsTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/DisjointDataFlowsTest.java
index b8809d7..aaee975 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/DisjointDataFlowsTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/DisjointDataFlowsTest.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.*;
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.junit.Test;
 
 @SuppressWarnings("serial")
@@ -41,7 +41,7 @@ public class DisjointDataFlowsTest extends CompilerTestBase {
 			Plan p = env.createProgramPlan();
 			OptimizedPlan op = compileNoStats(p);
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/DistinctCompilationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/DistinctCompilationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/DistinctCompilationTest.java
index d97f855..34aa9f8 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/DistinctCompilationTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/DistinctCompilationTest.java
@@ -78,10 +78,10 @@ public class DistinctCompilationTest extends CompilerTestBase implements java.io
 			assertEquals(new FieldList(0, 1), reduceNode.getInput().getLocalStrategyKeys());
 
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -136,13 +136,13 @@ public class DistinctCompilationTest extends CompilerTestBase implements java.io
 			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
 
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, keyExtractor.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, keyExtractor.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
 
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, keyProjector.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, keyProjector.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -192,10 +192,10 @@ public class DistinctCompilationTest extends CompilerTestBase implements java.io
 			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
 
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/HardPlansCompilationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/HardPlansCompilationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/HardPlansCompilationTest.java
index 146c085..6dadc19 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/HardPlansCompilationTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/HardPlansCompilationTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.api.java.record.operators.FileDataSource;
 import org.apache.flink.api.java.record.operators.MapOperator;
 import org.apache.flink.api.java.record.operators.ReduceOperator;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.util.DummyCrossStub;
 import org.apache.flink.optimizer.util.DummyInputFormat;
 import org.apache.flink.optimizer.util.DummyOutputFormat;
@@ -74,7 +74,7 @@ public class HardPlansCompilationTest extends CompilerTestBase {
 		plan.setDefaultParallelism(DEFAULT_PARALLELISM);
 		
 		OptimizedPlan oPlan = compileNoStats(plan);
-		NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+		JobGraphGenerator jobGen = new JobGraphGenerator();
 		jobGen.compileJobGraph(oPlan);
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/IterationsCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/IterationsCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/IterationsCompilerTest.java
index dc6fcad..ac4f820 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/IterationsCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/IterationsCompilerTest.java
@@ -42,7 +42,7 @@ import org.apache.flink.optimizer.plan.Channel;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.WorksetIterationPlanNode;
 import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.testfunctions.IdentityKeyExtractor;
 import org.apache.flink.optimizer.testfunctions.IdentityMapper;
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
@@ -79,7 +79,7 @@ public class IterationsCompilerTest extends CompilerTestBase {
 			new PlanJSONDumpGenerator().getOptimizerPlanAsJSON(p);
 			
 			// check that the JobGraphGenerator accepts the plan
-			new NepheleJobGraphGenerator().compileJobGraph(p);
+			new JobGraphGenerator().compileJobGraph(p);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -116,7 +116,7 @@ public class IterationsCompilerTest extends CompilerTestBase {
 			assertEquals(ShipStrategyType.PARTITION_HASH, wipn.getInput1().getShipStrategy());
 			assertTrue(wipn.getInput2().getTempMode().breaksPipeline());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -152,7 +152,7 @@ public class IterationsCompilerTest extends CompilerTestBase {
 			assertEquals(ShipStrategyType.PARTITION_HASH, wipn.getInput1().getShipStrategy());
 			assertTrue(wipn.getInput2().getTempMode().breaksPipeline());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -188,7 +188,7 @@ public class IterationsCompilerTest extends CompilerTestBase {
 			assertEquals(ShipStrategyType.FORWARD, wipn.getInput1().getShipStrategy());
 			assertTrue(wipn.getInput2().getTempMode().breaksPipeline());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -222,7 +222,7 @@ public class IterationsCompilerTest extends CompilerTestBase {
 				assertEquals(ShipStrategyType.PARTITION_HASH, c.getShipStrategy());
 			}
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -299,7 +299,7 @@ public class IterationsCompilerTest extends CompilerTestBase {
 			Plan p = env.createProgramPlan();
 			OptimizedPlan op = compileNoStats(p);
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/NestedIterationsTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/NestedIterationsTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/NestedIterationsTest.java
index d6852f5..e65758f 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/NestedIterationsTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/NestedIterationsTest.java
@@ -27,7 +27,7 @@ import org.apache.flink.api.java.operators.DeltaIteration;
 import org.apache.flink.api.java.operators.IterativeDataSet;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.testfunctions.DummyFlatJoinFunction;
 import org.apache.flink.optimizer.testfunctions.IdentityKeyExtractor;
 import org.apache.flink.optimizer.testfunctions.IdentityMapper;
@@ -133,7 +133,7 @@ public class NestedIterationsTest extends CompilerTestBase {
 			OptimizedPlan op = compileNoStats(p);
 			
 			// job graph generator should be able to translate this
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -171,7 +171,7 @@ public class NestedIterationsTest extends CompilerTestBase {
 			OptimizedPlan op = compileNoStats(p);
 			
 			// job graph generator should be able to translate this
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/PipelineBreakerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/PipelineBreakerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/PipelineBreakerTest.java
index e5983d9..86f01b0 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/PipelineBreakerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/PipelineBreakerTest.java
@@ -149,7 +149,7 @@ public class PipelineBreakerTest extends CompilerTestBase {
 				DataSet<Long> initialSource = env.generateSequence(1, 10);
 				
 				Configuration conf= new Configuration();
-				conf.setString(PactCompiler.HINT_LOCAL_STRATEGY, PactCompiler.HINT_LOCAL_STRATEGY_NESTEDLOOP_BLOCKED_OUTER_FIRST);
+				conf.setString(Optimizer.HINT_LOCAL_STRATEGY, Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_BLOCKED_OUTER_FIRST);
 				initialSource
 					.map(new IdentityMapper<Long>())
 					.cross(initialSource).withParameters(conf)
@@ -171,7 +171,7 @@ public class PipelineBreakerTest extends CompilerTestBase {
 				DataSet<Long> initialSource = env.generateSequence(1, 10);
 				
 				Configuration conf= new Configuration();
-				conf.setString(PactCompiler.HINT_LOCAL_STRATEGY, PactCompiler.HINT_LOCAL_STRATEGY_NESTEDLOOP_BLOCKED_OUTER_SECOND);
+				conf.setString(Optimizer.HINT_LOCAL_STRATEGY, Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_BLOCKED_OUTER_SECOND);
 				initialSource
 					.map(new IdentityMapper<Long>())
 					.cross(initialSource).withParameters(conf)
@@ -194,7 +194,7 @@ public class PipelineBreakerTest extends CompilerTestBase {
 				DataSet<Long> initialSource = env.generateSequence(1, 10);
 				
 				Configuration conf= new Configuration();
-				conf.setString(PactCompiler.HINT_LOCAL_STRATEGY, PactCompiler.HINT_LOCAL_STRATEGY_NESTEDLOOP_STREAMED_OUTER_FIRST);
+				conf.setString(Optimizer.HINT_LOCAL_STRATEGY, Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_STREAMED_OUTER_FIRST);
 				initialSource
 					.map(new IdentityMapper<Long>())
 					.cross(initialSource).withParameters(conf)
@@ -217,7 +217,7 @@ public class PipelineBreakerTest extends CompilerTestBase {
 				DataSet<Long> initialSource = env.generateSequence(1, 10);
 				
 				Configuration conf= new Configuration();
-				conf.setString(PactCompiler.HINT_LOCAL_STRATEGY, PactCompiler.HINT_LOCAL_STRATEGY_NESTEDLOOP_STREAMED_OUTER_SECOND);
+				conf.setString(Optimizer.HINT_LOCAL_STRATEGY, Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_STREAMED_OUTER_SECOND);
 				initialSource
 					.map(new IdentityMapper<Long>())
 					.cross(initialSource).withParameters(conf)

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/ReduceAllTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/ReduceAllTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/ReduceAllTest.java
index 23fa311..f6885c5 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/ReduceAllTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/ReduceAllTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.api.java.record.operators.FileDataSink;
 import org.apache.flink.api.java.record.operators.FileDataSource;
 import org.apache.flink.api.java.record.operators.ReduceOperator;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.util.DummyInputFormat;
 import org.apache.flink.optimizer.util.DummyOutputFormat;
 import org.apache.flink.optimizer.util.IdentityReduce;
@@ -51,7 +51,7 @@ public class ReduceAllTest extends CompilerTestBase {
 		
 		try {
 			OptimizedPlan oPlan = compileNoStats(plan);
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			jobGen.compileJobGraph(oPlan);
 		} catch(CompilerException ce) {
 			ce.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/SemanticPropertiesAPIToPlanTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/SemanticPropertiesAPIToPlanTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/SemanticPropertiesAPIToPlanTest.java
index 8356e94..1fe16bb 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/SemanticPropertiesAPIToPlanTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/SemanticPropertiesAPIToPlanTest.java
@@ -63,7 +63,7 @@ public class SemanticPropertiesAPIToPlanTest extends CompilerTestBase {
 		oPlan.accept(new Visitor<PlanNode>() {
 			@Override
 			public boolean preVisit(PlanNode visitable) {
-				if (visitable instanceof SingleInputPlanNode && visitable.getPactContract() instanceof ReduceOperatorBase) {
+				if (visitable instanceof SingleInputPlanNode && visitable.getProgramOperator() instanceof ReduceOperatorBase) {
 					for (Channel input: visitable.getInputs()) {
 						GlobalProperties gprops = visitable.getGlobalProperties();
 						LocalProperties lprops = visitable.getLocalProperties();
@@ -78,7 +78,7 @@ public class SemanticPropertiesAPIToPlanTest extends CompilerTestBase {
 								lprops.getGroupedFields().contains(1));
 					}
 				}
-				if (visitable instanceof SingleInputPlanNode && visitable.getPactContract() instanceof MapOperatorBase) {
+				if (visitable instanceof SingleInputPlanNode && visitable.getProgramOperator() instanceof MapOperatorBase) {
 					for (Channel input: visitable.getInputs()) {
 						GlobalProperties gprops = visitable.getGlobalProperties();
 						LocalProperties lprops = visitable.getLocalProperties();
@@ -124,7 +124,7 @@ public class SemanticPropertiesAPIToPlanTest extends CompilerTestBase {
 		oPlan.accept(new Visitor<PlanNode>() {
 			@Override
 			public boolean preVisit(PlanNode visitable) {
-				if (visitable instanceof DualInputPlanNode && visitable.getPactContract() instanceof JoinOperatorBase) {
+				if (visitable instanceof DualInputPlanNode && visitable.getProgramOperator() instanceof JoinOperatorBase) {
 					DualInputPlanNode node = ((DualInputPlanNode) visitable);
 
 					final Channel inConn1 = node.getInput1();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionBetweenDynamicAndStaticPathTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionBetweenDynamicAndStaticPathTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionBetweenDynamicAndStaticPathTest.java
index 6647483..92b4fc5 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionBetweenDynamicAndStaticPathTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionBetweenDynamicAndStaticPathTest.java
@@ -30,7 +30,7 @@ import org.apache.flink.optimizer.plan.Channel;
 import org.apache.flink.optimizer.plan.NAryUnionPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.junit.Test;
 
 @SuppressWarnings("serial")
@@ -80,7 +80,7 @@ public class UnionBetweenDynamicAndStaticPathTest extends CompilerTestBase {
 			assertEquals(0.5, mixedUnion.getInput1().getRelativeTempMemory(), 0.0);
 			assertEquals(0.0, mixedUnion.getInput2().getRelativeTempMemory(), 0.0);
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -133,7 +133,7 @@ public class UnionBetweenDynamicAndStaticPathTest extends CompilerTestBase {
 				assertTrue(c.isOnDynamicPath());
 			}
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java
index cb4bce4..5d15ed8 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java
@@ -43,7 +43,7 @@ import org.apache.flink.optimizer.plan.NAryUnionPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.PlanNode;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.util.DummyInputFormat;
 import org.apache.flink.optimizer.util.DummyOutputFormat;
 import org.apache.flink.optimizer.util.IdentityReduce;
@@ -78,7 +78,7 @@ public class UnionPropertyPropagationTest extends CompilerTestBase {
 		
 		OptimizedPlan oPlan = compileNoStats(plan);
 		
-		NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+		JobGraphGenerator jobGen = new JobGraphGenerator();
 		
 		// Compile plan to verify that no error is thrown
 		jobGen.compileJobGraph(oPlan);
@@ -87,7 +87,7 @@ public class UnionPropertyPropagationTest extends CompilerTestBase {
 			
 			@Override
 			public boolean preVisit(PlanNode visitable) {
-				if (visitable instanceof SingleInputPlanNode && visitable.getPactContract() instanceof ReduceOperator) {
+				if (visitable instanceof SingleInputPlanNode && visitable.getProgramOperator() instanceof ReduceOperator) {
 					for (Channel inConn : visitable.getInputs()) {
 						Assert.assertTrue("Reduce should just forward the input if it is already partitioned",
 								inConn.getShipStrategy() == ShipStrategyType.FORWARD); 
@@ -126,7 +126,7 @@ public class UnionPropertyPropagationTest extends CompilerTestBase {
 		// return the plan
 		Plan plan = env.createProgramPlan("Test union on new java-api");
 		OptimizedPlan oPlan = compileNoStats(plan);
-		NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+		JobGraphGenerator jobGen = new JobGraphGenerator();
 		
 		// Compile plan to verify that no error is thrown
 		jobGen.compileJobGraph(oPlan);
@@ -139,7 +139,7 @@ public class UnionPropertyPropagationTest extends CompilerTestBase {
 				/* Test on the union output connections
 				 * It must be under the GroupOperator and the strategy should be forward
 				 */
-				if (visitable instanceof SingleInputPlanNode && visitable.getPactContract() instanceof GroupReduceOperatorBase){
+				if (visitable instanceof SingleInputPlanNode && visitable.getProgramOperator() instanceof GroupReduceOperatorBase){
 					final Channel inConn = ((SingleInputPlanNode) visitable).getInput();
 					Assert.assertTrue("Union should just forward the Partitioning",
 							inConn.getShipStrategy() == ShipStrategyType.FORWARD ); 
@@ -156,7 +156,7 @@ public class UnionPropertyPropagationTest extends CompilerTestBase {
 						final Channel inConn = inputs.next();
 						PlanNode inNode = inConn.getSource();
 						Assert.assertTrue("Input of Union should be FlatMapOperators",
-								inNode.getPactContract() instanceof FlatMapOperatorBase);
+								inNode.getProgramOperator() instanceof FlatMapOperatorBase);
 						Assert.assertTrue("Shipment strategy under union should partition the data",
 								inConn.getShipStrategy() == ShipStrategyType.PARTITION_HASH); 
 					}

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
index f327259..1e4124c 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.junit.Test;
 
 import static org.junit.Assert.fail;
@@ -44,7 +44,7 @@ public class UnionReplacementTest extends CompilerTestBase {
 	
 			Plan plan = env.createProgramPlan();
 			OptimizedPlan oPlan = compileNoStats(plan);
-			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			JobGraphGenerator jobGen = new JobGraphGenerator();
 			jobGen.compileJobGraph(oPlan);
 		}
 		catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationCornerCasesTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationCornerCasesTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationCornerCasesTest.java
index d10803e..80c0bda 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationCornerCasesTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationCornerCasesTest.java
@@ -28,7 +28,7 @@ import org.apache.flink.api.java.operators.DeltaIteration;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.WorksetIterationPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.junit.Test;
 
 @SuppressWarnings("serial")
@@ -52,7 +52,7 @@ public class WorksetIterationCornerCasesTest extends CompilerTestBase {
 			WorksetIterationPlanNode wipn = (WorksetIterationPlanNode) op.getDataSinks().iterator().next().getInput().getSource();
 			assertTrue(wipn.getSolutionSetPlanNode().getOutgoingChannels().isEmpty());
 			
-			NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+			JobGraphGenerator jgg = new JobGraphGenerator();
 			jgg.compileJobGraph(op);
 		}
 		catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationsRecordApiCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationsRecordApiCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationsRecordApiCompilerTest.java
index f17842e..6e7c0a3 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationsRecordApiCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/WorksetIterationsRecordApiCompilerTest.java
@@ -34,7 +34,7 @@ import org.apache.flink.api.java.record.operators.ReduceOperator;
 import org.apache.flink.optimizer.plan.DualInputPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.util.DummyInputFormat;
 import org.apache.flink.optimizer.util.DummyMatchStub;
 import org.apache.flink.optimizer.util.DummyNonPreservingMatchStub;
@@ -106,7 +106,7 @@ public class WorksetIterationsRecordApiCompilerTest extends CompilerTestBase {
 		assertTrue( (ss1 == ShipStrategyType.FORWARD && ss2 == ShipStrategyType.PARTITION_HASH) ||
 					(ss2 == ShipStrategyType.FORWARD && ss1 == ShipStrategyType.PARTITION_HASH) );
 		
-		new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+		new JobGraphGenerator().compileJobGraph(oPlan);
 	}
 	
 	@Test
@@ -150,7 +150,7 @@ public class WorksetIterationsRecordApiCompilerTest extends CompilerTestBase {
 		assertEquals(ShipStrategyType.PARTITION_HASH, joinWithSolutionSetNode.getOutgoingChannels().get(0).getShipStrategy());
 		assertEquals(ShipStrategyType.PARTITION_HASH, joinWithSolutionSetNode.getOutgoingChannels().get(1).getShipStrategy());
 		
-		new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+		new JobGraphGenerator().compileJobGraph(oPlan);
 	}
 	
 	@Test
@@ -193,7 +193,7 @@ public class WorksetIterationsRecordApiCompilerTest extends CompilerTestBase {
 		assertEquals(1, joinWithSolutionSetNode.getOutgoingChannels().size());
 		assertEquals(ShipStrategyType.FORWARD, joinWithSolutionSetNode.getOutgoingChannels().get(0).getShipStrategy());
 		
-		new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+		new JobGraphGenerator().compileJobGraph(oPlan);
 	}
 	
 	private Plan getRecordTestPlan(boolean joinPreservesSolutionSet, boolean mapBeforeSolutionDelta) {

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/BinaryCustomPartitioningCompatibilityTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/BinaryCustomPartitioningCompatibilityTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/BinaryCustomPartitioningCompatibilityTest.java
index e876fbb..0273659 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/BinaryCustomPartitioningCompatibilityTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/BinaryCustomPartitioningCompatibilityTest.java
@@ -31,7 +31,7 @@ import org.apache.flink.optimizer.plan.DualInputPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
 import org.apache.flink.optimizer.plan.SinkPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.testfunctions.DummyCoGroupFunction;
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.junit.Test;
@@ -75,7 +75,7 @@ public class BinaryCustomPartitioningCompatibilityTest extends CompilerTestBase
 			assertEquals(partitioner, partitioner1.getInput().getPartitioner());
 			assertEquals(partitioner, partitioner2.getInput().getPartitioner());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -120,7 +120,7 @@ public class BinaryCustomPartitioningCompatibilityTest extends CompilerTestBase
 			assertEquals(partitioner, partitioner1.getInput().getPartitioner());
 			assertEquals(partitioner, partitioner2.getInput().getPartitioner());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/CustomPartitioningTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/CustomPartitioningTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/CustomPartitioningTest.java
index f6d8d0e..d397ea2 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/CustomPartitioningTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/custompartition/CustomPartitioningTest.java
@@ -64,17 +64,17 @@ public class CustomPartitioningTest extends CompilerTestBase {
 			SingleInputPlanNode balancer = (SingleInputPlanNode) partitioner.getInput().getSource();
 			
 			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			assertEquals(parallelism, sink.getDegreeOfParallelism());
+			assertEquals(parallelism, sink.getParallelism());
 			
 			assertEquals(ShipStrategyType.FORWARD, mapper.getInput().getShipStrategy());
-			assertEquals(parallelism, mapper.getDegreeOfParallelism());
+			assertEquals(parallelism, mapper.getParallelism());
 			
 			assertEquals(ShipStrategyType.PARTITION_CUSTOM, partitioner.getInput().getShipStrategy());
 			assertEquals(part, partitioner.getInput().getPartitioner());
-			assertEquals(parallelism, partitioner.getDegreeOfParallelism());
+			assertEquals(parallelism, partitioner.getParallelism());
 			
 			assertEquals(ShipStrategyType.PARTITION_FORCED_REBALANCE, balancer.getInput().getShipStrategy());
-			assertEquals(parallelism, balancer.getDegreeOfParallelism());
+			assertEquals(parallelism, balancer.getParallelism());
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -134,17 +134,17 @@ public class CustomPartitioningTest extends CompilerTestBase {
 			SingleInputPlanNode balancer = (SingleInputPlanNode) partitioner.getInput().getSource();
 			
 			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			assertEquals(parallelism, sink.getDegreeOfParallelism());
+			assertEquals(parallelism, sink.getParallelism());
 			
 			assertEquals(ShipStrategyType.FORWARD, mapper.getInput().getShipStrategy());
-			assertEquals(parallelism, mapper.getDegreeOfParallelism());
+			assertEquals(parallelism, mapper.getParallelism());
 			
 			assertEquals(ShipStrategyType.PARTITION_CUSTOM, partitioner.getInput().getShipStrategy());
 			assertEquals(part, partitioner.getInput().getPartitioner());
-			assertEquals(parallelism, partitioner.getDegreeOfParallelism());
+			assertEquals(parallelism, partitioner.getParallelism());
 			
 			assertEquals(ShipStrategyType.PARTITION_FORCED_REBALANCE, balancer.getInput().getShipStrategy());
-			assertEquals(parallelism, balancer.getDegreeOfParallelism());
+			assertEquals(parallelism, balancer.getParallelism());
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -206,23 +206,23 @@ public class CustomPartitioningTest extends CompilerTestBase {
 			SingleInputPlanNode balancer = (SingleInputPlanNode) keyExtractor.getInput().getSource();
 			
 			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			assertEquals(parallelism, sink.getDegreeOfParallelism());
+			assertEquals(parallelism, sink.getParallelism());
 			
 			assertEquals(ShipStrategyType.FORWARD, mapper.getInput().getShipStrategy());
-			assertEquals(parallelism, mapper.getDegreeOfParallelism());
+			assertEquals(parallelism, mapper.getParallelism());
 			
 			assertEquals(ShipStrategyType.FORWARD, keyRemover.getInput().getShipStrategy());
-			assertEquals(parallelism, keyRemover.getDegreeOfParallelism());
+			assertEquals(parallelism, keyRemover.getParallelism());
 			
 			assertEquals(ShipStrategyType.PARTITION_CUSTOM, partitioner.getInput().getShipStrategy());
 			assertEquals(part, partitioner.getInput().getPartitioner());
-			assertEquals(parallelism, partitioner.getDegreeOfParallelism());
+			assertEquals(parallelism, partitioner.getParallelism());
 			
 			assertEquals(ShipStrategyType.FORWARD, keyExtractor.getInput().getShipStrategy());
-			assertEquals(parallelism, keyExtractor.getDegreeOfParallelism());
+			assertEquals(parallelism, keyExtractor.getParallelism());
 			
 			assertEquals(ShipStrategyType.PARTITION_FORCED_REBALANCE, balancer.getInput().getShipStrategy());
-			assertEquals(parallelism, balancer.getDegreeOfParallelism());
+			assertEquals(parallelism, balancer.getParallelism());
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeClosedBranchingTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeClosedBranchingTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeClosedBranchingTest.java
index 2c5d235..cb4bd78 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeClosedBranchingTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeClosedBranchingTest.java
@@ -246,7 +246,7 @@ public class DataExchangeModeClosedBranchingTest extends CompilerTestBase {
 
 	private SinkPlanNode findSink(Collection<SinkPlanNode> collection, String name) {
 		for (SinkPlanNode node : collection) {
-			String nodeName = node.getOptimizerNode().getPactContract().getName();
+			String nodeName = node.getOptimizerNode().getOperator().getName();
 			if (nodeName != null && nodeName.equals(name)) {
 				return node;
 			}

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
index 6c0e88b..6b2691a 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
@@ -171,7 +171,7 @@ public class DataExchangeModeOpenBranchingTest extends CompilerTestBase {
 
 	private SinkPlanNode findSink(Collection<SinkPlanNode> collection, String name) {
 		for (SinkPlanNode node : collection) {
-			String nodeName = node.getOptimizerNode().getPactContract().getName();
+			String nodeName = node.getOptimizerNode().getOperator().getName();
 			if (nodeName != null && nodeName.equals(name)) {
 				return node;
 			}

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
index dae3c41..fe33635 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.DiscardingOutputFormat;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.dag.DataSinkNode;
 import org.apache.flink.optimizer.dag.OptimizerNode;
 import org.apache.flink.optimizer.dag.SingleInputNode;
@@ -292,8 +292,8 @@ public class PipelineBreakingTest {
 	}
 
 	private static List<DataSinkNode> convertPlan(Plan p) {
-		PactCompiler.GraphCreatingVisitor dagCreator =
-				new PactCompiler.GraphCreatingVisitor(17, p.getExecutionConfig().getExecutionMode());
+		Optimizer.GraphCreatingVisitor dagCreator =
+				new Optimizer.GraphCreatingVisitor(17, p.getExecutionConfig().getExecutionMode());
 
 		// create the DAG
 		p.accept(dagCreator);
@@ -312,8 +312,8 @@ public class PipelineBreakingTest {
 				rootNode = new SinkJoiner(rootNode, iter.next());
 			}
 		}
-		rootNode.accept(new PactCompiler.IdAndEstimatesVisitor(null));
-		rootNode.accept(new PactCompiler.BranchesVisitor());
+		rootNode.accept(new Optimizer.IdAndEstimatesVisitor(null));
+		rootNode.accept(new Optimizer.BranchesVisitor());
 
 		return sinks;
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
index 32aeab9..a683968 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
@@ -73,9 +73,9 @@ public class GroupReduceCompilationTest extends CompilerTestBase implements java
 			assertEquals(DriverStrategy.ALL_GROUP_REDUCE, reduceNode.getDriverStrategy());
 			
 			// check DOP
-			assertEquals(1, sourceNode.getDegreeOfParallelism());
-			assertEquals(1, reduceNode.getDegreeOfParallelism());
-			assertEquals(1, sinkNode.getDegreeOfParallelism());
+			assertEquals(1, sourceNode.getParallelism());
+			assertEquals(1, reduceNode.getParallelism());
+			assertEquals(1, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -121,10 +121,10 @@ public class GroupReduceCompilationTest extends CompilerTestBase implements java
 			assertEquals(DriverStrategy.ALL_GROUP_REDUCE_COMBINE, combineNode.getDriverStrategy());
 			
 			// check DOP
-			assertEquals(8, sourceNode.getDegreeOfParallelism());
-			assertEquals(8, combineNode.getDegreeOfParallelism());
-			assertEquals(1, reduceNode.getDegreeOfParallelism());
-			assertEquals(1, sinkNode.getDegreeOfParallelism());
+			assertEquals(8, sourceNode.getParallelism());
+			assertEquals(8, combineNode.getParallelism());
+			assertEquals(1, reduceNode.getParallelism());
+			assertEquals(1, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -172,9 +172,9 @@ public class GroupReduceCompilationTest extends CompilerTestBase implements java
 			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
 			
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -229,10 +229,10 @@ public class GroupReduceCompilationTest extends CompilerTestBase implements java
 			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
 			
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -285,12 +285,12 @@ public class GroupReduceCompilationTest extends CompilerTestBase implements java
 			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
 			
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, keyExtractor.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, keyExtractor.getParallelism());
 			
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, keyProjector.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, keyProjector.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -351,13 +351,13 @@ public class GroupReduceCompilationTest extends CompilerTestBase implements java
 			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
 			
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, keyExtractor.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, keyExtractor.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
 			
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, keyProjector.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, keyProjector.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
index 54b24dd..37a8e81 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
@@ -34,7 +34,7 @@ import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
 import org.apache.flink.optimizer.plan.SinkPlanNode;
 import org.apache.flink.optimizer.plan.WorksetIterationPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.testfunctions.IdentityMapper;
 import org.junit.Test;
 
@@ -53,7 +53,7 @@ public class IterationCompilerTest extends CompilerTestBase {
 			Plan p = env.createProgramPlan();
 			OptimizedPlan op = compileNoStats(p);
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -81,7 +81,7 @@ public class IterationCompilerTest extends CompilerTestBase {
 			Plan p = env.createProgramPlan();
 			OptimizedPlan op = compileNoStats(p);
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -120,7 +120,7 @@ public class IterationCompilerTest extends CompilerTestBase {
 			assertTrue(union.getCostWeight() >= 1);
 			
 			// see that the jobgraph generator can translate this
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();
@@ -179,7 +179,7 @@ public class IterationCompilerTest extends CompilerTestBase {
 			assertTrue(solutionDeltaUnion.isOnDynamicPath());
 			assertTrue(solutionDeltaUnion.getCostWeight() >= 1);
 			
-			new NepheleJobGraphGenerator().compileJobGraph(op);
+			new JobGraphGenerator().compileJobGraph(op);
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
index bbdad4a..0724a9f 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
@@ -72,9 +72,9 @@ public class ReduceCompilationTest extends CompilerTestBase implements java.io.S
 			assertEquals(reduceNode, sinkNode.getInput().getSource());
 			
 			// check DOP
-			assertEquals(1, sourceNode.getDegreeOfParallelism());
-			assertEquals(1, reduceNode.getDegreeOfParallelism());
-			assertEquals(1, sinkNode.getDegreeOfParallelism());
+			assertEquals(1, sourceNode.getParallelism());
+			assertEquals(1, reduceNode.getParallelism());
+			assertEquals(1, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -122,10 +122,10 @@ public class ReduceCompilationTest extends CompilerTestBase implements java.io.S
 			assertEquals(DriverStrategy.ALL_REDUCE, combineNode.getDriverStrategy());
 			
 			// check DOP
-			assertEquals(8, sourceNode.getDegreeOfParallelism());
-			assertEquals(8, combineNode.getDegreeOfParallelism());
-			assertEquals(1, reduceNode.getDegreeOfParallelism());
-			assertEquals(1, sinkNode.getDegreeOfParallelism());
+			assertEquals(8, sourceNode.getParallelism());
+			assertEquals(8, combineNode.getParallelism());
+			assertEquals(1, reduceNode.getParallelism());
+			assertEquals(1, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -180,10 +180,10 @@ public class ReduceCompilationTest extends CompilerTestBase implements java.io.S
 			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
 			
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -244,13 +244,13 @@ public class ReduceCompilationTest extends CompilerTestBase implements java.io.S
 			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
 			
 			// check DOP
-			assertEquals(6, sourceNode.getDegreeOfParallelism());
-			assertEquals(6, keyExtractor.getDegreeOfParallelism());
-			assertEquals(6, combineNode.getDegreeOfParallelism());
+			assertEquals(6, sourceNode.getParallelism());
+			assertEquals(6, keyExtractor.getParallelism());
+			assertEquals(6, combineNode.getParallelism());
 			
-			assertEquals(8, reduceNode.getDegreeOfParallelism());
-			assertEquals(8, keyProjector.getDegreeOfParallelism());
-			assertEquals(8, sinkNode.getDegreeOfParallelism());
+			assertEquals(8, reduceNode.getParallelism());
+			assertEquals(8, keyProjector.getParallelism());
+			assertEquals(8, sinkNode.getParallelism());
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/java/WorksetIterationsJavaApiCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/WorksetIterationsJavaApiCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/WorksetIterationsJavaApiCompilerTest.java
index 658cf7a..8720aa7 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/WorksetIterationsJavaApiCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/WorksetIterationsJavaApiCompilerTest.java
@@ -37,7 +37,7 @@ import org.apache.flink.optimizer.CompilerTestBase;
 import org.apache.flink.optimizer.plan.DualInputPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.SingleInputPlanNode;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.apache.flink.util.Collector;
 import org.junit.Test;
@@ -93,7 +93,7 @@ public class WorksetIterationsJavaApiCompilerTest extends CompilerTestBase {
 			assertTrue( (ss1 == ShipStrategyType.FORWARD && ss2 == ShipStrategyType.PARTITION_HASH) ||
 						(ss2 == ShipStrategyType.FORWARD && ss1 == ShipStrategyType.PARTITION_HASH) );
 		
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -137,7 +137,7 @@ public class WorksetIterationsJavaApiCompilerTest extends CompilerTestBase {
 			assertEquals(ShipStrategyType.PARTITION_HASH, joinWithSolutionSetNode.getOutgoingChannels().get(0).getShipStrategy());
 			assertEquals(ShipStrategyType.PARTITION_HASH, joinWithSolutionSetNode.getOutgoingChannels().get(1).getShipStrategy());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());
@@ -182,7 +182,7 @@ public class WorksetIterationsJavaApiCompilerTest extends CompilerTestBase {
 			assertEquals(1, joinWithSolutionSetNode.getOutgoingChannels().size());
 			assertEquals(ShipStrategyType.FORWARD, joinWithSolutionSetNode.getOutgoingChannels().get(0).getShipStrategy());
 			
-			new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+			new JobGraphGenerator().compileJobGraph(oPlan);
 		}
 		catch (Exception e) {
 			System.err.println(e.getMessage());

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/CoGroupOnConflictingPartitioningsTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/CoGroupOnConflictingPartitioningsTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/CoGroupOnConflictingPartitioningsTest.java
index 21d95ee..e7807c9 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/CoGroupOnConflictingPartitioningsTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/CoGroupOnConflictingPartitioningsTest.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.optimizer.CompilerException;
 import org.apache.flink.optimizer.CompilerTestBase;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.testfunctions.DummyCoGroupFunction;
 import org.apache.flink.configuration.Configuration;
 import org.junit.Test;
@@ -42,8 +42,8 @@ public class CoGroupOnConflictingPartitioningsTest extends CompilerTestBase {
 			DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<Long, Long>(0L, 0L));
 			
 			Configuration cfg = new Configuration();
-			cfg.setString(PactCompiler.HINT_SHIP_STRATEGY_FIRST_INPUT, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION_HASH);
-			cfg.setString(PactCompiler.HINT_SHIP_STRATEGY_SECOND_INPUT, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION_RANGE);
+			cfg.setString(Optimizer.HINT_SHIP_STRATEGY_FIRST_INPUT, Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH);
+			cfg.setString(Optimizer.HINT_SHIP_STRATEGY_SECOND_INPUT, Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE);
 			
 			input.coGroup(input).where(0).equalTo(0)
 				.with(new DummyCoGroupFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>())

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/JoinOnConflictingPartitioningsTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/JoinOnConflictingPartitioningsTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/JoinOnConflictingPartitioningsTest.java
index 4cfa189..9171cc7 100644
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/JoinOnConflictingPartitioningsTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/optimizer/operators/JoinOnConflictingPartitioningsTest.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.optimizer.CompilerException;
 import org.apache.flink.optimizer.CompilerTestBase;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.configuration.Configuration;
 import org.junit.Test;
 
@@ -41,8 +41,8 @@ public class JoinOnConflictingPartitioningsTest extends CompilerTestBase {
 			DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<Long, Long>(0L, 0L));
 			
 			Configuration cfg = new Configuration();
-			cfg.setString(PactCompiler.HINT_SHIP_STRATEGY_FIRST_INPUT, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION_HASH);
-			cfg.setString(PactCompiler.HINT_SHIP_STRATEGY_SECOND_INPUT, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION_RANGE);
+			cfg.setString(Optimizer.HINT_SHIP_STRATEGY_FIRST_INPUT, Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH);
+			cfg.setString(Optimizer.HINT_SHIP_STRATEGY_SECOND_INPUT, Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE);
 			
 			input.join(input).where(0).equalTo(0)
 				.withParameters(cfg)

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-staging/flink-spargel/src/test/java/org/apache/flink/spargel/java/SpargelCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-spargel/src/test/java/org/apache/flink/spargel/java/SpargelCompilerTest.java b/flink-staging/flink-spargel/src/test/java/org/apache/flink/spargel/java/SpargelCompilerTest.java
index 79e04fb..019345f 100644
--- a/flink-staging/flink-spargel/src/test/java/org/apache/flink/spargel/java/SpargelCompilerTest.java
+++ b/flink-staging/flink-spargel/src/test/java/org/apache/flink/spargel/java/SpargelCompilerTest.java
@@ -68,24 +68,24 @@ public class SpargelCompilerTest extends CompilerTestBase {
 			// check the sink
 			SinkPlanNode sink = op.getDataSinks().iterator().next();
 			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			assertEquals(DEFAULT_PARALLELISM, sink.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, sink.getParallelism());
 			
 			// check the iteration
 			WorksetIterationPlanNode iteration = (WorksetIterationPlanNode) sink.getInput().getSource();
-			assertEquals(DEFAULT_PARALLELISM, iteration.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, iteration.getParallelism());
 			
 			// check the solution set join and the delta
 			PlanNode ssDelta = iteration.getSolutionSetDeltaPlanNode();
 			assertTrue(ssDelta instanceof DualInputPlanNode); // this is only true if the update functions preserves the partitioning
 			
 			DualInputPlanNode ssJoin = (DualInputPlanNode) ssDelta;
-			assertEquals(DEFAULT_PARALLELISM, ssJoin.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, ssJoin.getParallelism());
 			assertEquals(ShipStrategyType.PARTITION_HASH, ssJoin.getInput1().getShipStrategy());
 			assertEquals(new FieldList(0), ssJoin.getInput1().getShipStrategyKeys());
 			
 			// check the workset set join
 			DualInputPlanNode edgeJoin = (DualInputPlanNode) ssJoin.getInput1().getSource();
-			assertEquals(DEFAULT_PARALLELISM, edgeJoin.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, edgeJoin.getParallelism());
 			assertEquals(ShipStrategyType.PARTITION_HASH, edgeJoin.getInput1().getShipStrategy());
 			assertEquals(ShipStrategyType.FORWARD, edgeJoin.getInput2().getShipStrategy());
 			assertTrue(edgeJoin.getInput1().getTempMode().isCached());
@@ -143,24 +143,24 @@ public class SpargelCompilerTest extends CompilerTestBase {
 			// check the sink
 			SinkPlanNode sink = op.getDataSinks().iterator().next();
 			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			assertEquals(DEFAULT_PARALLELISM, sink.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, sink.getParallelism());
 			
 			// check the iteration
 			WorksetIterationPlanNode iteration = (WorksetIterationPlanNode) sink.getInput().getSource();
-			assertEquals(DEFAULT_PARALLELISM, iteration.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, iteration.getParallelism());
 			
 			// check the solution set join and the delta
 			PlanNode ssDelta = iteration.getSolutionSetDeltaPlanNode();
 			assertTrue(ssDelta instanceof DualInputPlanNode); // this is only true if the update functions preserves the partitioning
 			
 			DualInputPlanNode ssJoin = (DualInputPlanNode) ssDelta;
-			assertEquals(DEFAULT_PARALLELISM, ssJoin.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, ssJoin.getParallelism());
 			assertEquals(ShipStrategyType.PARTITION_HASH, ssJoin.getInput1().getShipStrategy());
 			assertEquals(new FieldList(0), ssJoin.getInput1().getShipStrategyKeys());
 			
 			// check the workset set join
 			DualInputPlanNode edgeJoin = (DualInputPlanNode) ssJoin.getInput1().getSource();
-			assertEquals(DEFAULT_PARALLELISM, edgeJoin.getDegreeOfParallelism());
+			assertEquals(DEFAULT_PARALLELISM, edgeJoin.getParallelism());
 			assertEquals(ShipStrategyType.PARTITION_HASH, edgeJoin.getInput1().getShipStrategy());
 			assertEquals(ShipStrategyType.FORWARD, edgeJoin.getInput2().getShipStrategy());
 			assertTrue(edgeJoin.getInput1().getTempMode().isCached());

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-test-utils/src/main/java/org/apache/flink/test/compiler/util/CompilerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/compiler/util/CompilerTestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/compiler/util/CompilerTestBase.java
index acf20d3..788327a 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/compiler/util/CompilerTestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/compiler/util/CompilerTestBase.java
@@ -30,7 +30,7 @@ import org.apache.flink.api.common.io.FileInputFormat.FileBaseStatistics;
 import org.apache.flink.api.common.operators.GenericDataSourceBase;
 import org.apache.flink.api.common.operators.Operator;
 import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.costs.DefaultCostEstimator;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plan.PlanNode;
@@ -57,9 +57,9 @@ public abstract class CompilerTestBase {
 	
 	protected DataStatistics dataStats;
 	
-	protected PactCompiler withStatsCompiler;
+	protected Optimizer withStatsCompiler;
 	
-	protected PactCompiler noStatsCompiler;
+	protected Optimizer noStatsCompiler;
 	
 	private int statCounter;
 	
@@ -68,10 +68,10 @@ public abstract class CompilerTestBase {
 	@Before
 	public void setup() {
 		this.dataStats = new DataStatistics();
-		this.withStatsCompiler = new PactCompiler(this.dataStats, new DefaultCostEstimator());
+		this.withStatsCompiler = new Optimizer(this.dataStats, new DefaultCostEstimator());
 		this.withStatsCompiler.setDefaultDegreeOfParallelism(DEFAULT_PARALLELISM);
 		
-		this.noStatsCompiler = new PactCompiler(null, new DefaultCostEstimator());
+		this.noStatsCompiler = new Optimizer(null, new DefaultCostEstimator());
 		this.noStatsCompiler.setDefaultDegreeOfParallelism(DEFAULT_PARALLELISM);
 	}
 	
@@ -113,7 +113,7 @@ public abstract class CompilerTestBase {
 			HashMap<String, ArrayList<PlanNode>> map = new HashMap<String, ArrayList<PlanNode>>();
 			
 			for (PlanNode n : p.getAllNodes()) {
-				Operator<?> c = n.getOriginalOptimizerNode().getPactContract();
+				Operator<?> c = n.getOriginalOptimizerNode().getOperator();
 				String name = c.getName();
 				
 				ArrayList<PlanNode> list = map.get(name);
@@ -126,7 +126,7 @@ public abstract class CompilerTestBase {
 				boolean shouldAdd = true;
 				for (Iterator<PlanNode> iter = list.iterator(); iter.hasNext();) {
 					PlanNode in = iter.next();
-					if (in.getOriginalOptimizerNode().getPactContract() == c) {
+					if (in.getOriginalOptimizerNode().getOperator() == c) {
 						// is this the child or is our node the child
 						if (in instanceof SingleInputPlanNode && n instanceof SingleInputPlanNode) {
 							SingleInputPlanNode thisNode = (SingleInputPlanNode) n;

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
index 7cd3ff0..67a4797 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
@@ -23,10 +23,10 @@ import akka.actor.ActorRef;
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.client.JobClient;
 import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -72,14 +72,14 @@ public abstract class RecordAPITestBase extends AbstractTestBase {
 			Assert.fail("Error: Cannot obtain Pact plan. Did the thest forget to override either 'getPactPlan()' or 'getJobGraph()' ?");
 		}
 		
-		PactCompiler pc = new PactCompiler(new DataStatistics());
+		Optimizer pc = new Optimizer(new DataStatistics());
 		OptimizedPlan op = pc.compile(p);
 		
 		if (printPlan) {
 			System.out.println(new PlanJSONDumpGenerator().getOptimizerPlanAsJSON(op)); 
 		}
 
-		NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+		JobGraphGenerator jgg = new JobGraphGenerator();
 		return jgg.compileJobGraph(op);
 	}
 	

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
index db1fc4d..44f35e7 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
@@ -24,10 +24,10 @@ import org.apache.flink.api.common.Plan;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.ExecutionEnvironmentFactory;
 import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
 import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.runtime.client.JobClient;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.junit.Assert;
@@ -49,7 +49,7 @@ public class TestEnvironment extends ExecutionEnvironment {
 		try {
 			OptimizedPlan op = compileProgram(jobName);
 
-			NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+			JobGraphGenerator jgg = new JobGraphGenerator();
 			JobGraph jobGraph = jgg.compileJobGraph(op);
 
 			ActorRef client = this.executor.getJobClient();
@@ -80,7 +80,7 @@ public class TestEnvironment extends ExecutionEnvironment {
 	private OptimizedPlan compileProgram(String jobName) {
 		Plan p = createProgramPlan(jobName);
 
-		PactCompiler pc = new PactCompiler(new DataStatistics());
+		Optimizer pc = new Optimizer(new DataStatistics());
 		return pc.compile(p);
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
index 9ab62e1..400ed3a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
@@ -38,9 +38,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.PactCompiler;
+import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.util.StringUtils;
 import org.apache.hadoop.fs.FileSystem;
@@ -140,9 +140,9 @@ public abstract class CancellingTestBase {
 	}
 
 	private JobGraph getJobGraph(final Plan plan) throws Exception {
-		final PactCompiler pc = new PactCompiler(new DataStatistics());
+		final Optimizer pc = new Optimizer(new DataStatistics());
 		final OptimizedPlan op = pc.compile(plan);
-		final NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+		final JobGraphGenerator jgg = new JobGraphGenerator();
 		return jgg.compileJobGraph(op);
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsCoGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsCoGroupTest.java b/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsCoGroupTest.java
index 1b73c4b..b38b784 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsCoGroupTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsCoGroupTest.java
@@ -27,7 +27,7 @@ import org.apache.flink.optimizer.plan.SinkPlanNode;
 import org.apache.flink.optimizer.plan.SourcePlanNode;
 import org.apache.flink.optimizer.plan.WorksetIterationPlanNode;
 import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.runtime.operators.DriverStrategy;
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.apache.flink.runtime.operators.util.LocalStrategy;
@@ -130,7 +130,7 @@ public class ConnectedComponentsCoGroupTest extends CompilerTestBase {
 		// check the caches
 		Assert.assertTrue(TempMode.CACHED == neighborsJoin.getInput2().getTempMode());
 		
-		NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+		JobGraphGenerator jgg = new JobGraphGenerator();
 		jgg.compileJobGraph(optPlan);
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a9150b30/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsTest.java b/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsTest.java
index 2c23eaf..dcc9c15 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/compiler/iterations/ConnectedComponentsTest.java
@@ -40,7 +40,7 @@ import org.apache.flink.optimizer.plan.SinkPlanNode;
 import org.apache.flink.optimizer.plan.SourcePlanNode;
 import org.apache.flink.optimizer.plan.WorksetIterationPlanNode;
 import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
-import org.apache.flink.optimizer.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.runtime.operators.DriverStrategy;
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.apache.flink.runtime.operators.util.LocalStrategy;
@@ -152,7 +152,7 @@ public class ConnectedComponentsTest extends CompilerTestBase {
 		Assert.assertTrue(TempMode.PIPELINE_BREAKER == iter.getInitialWorksetInput().getTempMode() ||
 							LocalStrategy.SORT == iter.getInitialWorksetInput().getLocalStrategy());
 		
-		NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+		JobGraphGenerator jgg = new JobGraphGenerator();
 		jgg.compileJobGraph(optPlan);
 	}
 	
@@ -233,7 +233,7 @@ public class ConnectedComponentsTest extends CompilerTestBase {
 		Assert.assertTrue(TempMode.PIPELINE_BREAKER == iter.getInitialWorksetInput().getTempMode() ||
 							LocalStrategy.SORT == iter.getInitialWorksetInput().getLocalStrategy());
 		
-		NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+		JobGraphGenerator jgg = new JobGraphGenerator();
 		jgg.compileJobGraph(optPlan);
 	}
 	


Mime
View raw message