PlanStagesGenerator implements ILogicalOperatorVisitor<Void, Void> { public List<PlanStage> getStages() { return stages; } PlanStagesGenerator(); @Override Void visitAggregateOperator(AggregateOperator op, Void arg); @Override Void visitRunningAggregateOperator(RunningAggregateOperator op, Void arg); @Override Void visitEmptyTupleSourceOperator(EmptyTupleSourceOperator op, Void arg); @Override Void visitGroupByOperator(GroupByOperator op, Void arg); @Override Void visitLimitOperator(LimitOperator op, Void arg); @Override Void visitInnerJoinOperator(InnerJoinOperator op, Void arg); @Override Void visitLeftOuterJoinOperator(LeftOuterJoinOperator op, Void arg); @Override Void visitNestedTupleSourceOperator(NestedTupleSourceOperator op, Void arg); @Override Void visitOrderOperator(OrderOperator op, Void arg); @Override Void visitAssignOperator(AssignOperator op, Void arg); @Override Void visitSelectOperator(SelectOperator op, Void arg); @Override Void visitDelegateOperator(DelegateOperator op, Void arg); @Override Void visitProjectOperator(ProjectOperator op, Void arg); @Override Void visitReplicateOperator(ReplicateOperator op, Void arg); @Override Void visitSplitOperator(SplitOperator op, Void arg); @Override Void visitMaterializeOperator(MaterializeOperator op, Void arg); @Override Void visitScriptOperator(ScriptOperator op, Void arg); @Override Void visitSubplanOperator(SubplanOperator op, Void arg); @Override Void visitSinkOperator(SinkOperator op, Void arg); @Override Void visitUnionOperator(UnionAllOperator op, Void arg); @Override Void visitIntersectOperator(IntersectOperator op, Void arg); @Override Void visitUnnestOperator(UnnestOperator op, Void arg); @Override Void visitLeftOuterUnnestOperator(LeftOuterUnnestOperator op, Void arg); @Override Void visitUnnestMapOperator(UnnestMapOperator op, Void arg); @Override Void visitLeftOuterUnnestMapOperator(LeftOuterUnnestMapOperator op, Void arg); @Override Void visitDataScanOperator(DataSourceScanOperator op, Void arg); @Override Void visitDistinctOperator(DistinctOperator op, Void arg); @Override Void visitExchangeOperator(ExchangeOperator op, Void arg); @Override Void visitWriteOperator(WriteOperator op, Void arg); @Override Void visitDistributeResultOperator(DistributeResultOperator op, Void arg); @Override Void visitWriteResultOperator(WriteResultOperator op, Void arg); @Override Void visitInsertDeleteUpsertOperator(InsertDeleteUpsertOperator op, Void arg); @Override Void visitIndexInsertDeleteUpsertOperator(IndexInsertDeleteUpsertOperator op, Void arg); @Override Void visitTokenizeOperator(TokenizeOperator op, Void arg); @Override Void visitForwardOperator(ForwardOperator op, Void arg); @Override Void visitWindowOperator(WindowOperator op, Void arg); List<PlanStage> getStages();  }
@Test public void noBlockingPlan() throws AlgebricksException { EmptyTupleSourceOperator ets = new EmptyTupleSourceOperator(); ets.setExecutionMode(UNPARTITIONED); ets.setPhysicalOperator(new EmptyTupleSourcePOperator()); AssignOperator assignOperator = new AssignOperator(Collections.emptyList(), null); assignOperator.setExecutionMode(UNPARTITIONED); assignOperator.setPhysicalOperator(new AssignPOperator()); assignOperator.getInputs().add(new MutableObject<>(ets)); ExchangeOperator exchange = new ExchangeOperator(); exchange.setExecutionMode(UNPARTITIONED); exchange.setPhysicalOperator(new OneToOneExchangePOperator()); exchange.getInputs().add(new MutableObject<>(assignOperator)); DistributeResultOperator resultOperator = new DistributeResultOperator(null, null, null); resultOperator.setExecutionMode(UNPARTITIONED); resultOperator.setPhysicalOperator(new DistributeResultPOperator()); resultOperator.getInputs().add(new MutableObject<>(exchange)); ALogicalPlanImpl plan = new ALogicalPlanImpl(Collections.singletonList(new MutableObject<>(resultOperator))); List<PlanStage> stages = ResourceUtils.getStages(plan); final int expectedStages = 1; Assert.assertEquals(expectedStages, stages.size()); validateStages(stages, resultOperator, exchange, ets, assignOperator); final long expectedMemory = stages.get(0).getOperators().size() * FRAME_SIZE; assertRequiredMemory(stages, expectedMemory); }
@Test public void testNonBlockingGroupByOrderBy() throws AlgebricksException { EmptyTupleSourceOperator ets = new EmptyTupleSourceOperator(); ets.setExecutionMode(PARTITIONED); ets.setPhysicalOperator(new EmptyTupleSourcePOperator()); DataSourceScanOperator scanOperator = new DataSourceScanOperator(Collections.emptyList(), null); scanOperator.setExecutionMode(PARTITIONED); scanOperator.setPhysicalOperator(new DataSourceScanPOperator(null)); scanOperator.getInputs().add(new MutableObject<>(ets)); ExchangeOperator exchange = new ExchangeOperator(); exchange.setExecutionMode(PARTITIONED); exchange.setPhysicalOperator(new OneToOneExchangePOperator()); exchange.getInputs().add(new MutableObject<>(scanOperator)); GroupByOperator groupByOperator = new GroupByOperator(); groupByOperator.setExecutionMode(PARTITIONED); groupByOperator.setPhysicalOperator(new PreclusteredGroupByPOperator(Collections.emptyList(), true)); groupByOperator.getInputs().add(new MutableObject<>(exchange)); OrderOperator orderOperator = new OrderOperator(); orderOperator.setExecutionMode(PARTITIONED); orderOperator.setPhysicalOperator(new StableSortPOperator()); orderOperator.getInputs().add(new MutableObject<>(groupByOperator)); DistributeResultOperator resultOperator = new DistributeResultOperator(null, null, null); resultOperator.setExecutionMode(PARTITIONED); resultOperator.setPhysicalOperator(new DistributeResultPOperator()); resultOperator.getInputs().add(new MutableObject<>(orderOperator)); ALogicalPlanImpl plan = new ALogicalPlanImpl(Collections.singletonList(new MutableObject<>(resultOperator))); final List<PlanStage> stages = ResourceUtils.getStages(plan); validateStages(stages, ets, exchange, groupByOperator, orderOperator, resultOperator); final int expectedStages = 2; Assert.assertEquals(expectedStages, stages.size()); long orderOperatorRequiredMemory = AbstractStableSortPOperator.MIN_FRAME_LIMIT_FOR_SORT * FRAME_SIZE * PARALLELISM; long groupByOperatorRequiredMemory = AbstractGroupByPOperator.MIN_FRAME_LIMIT_FOR_GROUP_BY * FRAME_SIZE * PARALLELISM; long exchangeRequiredMemory = PARALLELISM * FRAME_SIZE; long scanOperatorRequiredMemory = PARALLELISM * FRAME_SIZE; long etsRequiredMemory = FRAME_SIZE * PARALLELISM; final long expectedMemory = orderOperatorRequiredMemory + groupByOperatorRequiredMemory + exchangeRequiredMemory + scanOperatorRequiredMemory + etsRequiredMemory; assertRequiredMemory(stages, expectedMemory); }
@Test public void testJoinGroupby() throws AlgebricksException { EmptyTupleSourceOperator ets1 = new EmptyTupleSourceOperator(); ets1.setExecutionMode(PARTITIONED); ets1.setPhysicalOperator(new EmptyTupleSourcePOperator()); DataSourceScanOperator scanOperator1 = new DataSourceScanOperator(Collections.emptyList(), null); scanOperator1.setExecutionMode(PARTITIONED); scanOperator1.setPhysicalOperator(new DataSourceScanPOperator(null)); scanOperator1.getInputs().add(new MutableObject<>(ets1)); EmptyTupleSourceOperator ets2 = new EmptyTupleSourceOperator(); ets2.setExecutionMode(PARTITIONED); ets2.setPhysicalOperator(new EmptyTupleSourcePOperator()); DataSourceScanOperator scanOperator2 = new DataSourceScanOperator(Collections.emptyList(), null); scanOperator2.setExecutionMode(PARTITIONED); scanOperator2.setPhysicalOperator(new DataSourceScanPOperator(null)); scanOperator2.getInputs().add(new MutableObject<>(ets2)); InnerJoinOperator firstJoin = new InnerJoinOperator(new MutableObject<>(ConstantExpression.TRUE)); firstJoin.setExecutionMode(PARTITIONED); firstJoin.setPhysicalOperator(new NestedLoopJoinPOperator(firstJoin.getJoinKind(), AbstractJoinPOperator.JoinPartitioningType.BROADCAST)); firstJoin.getInputs().add(new MutableObject<>(scanOperator1)); firstJoin.getInputs().add(new MutableObject<>(scanOperator2)); ExchangeOperator exchangeOperator1 = new ExchangeOperator(); exchangeOperator1.setExecutionMode(PARTITIONED); exchangeOperator1.setPhysicalOperator(new HashPartitionExchangePOperator(Collections.emptyList(), null)); exchangeOperator1.getInputs().add(new MutableObject<>(firstJoin)); EmptyTupleSourceOperator ets3 = new EmptyTupleSourceOperator(); ets3.setExecutionMode(PARTITIONED); ets3.setPhysicalOperator(new EmptyTupleSourcePOperator()); GroupByOperator groupByOperator = new GroupByOperator(); groupByOperator.setPhysicalOperator(new ExternalGroupByPOperator(Collections.emptyList())); groupByOperator.setExecutionMode(LOCAL); groupByOperator.getInputs().add(new MutableObject<>(ets3)); ExchangeOperator exchangeOperator2 = new ExchangeOperator(); exchangeOperator2.setExecutionMode(PARTITIONED); exchangeOperator2.setPhysicalOperator(new HashPartitionExchangePOperator(Collections.emptyList(), null)); exchangeOperator2.getInputs().add(new MutableObject<>(groupByOperator)); LeftOuterJoinOperator secondJoin = new LeftOuterJoinOperator(new MutableObject<>(ConstantExpression.TRUE)); secondJoin.setExecutionMode(PARTITIONED); secondJoin.setPhysicalOperator(new NestedLoopJoinPOperator(secondJoin.getJoinKind(), AbstractJoinPOperator.JoinPartitioningType.BROADCAST)); secondJoin.getInputs().add(new MutableObject<>(exchangeOperator1)); secondJoin.getInputs().add(new MutableObject<>(exchangeOperator2)); DistributeResultOperator resultOperator = new DistributeResultOperator(null, null, null); resultOperator.setExecutionMode(PARTITIONED); resultOperator.setPhysicalOperator(new DistributeResultPOperator()); resultOperator.getInputs().add(new MutableObject<>(secondJoin)); ALogicalPlanImpl plan = new ALogicalPlanImpl(Collections.singletonList(new MutableObject<>(resultOperator))); List<PlanStage> stages = ResourceUtils.getStages(plan); final int expectedStages = 4; Assert.assertEquals(expectedStages, stages.size()); validateStages(stages, ets1, scanOperator1, ets2, scanOperator2, firstJoin, exchangeOperator1, ets3, groupByOperator, exchangeOperator2, secondJoin, resultOperator); long resultOperatorRequiredMemory = FRAME_SIZE * PARALLELISM; long secondJoinRequiredMemory = AbstractJoinPOperator.MIN_FRAME_LIMIT_FOR_JOIN * FRAME_SIZE * PARALLELISM; long exchangeOperator1RequiredMemory = 2 * MAX_BUFFER_PER_CONNECTION * PARALLELISM * PARALLELISM * FRAME_SIZE; long firstJoinRequiredMemory = AbstractJoinPOperator.MIN_FRAME_LIMIT_FOR_JOIN * FRAME_SIZE * PARALLELISM; long scanOperator1RequiredMemory = FRAME_SIZE * PARALLELISM; long ets1RequiredMemory = FRAME_SIZE * PARALLELISM; long expectedMemory = resultOperatorRequiredMemory + secondJoinRequiredMemory + exchangeOperator1RequiredMemory + firstJoinRequiredMemory + scanOperator1RequiredMemory + ets1RequiredMemory; assertRequiredMemory(stages, expectedMemory); }
@Test public void testReplicateSortJoin() throws AlgebricksException { EmptyTupleSourceOperator ets = new EmptyTupleSourceOperator(); ets.setExecutionMode(PARTITIONED); ets.setPhysicalOperator(new EmptyTupleSourcePOperator()); DataSourceScanOperator scanOperator = new DataSourceScanOperator(Collections.emptyList(), null); scanOperator.setExecutionMode(PARTITIONED); scanOperator.setPhysicalOperator(new DataSourceScanPOperator(null)); scanOperator.getInputs().add(new MutableObject<>(ets)); ReplicateOperator replicateOperator = new ReplicateOperator(2); replicateOperator.setExecutionMode(PARTITIONED); replicateOperator.setPhysicalOperator(new ReplicatePOperator()); replicateOperator.getInputs().add(new MutableObject<>(scanOperator)); OrderOperator order1 = new OrderOperator(); order1.setExecutionMode(PARTITIONED); order1.setPhysicalOperator(new StableSortPOperator()); order1.getInputs().add(new MutableObject<>(replicateOperator)); OrderOperator order2 = new OrderOperator(); order2.setExecutionMode(PARTITIONED); order2.setPhysicalOperator(new StableSortPOperator()); order2.getInputs().add(new MutableObject<>(replicateOperator)); LeftOuterJoinOperator secondJoin = new LeftOuterJoinOperator(new MutableObject<>(ConstantExpression.TRUE)); secondJoin.setExecutionMode(PARTITIONED); secondJoin.setPhysicalOperator(new NestedLoopJoinPOperator(secondJoin.getJoinKind(), AbstractJoinPOperator.JoinPartitioningType.BROADCAST)); secondJoin.getInputs().add(new MutableObject<>(order1)); secondJoin.getInputs().add(new MutableObject<>(order2)); DistributeResultOperator resultOperator = new DistributeResultOperator(null, null, null); resultOperator.setExecutionMode(PARTITIONED); resultOperator.setPhysicalOperator(new DistributeResultPOperator()); resultOperator.getInputs().add(new MutableObject<>(secondJoin)); ALogicalPlanImpl plan = new ALogicalPlanImpl(Collections.singletonList(new MutableObject<>(resultOperator))); List<PlanStage> stages = ResourceUtils.getStages(plan); final int expectedStages = 3; Assert.assertEquals(expectedStages, stages.size()); validateStages(stages); long order1RequiredMemory = AbstractStableSortPOperator.MIN_FRAME_LIMIT_FOR_SORT * FRAME_SIZE * PARALLELISM; long order2RequiredMemory = AbstractStableSortPOperator.MIN_FRAME_LIMIT_FOR_SORT * FRAME_SIZE * PARALLELISM; long replicateOperatorRequiredMemory = FRAME_SIZE * PARALLELISM; long scanOperator1RequiredMemory = FRAME_SIZE * PARALLELISM; long etsRequiredMemory = FRAME_SIZE * PARALLELISM; long expectedMemory = order1RequiredMemory + order2RequiredMemory + replicateOperatorRequiredMemory + scanOperator1RequiredMemory + etsRequiredMemory; assertRequiredMemory(stages, expectedMemory); }