based on tag: 0.287
presto的 scheduler 是 SqlQueryScheduler 这个类控制的, 这个class主要是负责调度物理执行计划。调度具体的每个SqlStageExecution. 这个Stage可以当成Fragment的一个概念
他会先把所有的stage创建一个schedule计划。一共有两种schedule计划，一个是all-at-once，另一个是phased，还有一个 adaptive-phased (其实就是根据stage数量来决定到底是all-at-once还是phased).
// AdaptivePhasedExecutionPolicy.java createExecutionScheduleif (stages.size() > getMaxStageCountForEagerScheduling(session)) {return new PhasedExecutionSchedule(stages);}else {return new AllAtOnceExecutionSchedule(stages);}
SqlQueryScheduler 的主要流程是什么呢?
scheduler:schedules = // 每个 stages (section 可以理解为单个查询的所有SQL的fragment) 创建一个schedules while (schedules.all_finish()) {stage_wait_schedule = schedule.get_next_stages();stages.addAll(stage_wait_schedule)for stage in stages:stage.scheduler.schedule() // (这里是stage的scheduler，比如发async rpc, 分配split/计算节点)collect_block_schedule_result() // 收集block schedule resultwait_if_has_block_events() // 这里是等block result 的 future 最多等1s}
可以看到其实SQLSchedule的流程就是先获取下次要调度哪些stage，然后执行每个stage中的scheduler的schedule()方法，
那么这个 ExecutionSchedule (schedule计划) 里面是如何提供这个顺序的呢?
AllAtOnceExecutionSchedule:
给出一个调度顺序，然后把所有的stage都拿出来调度。
PhasedExecutionSchedule:
获取一个拓扑执行序，每次返回一部分来执行, 如果存在join这类的fragment，会优先调度 build side，再调度probe side。对于union这类的会一个一个执行
他的大概算法就是在plan中添加一个 build 到 probe 的 edge
对于这样的一个plan (里面每个节点代表一个fragment)，会先调度 build side，然后再调度 probe side。
但是对于broadcast join 这个算法会出现环，所以他这里做了一个处理:
@VisibleForTestingstatic List<Set<PlanFragmentId>> extractPhases(Collection<PlanFragment> fragments){// Build a graph where the plan fragments are vertexes and the edges represent// a before -> after relationship. For example, a join hash build has an edge// to the join probe.// 先构建这个 graphGraph<PlanFragmentId, DefaultEdge> graph = new DefaultDirectedGraph<>(DefaultEdge.class);fragments.forEach(fragment -> graph.addVertex(fragment.getId()));Visitor visitor = new Visitor(fragments, graph);for (PlanFragment fragment : fragments) {visitor.processFragment(fragment.getId());}// Computes all the strongly connected components of the directed graph.// These are the "phases" which hold the set of fragments that must be started// at the same time to avoid deadlock.// 找到所有强关系集合 如果有A->B B->A 这两个就是强关系 List<Set<PlanFragmentId>> components = new KosarajuStrongConnectivityInspector<>(graph).stronglyConnectedSets();Map<PlanFragmentId, Set<PlanFragmentId>> componentMembership = new HashMap<>();for (Set<PlanFragmentId> component : components) {for (PlanFragmentId planFragmentId : component) {componentMembership.put(planFragmentId, component);}}// build graph of components (phases)// 只有两个边不是强关系才会添加Graph<Set<PlanFragmentId>, DefaultEdge> componentGraph = new DefaultDirectedGraph<>(DefaultEdge.class);components.forEach(componentGraph::addVertex);for (DefaultEdge edge : graph.edgeSet()) {PlanFragmentId source = graph.getEdgeSource(edge);PlanFragmentId target = graph.getEdgeTarget(edge);Set<PlanFragmentId> from = componentMembership.get(source);Set<PlanFragmentId> to = componentMembership.get(target);if (!from.equals(to)) { // the topological order iterator below doesn't include vertices that have self-edges, so don't add themcomponentGraph.addEdge(from, to);}}// 所以对于broadcast来说Join的fragment和 build side的fragment会同时被调度, 因为他们是一组强关系List<Set<PlanFragmentId>> schedulePhases = ImmutableList.copyOf(new TopologicalOrderIterator<>(componentGraph));return schedulePhases;}
那么对于一个task，是否可以被调度多次呢？
file: PhasedExecutionSchedule.javaprivate void removeCompletedStages(){for (Iterator<StageExecutionAndScheduler> stageIterator = activeSources.iterator(); stageIterator.hasNext(); ) {StageExecutionState state = stageIterator.next().getStageExecution().getState();// state 的状态有 PLANNED SCHEDULING SCHEDULED RUNNING 以及 isDone(CANCELED FINISHED 等)// 所以一个一次调度之后有可能是SCHEDUING (比如还有split没有分配)，但是没调度完成，所以下次调度还要调度这些 taskif (state == SCHEDULED || state == RUNNING || state.isDone()) {stageIterator.remove();}}}
