PhysicalPlan implements Serializable { public List<PTOperator> getOperators(OperatorMeta logicalOperator) { return this.logicalToPTOperator.get(logicalOperator).partitions; } 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 testDefaultPartitioning() { LogicalPlan dag = new LogicalPlan(); dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent()); GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class); GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class); dag.addStream("node1.outport1", node1.outport1, node2.inport2, node2.inport1); int initialPartitionCount = 5; OperatorMeta node2Decl = dag.getMeta(node2); node2Decl.getAttributes().put(OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(initialPartitionCount)); PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext()); List<PTOperator> n2Instances = plan.getOperators(node2Decl); Assert.assertEquals("partition instances " + n2Instances, initialPartitionCount, n2Instances.size()); List<Integer> assignedPartitionKeys = Lists.newArrayList(); for (int i = 0; i < n2Instances.size(); i++) { PTOperator n2Partition = n2Instances.get(i); Assert.assertNotNull("partition keys null: " + n2Partition, n2Partition.getPartitionKeys()); Map<InputPort<?>, PartitionKeys> pkeys = n2Partition.getPartitionKeys(); Assert.assertEquals("partition keys size: " + pkeys, 1, pkeys.size()); InputPort<?> expectedPort = node2.inport2; Assert.assertEquals("partition port: " + pkeys, expectedPort, pkeys.keySet().iterator().next()); Assert.assertEquals("partition mask: " + pkeys, "111", Integer.toBinaryString(pkeys.get(expectedPort).mask)); Set<Integer> pks = pkeys.get(expectedPort).partitions; Assert.assertTrue("number partition keys: " + pkeys, pks.size() == 1 || pks.size() == 2); assignedPartitionKeys.addAll(pks); } int expectedMask = Integer.parseInt("111", 2); Assert.assertEquals("assigned partitions ", expectedMask + 1, assignedPartitionKeys.size()); for (int i = 0; i <= expectedMask; i++) { Assert.assertTrue("" + assignedPartitionKeys, assignedPartitionKeys.contains(i)); } }
@Test public void testStatsListenerContextWrappers() { LogicalPlan dag = new LogicalPlan(); dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent()); StatsListenerOperator o1 = dag.addOperator("o1", new StatsListenerOperator()); GenericTestOperator o2 = dag.addOperator("o2", new GenericTestOperator()); dag.setAttribute(o2, OperatorContext.STATS_LISTENERS, Lists.<StatsListener>newArrayList(mock(StatsListener.class))); GenericTestOperator o3 = dag.addOperator("o3", new GenericTestOperator()); dag.setAttribute(o3, OperatorContext.STATS_LISTENERS, Lists.<StatsListener>newArrayList(mock(StatsListenerWithContext.class))); StatsListenerOperatorOld o4 = dag.addOperator("o4", new StatsListenerOperatorOld()); PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext()); PTOperator p1 = plan.getOperators(dag.getMeta(o1)).get(0); StatsListener l = p1.statsListeners.get(0); Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext); PTOperator p2 = plan.getOperators(dag.getMeta(o2)).get(0); l = p1.statsListeners.get(0); Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext); PTOperator p3 = plan.getOperators(dag.getMeta(o3)).get(0); l = p1.statsListeners.get(0); Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext); PTOperator p4 = plan.getOperators(dag.getMeta(o4)).get(0); l = p1.statsListeners.get(0); Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext); }