PhysicalPlan implements Serializable { public List<PTContainer> getContainers() { return this.containers; } PhysicalPlan(LogicalPlan dag, PlanContext ctx); void setAntiAffinityForContainers(LogicalPlan dag, Collection<AffinityRule> affinityRules, Map<PTOperator, PTContainer> operatorContainerMap); void deployChanges(); Operator loadOperator(PTOperator oper); void removeTerminatedPartition(PTOperator p); PlanContext getContext(); LogicalPlan getLogicalPlan(); List<PTContainer> getContainers(); Map<Integer, PTOperator> getAllOperators(); List<PTOperator> getOperators(OperatorMeta logicalOperator); Collection<PTOperator> getAllOperators(OperatorMeta logicalOperator); List<PTOperator> getLeafOperators(); boolean hasMapping(OperatorMeta om); @VisibleForTesting List<PTOperator> getMergeOperators(OperatorMeta logicalOperator); Set<PTOperator> getDependents(Collection<PTOperator> operators); final void addLogicalOperator(OperatorMeta om); void removeLogicalStream(StreamMeta sm); void connectInput(InputPortMeta ipm); void removeLogicalOperator(OperatorMeta om); void setAvailableResources(int memoryMB); void onStatusUpdate(PTOperator oper); void syncCheckpoints(long startTime, long currentTime); Integer getStreamCodecIdentifier(StreamCodec<?> streamCodecInfo); @VisibleForTesting Map<StreamCodec<?>, Integer> getStreamCodecIdentifiers();  }
@Test public void testContainerCores() { LogicalPlan dag = new LogicalPlan(); dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent()); GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class); GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class); GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class); GenericTestOperator o4 = dag.addOperator("o4", GenericTestOperator.class); GenericTestOperator o5 = dag.addOperator("o5", GenericTestOperator.class); GenericTestOperator o6 = dag.addOperator("o6", GenericTestOperator.class); dag.setOperatorAttribute(o1,OperatorContext.VCORES,1); dag.setOperatorAttribute(o2,OperatorContext.VCORES,2); dag.setOperatorAttribute(o3,OperatorContext.VCORES,3); dag.setOperatorAttribute(o4,OperatorContext.VCORES,4); dag.setOperatorAttribute(o5,OperatorContext.VCORES,5); dag.setOperatorAttribute(o6,OperatorContext.VCORES,6); dag.addStream("o1.outport1", o1.outport1, o2.inport1).setLocality(Locality.CONTAINER_LOCAL); dag.addStream("o2.outport1", o2.outport1, o3.inport1, o4.inport1).setLocality(Locality.THREAD_LOCAL); dag.addStream("o3.output1", o3.outport1, o5.inport1).setLocality(Locality.THREAD_LOCAL); dag.addStream("o4.output1", o4.outport1, o5.inport2).setLocality(Locality.THREAD_LOCAL); dag.addStream("o5.output1", o5.outport1, o6.inport1).setLocality(Locality.CONTAINER_LOCAL); dag.setAttribute(LogicalPlan.CONTAINERS_MAX_COUNT, 2); PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext()); Assert.assertEquals("number of containers", 1, plan.getContainers().size()); Assert.assertEquals("vcores container 1 is 12", 12, plan.getContainers().get(0).getRequiredVCores()); }
@Test public void testContainersForSlidingWindow() { LogicalPlan dag = new LogicalPlan(); dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent()); GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class); GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class); GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class); dag.setOperatorAttribute(o1, OperatorContext.APPLICATION_WINDOW_COUNT, 4); dag.setOperatorAttribute(o1, OperatorContext.SLIDE_BY_WINDOW_COUNT, 2); dag.getOperatorMeta("o1").getMeta(o1.outport1).getUnifierMeta().getAttributes().put(OperatorContext.MEMORY_MB, 2000); dag.getOperatorMeta("o1").getMeta(o1.outport2).getUnifierMeta().getAttributes().put(OperatorContext.MEMORY_MB, 4000); dag.addStream("o1.outport1", o1.outport1, o2.inport1); dag.addStream("o1.outport2", o1.outport2, o2.inport2); dag.addStream("o2.outport1", o2.outport1, o3.inport1); PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext()); Assert.assertEquals("number of containers", 5, plan.getContainers().size()); boolean sawOutput1Slider = false; boolean sawOutput2Slider = false; for (PTContainer container : plan.getContainers()) { Assert.assertEquals("number of operators in each container is 1", container.operators.size(), 1); if (container.operators.get(0).isUnifier()) { String name = container.operators.get(0).getName(); if (name.equals("o1.outport1#slider")) { sawOutput1Slider = true; Assert.assertEquals("container memory is 2512", container.getRequiredMemoryMB(), 2512); } else if (name.equals("o1.outport2#slider")) { sawOutput2Slider = true; Assert.assertEquals("container memory is 2512", container.getRequiredMemoryMB(), 4512); } } } Assert.assertEquals("Found output1 slider", true, sawOutput1Slider); Assert.assertEquals("Found output2 slider", true, sawOutput2Slider); }
@Test public void testMxNPartitionForSlidingWindow() { LogicalPlan dag = new LogicalPlan(); dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent()); GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class); GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class); GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class); dag.setOperatorAttribute(o1, OperatorContext.APPLICATION_WINDOW_COUNT, 4); dag.setOperatorAttribute(o1, OperatorContext.SLIDE_BY_WINDOW_COUNT, 2); dag.setOperatorAttribute(o1, OperatorContext.PARTITIONER, new StatelessPartitioner<>(2)); dag.getOperatorMeta("o1").getMeta(o1.outport1).getUnifierMeta().getAttributes().put(OperatorContext.MEMORY_MB, 1024); dag.setOperatorAttribute(o2, OperatorContext.PARTITIONER, new StatelessPartitioner<>(2)); dag.setOperatorAttribute(o2, OperatorContext.SLIDE_BY_WINDOW_COUNT, 2); dag.setOperatorAttribute(o2, OperatorContext.APPLICATION_WINDOW_COUNT, 4); dag.addStream("o1.outport1", o1.outport1, o2.inport1); dag.addStream("o2.outport1", o2.outport1, o3.inport1); PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext()); Assert.assertEquals("number of containers", 9, plan.getContainers().size()); }
@Test public void testParallelPartitionForSlidingWindow() { LogicalPlan dag = new LogicalPlan(); dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent()); GenericTestOperator o1 = dag.addOperator("o1", GenericTestOperator.class); GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class); GenericTestOperator o3 = dag.addOperator("o3", GenericTestOperator.class); dag.setOperatorAttribute(o1, OperatorContext.SLIDE_BY_WINDOW_COUNT, 2); dag.setOperatorAttribute(o1, OperatorContext.PARTITIONER, new StatelessPartitioner<>(2)); dag.setInputPortAttribute(o2.inport1, PortContext.PARTITION_PARALLEL, true); dag.setOperatorAttribute(o1, OperatorContext.APPLICATION_WINDOW_COUNT, 4); dag.addStream("o1.outport1", o1.outport1, o2.inport1); dag.addStream("o2.outport1", o2.outport1, o3.inport1); PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext()); Assert.assertEquals("number of containers", 7, plan.getContainers().size()); }