PhysicalPlanGenerator implements Function<IRDAG, DAG<Stage, StageEdge>> { @Override public DAG<Stage, StageEdge> apply(final IRDAG irDAG) { final DAG<Stage, StageEdge> dagOfStages = stagePartitionIrDAG(irDAG); dagOfStages.getVertices().forEach(this::integrityCheck); handleDuplicateEdgeGroupProperty(dagOfStages); dagOfStages.storeJSON(dagDirectory, "plan-logical", "logical execution plan"); return dagOfStages; } @Inject private  PhysicalPlanGenerator(@Parameter(JobConf.DAGDirectory.class) final String dagDirectory); @Override DAG<Stage, StageEdge> apply(final IRDAG irDAG); DAG<Stage, StageEdge> stagePartitionIrDAG(final IRDAG irDAG);  }
@Test public void testBasic() throws Exception { final Injector injector = Tang.Factory.getTang().newInjector(); final PhysicalPlanGenerator physicalPlanGenerator = injector.getInstance(PhysicalPlanGenerator.class); final IRVertex v0 = newIRVertex(0, 5); final IRVertex v1 = newIRVertex(0, 3); final IRDAG irDAG = new IRDAG(new DAGBuilder<IRVertex, IREdge>() .addVertex(v0) .addVertex(v1) .connectVertices(newIREdge(v0, v1, CommunicationPatternProperty.Value.ONE_TO_ONE, DataFlowProperty.Value.PULL)) .buildWithoutSourceSinkCheck()); final DAG<Stage, StageEdge> stageDAG = physicalPlanGenerator.apply(irDAG); final Iterator<Stage> stages = stageDAG.getVertices().iterator(); final Stage s0 = stages.next(); final Stage s1 = stages.next(); }