问题导读：
1、Scheduler任务中Distributed Plan、Scan Range是什么？
2、Scheduler基本接口有哪些？
3、QuerySchedule这个类如何理解？
4、SimpleScheduler接口如何实现的？
Scheduler的任务
1、相关概念：Distributed Plan
在Frontend中已经把SQL转换成了single node plan，然后又将其切分成了distributed plan。代码见Planner::createPlan和DistributedPlanner::createPlanFragments.
比如左图的single node plan会切成为右图的distributed plan（图片来源自impala官方PPT：Query Compilation in Impala）。在impala-shell中set explain_level=3后用explain可以看到distributed plan。
上图的Distributed Plan中，
一个带颜色的圆角矩形是一个Plan Fragment;
Plan Fragment间的箭头是Fragment间的数据流动;
Plan Fragment内矩形间的箭头是Fragment内各个Plan Node的数据流动;
上图其实省略了Distributed Plan中每个非叶子结点里的ExchangeNode，其是用来接收其它Plan Fragment的数据的。
一个Plan Fragment会有一个或多个instance，运行在不同impalad上。
2、相关概念：Scan Range
在distributed plan中，每个scan node已经设置好要处理的scan ranges，即有哪些输入文件要处理，每个有哪些block。从Frontend传来的scan range对应一个block，由于一个block有多个replica，因此一个scan range会有多个location。了解scan range的概念非常重要，下面是相关的thrift定义：
// PlanNodes.thrift
struct TScanRange {
// one of these must be set for every TScanRange
1: optional THdfsFileSplit hdfs_file_split
2: optional THBaseKeyRange hbase_key_range
}
struct THdfsFileSplit {
// File name (not the full path). The path is assumed to be the
// 'location' of the THdfsPartition referenced by partition_id.
1: required string file_name
2: required i64 offset
3: required i64 length
// ID of partition within the THdfsTable associated with this scan node
4: required i64 partition_id
5: required i64 file_length
6: required CatalogObjects.THdfsCompression file_compression
7: required i64 mtime // last modified time
}
struct THBaseKeyRange {
1: optional string startKey // inclusive
2: optional string stopKey // exclusive
}
一个ScanRange要么是HDFS文件上的一部分（用file_name，offset和length等来表示），要么是Hbase一片连续的rowKey（用[startKey, endKey)表示）。
// Planer.thrift
struct TScanRangeLocation { // 对应HDFS上一个block位置，定位到硬盘编号
// Index into TQueryExecRequest.host_list.
1: required i32 host_idx;
2: optional i32 volume_id = -1
3: optional bool is_cached = false // If true, this block is cached on this server
} struct TScanRangeLocations { // 一个ScanRange对应的各个replica的位置列表
1: required PlanNodes.TScanRange scan_range
// non-empty list
2: list<TScanRangeLocation> locations
}
TScanRangeLocations就记录了Scan Range的位置列表，这个类型在TQueryExecRequest中会被用到。TQueryExecRequest的成员per_node_scan_ranges记录了每个ScanNode要处理的一系列ScanRange，每个ScanRange用TScanRangeLocations来表示。因此per_node_scan_ranges的类型是 map<Types.TPlanNodeId, list<Planner.TScanRangeLocations>>.
TQueryExecRequest是Frontend传给Backend的执行计划，Backend会生成 fragment 的执行计划发送给其它 impalad 去执行，这其中就包含了Scheduler的工作。impalad之间通信的数据结构定义在ImpalaInternalService.thrift中，关于ScanRange使用的是TScanRangeParams，此时已经确定了Scan任务要放在哪台机器上去执行（就是接收信息的impalad），因此不再需要location信息里的host_idx。
// ImpalaInternalService.thrift
// A scan range plus the parameters needed to execute that scan.
struct TScanRangeParams {
1: required PlanNodes.TScanRange scan_range
2: optional i32 volume_id = -1
3: optional bool is_cached = false
4: optional bool is_remote
}
3、Scheduler的任务
我们知道HDFS是有多备份的，一个block经常会有多于一个副本，因此在HDFS上的位置也会有多于一个。那么读取的时候到底在哪个复本上执行，这就是Scheduler要做的事情。另外其它的上层plan fragment也需要确定在哪些机器上执行。
代码注释如是说：
Given a list of resources and locations returns a list of hosts on which to execute plan fragments requiring those resources.
这其实很像MapReduce里对mapper和reducer的调度。
Scheduler基本接口
//该接口初始化本Scheduler，返回时表示已获取了做调试决策所需要的所有资源。
virtual Status Init() = ;
//定义BackendList类型表示一列Backend描述符。
typedef std::vector<TBackendDescriptor> BackendList;
//该接口输入是一组 host/port 二元组表示的data location，输出是一组运行在那些机器上或附近（比如同机架）的Backend列表。
virtual Status GetBackends(const std::vector<TNetworkAddress>& data_locations, BackendList* backends) = ;
//该接口输入是一个 host/port 二元组表示的 data location，输出是一个运行在该机器上或附近的Backend。
virtual Status GetBackend(const TNetworkAddress& data_location, TBackendDescriptor* backend) = ;
//返回是否有Backend运行在资源所在的机器上。
virtual bool HasLocalBackend(const TNetworkAddress& data_location) = ;
//返回该scheduler知道的所有Backend
virtual void GetAllKnownBackends(BackendList* backends) = ;
/*
该接口最主要的输入是QueryScheduler::request()，即一个TQueryExecRequest对象，描述了query的执行计划。参数里的Coordinator只提供一些全局的信息，比如user name等。
TQueryExecRequest的per_node_scan_ranges记录了要处理的scan ranges。
该接口将这些scan range分配到对应的机器上去，分配信息（即调度信息）记录在QuerySchedule的fragment_exec_params_中。
如果开启了资源管理，即impala运行在Yarn上时，该接口也会向resource manager（通过Llama）申请资源来执行query。该接口在资源申请被同意或拒绝前是阻塞的。
*/
virtual Status Schedule(Coordinator* coord, QuerySchedule* schedule) = ;
//释放资源
virtual Status Release(QuerySchedule* schedule) = ;
// Impapa on yarn callback
virtual void HandlePreemptedReservation(const TUniqueId& reservation_id) = ;
virtual void HandlePreemptedResource(const TUniqueId& client_resource_id) = ;
virtual void HandleLostResource(const TUniqueId& client_resource_id) = ;
QuerySchedule
在读懂SimpleScheduler的代码前需要先过一下QuerySchedule这个类，SimpleScheduler的任务是把QueryScheduler对象补充完整。
一个QueryScheduler包含了coordinator生成fragment执行请求从而启动query执行所需要的所有信息。Coordinator根据QueryScheduler生成FragmentExecParams，发送给各个impalad去执行fragment instance。
如果开启了资源管理（即Impala on Yarn），则QueryScheduler中还会包含需要申请的资源和已经获得的资源。
query-schedule.h中首先定义了两个类型：
//ScanNode到ScanRange列表的映射
typedef std::map<TPlanNodeId, std::vector<TScanRangeParams> > PerNodeScanRanges;
//impalad到PerNodeScanRanges的映射
typedef boost::unordered_map<TNetworkAddress, PerNodeScanRanges> FragmentScanRangeAssignment;
每个Scan Node可能有多个scan range，分配到不同impalad上去执行。因此这里有个impalad到其任务的映射，即它要处理哪些scan node的哪些scan range。
QuerySchedule有几个重要的成员变量：
<ignore_js_op>
其中plan_node_to_fragment_idx_和plan_node_to_plan_node_idx_是在进入SimpleScheduler::Schedule前就生成好的了。Schedule函数最主要的任务是填充fragment_exec_params_和unique_hosts等成员变量。fragment_exec_params是一个FragmentExecParams数组，记录了每个fragment的执行请求，用来发送给其它impalad去运行fragment instance。FragmentExecParams的定义如下：
struct FragmentExecParams {
std::vector<TNetworkAddress> hosts; // 将在哪些backend上执行，每个backend执行一个instance
std::vector<TUniqueId> instance_ids; // 各个instance的id
std::vector<TPlanFragmentDestination> destinations; // 所有输出目标，TPlanFragmentDestination是目标fragment的各个instance id及其所运行在的server
std::map<PlanNodeId, int> per_exch_num_senders; // 本fragment里各个exchange node的输入instance总数
FragmentScanRangeAssignment scan_range_assignment; // scan range的调度信息，比如hdfs上replica的选择信息
int sender_id_base; // 本fragment作为一个sender时的base id
};
这里的sender_id_base需要再解释一下。除了根fragment以外的每个fragment都有另一个fragment作为输出目标，除了叶子fragment外的所有fragment都有若干输入fragment。因此除了根fragment以外的每个fragment都需要扮演sender的角色，除了叶子fragment外的所有fragment也都要扮演receiver的角色。每一个fragment都有一个或多个instance，receiver会为自己的各个sender (instance)分配id用来区分，id从0开始。为了方便，我们给同一个fragment的不同instance分配连续的id，这样只需要记住起始id和该fragment的instance数目就够了。成员变量sender_id_base记录的就是自己作为sender时，由receiver分配的起始id。
SimpleScheduler接口实现
目前Impala on Yarn还不是很流行，我们可以先跳过跟resource management相关的代码，这样看起来会更容易些。最主要的逻辑还是Schedule接口：
Status SimpleScheduler::Schedule(Coordinator* coord, QuerySchedule* schedule)
主要做三件事情：
- ComputeScanRangeAssignment(schedule->request(), schedule));
填充各个FragmentExecParams对象中的scan_range_assignment
- ComputeFragmentHosts(schedule->request(), schedule);
填充各个FragmentExecParams对象中的hosts，即计算每个fragment instance在哪个impalad上执行
- ComputeFragmentExecParams(schedule->request(), schedule);
填充各个FragmentExecParams对象中剩下的内容，即destinations、per_exch_num_senders和sender_id_base
调用完这三个函数后，QuerySchedule的fragment_exec_params_就被填充完整了。下面分别介绍这三部分。
ComputeScanRangeAssignment
Status SimpleScheduler::ComputeScanRangeAssignment(const TQueryExecRequest& exec_request, QuerySchedule* schedule)
正如函数名所述，该函数主要处理scan range的分配，即每个scan range的data host的选择。这里主要是hdfs上的scan range需要处理，因为每个block有多个replica，有多个DataNode可供选择，而hbase上的region只归一个RegionServer（即这里的data host）管理。
exec_request.per_node_scan_ranges是一个map，记录了每个scan node对应的所有scan ranges。该函数的主体是一个for循环，每次处理一个scan node。
for (entry = exec_request.per_node_scan_ranges.begin();
entry != exec_request.per_node_scan_ranges.end(); ++entry) {
const TPlanNodeId node_id = entry->first; // 获取该scan node在整个执行计划树中的id
int fragment_idx = schedule->GetFragmentIdx(node_id); // 获取该scan node所在的fragment的下标。
const TPlanFragment& fragment = exec_request.fragments[fragment_idx]; // 获取该scan node所在的fragment引用
bool exec_at_coord = (fragment.partition.type == TPartitionType::UNPARTITIONED); // 该scan node是否在Coordinator处执行 // 获取该scan node的引用。schedule->GetNodeIdx返回的是node在其fragment的执行计划树中的下标。
const TPlanNode& node = fragment.plan.nodes[schedule->GetNodeIdx(node_id)];
DCHECK_EQ(node.node_id, node_id); // 下面的TReplicaPreference类型是复本选择时的默认策略，是个枚举类型，总共有5种：
// CACHE_LOCAL, CACHE_RACK, DISK_LOCAL, DISK_RACK, REMOTE
const TReplicaPreference::type* node_replica_preference = node.__isset.hdfs_scan_node
&& node.hdfs_scan_node.__isset.replica_preference
? &node.hdfs_scan_node.replica_preference : NULL; // 如果scan node里有设置，就拿这个默认配置，否则记为NULL
// random_replica表示当有几个复本各方面条件都一样时，是否随机选择。否则直接选第一个复本。
bool node_random_replica = node.__isset.hdfs_scan_node &&
node.hdfs_scan_node.__isset.random_replica &&
node.hdfs_scan_node.random_replica; // FragmentScanRangeAssignment类型记录一个fragment里所有scan range的分配，即选择哪些复本。实际是一个map：
// typedef boost::unordered_map<TNetworkAddress, PerNodeScanRanges> FragmentScanRangeAssignment;
// 上面的TNetworkAddress是impalad的地址（用host+port表示），PerNodeScanRanges也是一个map，记录每个scan node的所有scan range：
// typedef std::map<TPlanNodeId, std::vector<TScanRangeParams> > PerNodeScanRanges;
// 下面这行获取该fragment的assignment引用，所有fragment执行信息（即FragmentExecParams）在schedule->exec_params()数组中
FragmentScanRangeAssignment* assignment =
&(*schedule->exec_params())[fragment_idx].scan_range_assignment;
// 根据以上获取的对象进行实质的分配，并把结果写入assignment对象中
RETURN_IF_ERROR(ComputeScanRangeAssignment(
node_id, node_replica_preference, node_random_replica, entry->second,
exec_request.host_list, exec_at_coord, schedule->query_options(), assignment));
schedule->AddScanRanges(entry->second.size()); // 更新schedule对象中的scan range计数
}
ComputeScanRangeAssignment
Status SimpleScheduler::ComputeScanRangeAssignment(
PlanNodeId node_id, const TReplicaPreference::type* node_replica_preference,
bool node_random_replica, const vector<TScanRangeLocations>& locations,
const vector<TNetworkAddress>& host_list, bool exec_at_coord,
const TQueryOptions& query_options, FragmentScanRangeAssignment* assignment)
逐个过下参数：
- node_id：要处理的scan node的id
- node_replica_preference：replica的默认选择策略，是个枚举类例，取值可以是CACHE_LOCAL、DISK_LOCAL等，没有的话就是NULL
- node_random_replica：对于条件相同的replica，是否随机选择。否的话就直接选第一个。
- locations：各个scan range（也即block）的location列表（replica列表）
- host_list：复本所在的host列表，即DataNode列表
- exec_at_coord：该scan node是否在Coordinator处运行
- query_options：各种查询参数，详见TQueryOption定义
- assignment：调度结果存储在该对象中
函数的主体是两层的for循环，第一层为每个scan range，第二层为该scan range的各个location，也即replica。
代码中比较难懂的局部变量是base_distance，所谓的distance不过是CACHE_LOCAL, CACHE_RACK, DISK_LOCAL, DISK_RACK, REMOTE中的一种，表示impalad和它要读取的数据的距离。base_distance是一个基准线，就是说距离比这个还小（优）的我们就当成一样的来对待了。
另一个局部变量random_non_cached_tiebreak意义与参数node_random_replica相同，就是条件都一样的replica中是随机挑一个还是直接选第一个。
除去profile和打log的代码，函数主体的伪代码如下：
assigned_bytes_per_host是一个map，记录各个DataNode被分配的数据量，即有多少数据要从该DataNode读取
foreach ScanRange in ScanNode
//min_distance 记录replica到impalad的最小距离，初始值为REMOTE
// min_assigned_bytes，该变量帮助我们找到被分配任务最少的DataNode。每个replica从属于一个DataNode，如果有几个replica的distance相等，则比较它们所在DataNode所要读取的数据量。
// 这几个局部变量记录最终的选择结果：data_host、volume_id、is_cached、remote_read
//num_equivalent_replicas记录各方面条件都相同的replica数目
foreach Location in ScanRange
// replica_host 是replica所在的DataNode
//计算memory_distance，如果优于base_distance则折合成base_distance
//assigned_bytes 是replica所在DataNode当前被分配的数据量
//bool变量found_new_replica表示是否选择该replica
if (memory_distance < min_distance) {
min_distance = memory_distance;
num_equivalent_replicas = ;
found_new_replica = true;
} else if (memory_distance == min_distance) {
bool cached_replica = memory_distance == TReplicaPreference::CACHE_LOCAL;
if (assigned_bytes < min_assigned_bytes) {
num_equivalent_replicas = ;
found_new_replica = true;
} else if (assigned_bytes == min_assigned_bytes &&
(random_non_cached_tiebreak || cached_replica)) {
// 如果之前已经有k个相等的replica，则以1/(k+1)的概率决定是否采用本replica.
// 这样前面的replica被选中的概率各是 1/k * k/(k+1) = 1/(k+1)，从而实现随机挑选
++num_equivalent_replicas;
const int r = rand(); // make debugging easier.
found_new_replica = (r % num_equivalent_replicas == );
}
}
if (found_new_replica) { // 如果采用当前replica，则更新一系列值
min_assigned_bytes = assigned_bytes;
data_host = &replica_host;
volume_id = location.volume_id;
is_cached = location.is_cached;
remote_read = min_distance == TReplicaPreference::REMOTE;
}
} // end of each location (replica) 更新remote_bytes、remote_hosts、local_bytes、cached_bytes、assigned_bytes_per_host 为data_host找一个backend host(即exec_hostport)，如果该DataNode所在机器上就有impalad，则选择该impalad，
否则以round robin（轮循）的方式在impalad列表中挑一个。另外，如果函数参数中exec_at_coord为true，
则该scan node只在Coordinator所在impalad上执行，也就是当前机器。 ///////////// 把结果存入assignment中 /////////////
PerNodeScanRanges* scan_ranges =
FindOrInsert(assignment, exec_hostport, PerNodeScanRanges());
vector<TScanRangeParams>* scan_range_params_list =
FindOrInsert(scan_ranges, node_id, vector<TScanRangeParams>());
// add scan range
TScanRangeParams scan_range_params;
scan_range_params.scan_range = scan_range_locations.scan_range;
// Explicitly set the optional fields.
scan_range_params.__set_volume_id(volume_id);
scan_range_params.__set_is_cached(is_cached);
scan_range_params.__set_is_remote(remote_read);
scan_range_params_list->push_back(scan_range_params);
} // end of each scan range
回顾一下，assignment是一个映射，记录每个impalad的scan任务，存在一个PerNodeScanRanges对象中。每个PerNodeScanRanges对象也是一个映射，记录一个scan node在该impalad要读的数据，用vector表示。每个TScanRangeParams表示一个读任务，如果is_remote为false，则在所在impalad机器上的DataNode处读取，此时is_cached表示数据是否已缓存在该impalad中，volume_id记录了数据在哪块磁盘。如果is_remote为true，则根据PlanNodes.TScanRange去获取数据源来读取，使用hdfs或hbase的api。
ComputeFragmentHosts
void SimpleScheduler::ComputeFragmentHosts(const TQueryExecRequest& exec_request,QuerySchedule* schedule)
上一个Compute函数进行了replica的选取，接下来要给出每个fragment instance在哪个impalad上去执行。
这个函数的输出是schedule->exec_params()中各个FragmentExecParams的hosts部分，最后还会更新下schedule里的unique_hosts_，即整个query涉及到的其它impalad（除去本机）
vector<FragmentExecParams>* fragment_exec_params = schedule->exec_params(); // 拿出要修改的对象列表
TNetworkAddress coord = MakeNetworkAddress(FLAGS_hostname, FLAGS_be_port); // Coordinator地址即是本进程地址
DCHECK_EQ(fragment_exec_params->size(), exec_request.fragments.size());
vector<TPlanNodeType::type> scan_node_types; // 目前的scan node总共有三种类型
scan_node_types.push_back(TPlanNodeType::HDFS_SCAN_NODE);
scan_node_types.push_back(TPlanNodeType::HBASE_SCAN_NODE);
scan_node_types.push_back(TPlanNodeType::DATA_SOURCE_NODE); // 从后往前处理各个fragment，因为前面的fragment依赖于后面的输出，可能会调度到后面fragment同样的hosts上去，
// 因此底层的fragment要先处理。
for (int i = exec_request.fragments.size() - ; i >= ; --i) {
const TPlanFragment& fragment = exec_request.fragments[i];
[/i]FragmentExecParams& params = (*fragment_exec_params);
if (fragment.partition.type == TPartitionType::UNPARTITIONED) {
// all single-node fragments run on the coordinator host
params.hosts.push_back(coord);
continue;
} // UnionNode会通过ExchangeNode接收其它fragment的输入，也会接收所在fragment的scan结果。
// 包含UnionNode的fragment不仅会被调度到scan range选择的replica所在机器上，也会被调度到所有输入fragment的instance所在的机器上。
// （使得以partitioned join或grouping aggregate为子结点的UnionNode所运行的机器数目，不小于其子结点的输入数目）
if (ContainsNode(fragment.plan, TPlanNodeType::UNION_NODE)) {
vector<TPlanNodeId> scan_nodes;
FindNodes(fragment.plan, scan_node_types, &scan_nodes); // 取出该fragment的所有scan node
vector<TPlanNodeId> exch_nodes;
FindNodes(fragment.plan,
vector<TPlanNodeType::type>(, TPlanNodeType::EXCHANGE_NODE),
&exch_nodes); // 取出该fragment的所有exchange node // 把scan nodes所选择的host加入进来
vector<TNetworkAddress> scan_hosts;
for (int j = ; j < scan_nodes.size(); ++j) {
GetScanHosts(scan_nodes[j], exec_request, params, &scan_hosts);
}
unordered_set<TNetworkAddress> hosts(scan_hosts.begin(), scan_hosts.end()); // 把input fragments所选择的host加入进来
for (int j = ; j < exch_nodes.size(); ++j) { // 处理每个exchange node
int input_fragment_idx = FindSenderFragment(exch_nodes[j], i, exec_request); // 找到该exchange node的输入fragment
// 获取该输入fragment的所有hosts，所有输入的fragment应该在处理本fragment之前就已经计算好hosts了
const vector<TNetworkAddress>& input_fragment_hosts =
(*fragment_exec_params)[input_fragment_idx].hosts;
hosts.insert(input_fragment_hosts.begin(), input_fragment_hosts.end());
}
DCHECK(!hosts.empty()) << "no hosts for fragment " << i << " with a UnionNode"; params.hosts.assign(hosts.begin(), hosts.end());
continue;
} // 查看最左结点是否是scan node. 注意传入的types是scan_node_types
PlanNodeId leftmost_scan_id = FindLeftmostNode(fragment.plan, scan_node_types);
if (leftmost_scan_id == g_ImpalaInternalService_constants.INVALID_PLAN_NODE_ID) {
// 如果最左节点不是scan node，那只能是exchange node。我们把本fragment调度到其对应的输入fragment所在的所有机器上。
// 从而像分布式aggregation这类的fragment可以在本机获得输入fragment的数据
int input_fragment_idx = FindLeftmostInputFragment(i, exec_request);
params.hosts = (*fragment_exec_params)[input_fragment_idx].hosts;
continue;
} // 找到了最左的scan node，本fragment将在其scan ranges选定的各个replica机器上去执行
GetScanHosts(leftmost_scan_id, exec_request, params, ¶ms.hosts);
} // 最后再填一下schedule对象中的unique_hosts，即把所有用到的hosts集合的并集
unordered_set<TNetworkAddress> unique_hosts;
BOOST_FOREACH(const FragmentExecParams& exec_params, *fragment_exec_params) {
unique_hosts.insert(exec_params.hosts.begin(), exec_params.hosts.end());
} schedule->SetUniqueHosts(unique_hosts);
}
ComputeFragmentExecParams
void SimpleScheduler::ComputeFragmentExecParams(const TQueryExecRequest& exec_request,QuerySchedule* schedule)
这是SimpleSchedule::Schedule里调用的最后一个Compute函数，用来把各个FragmentExecParams剩下的内容填补完。
函数主体有两部分：先为各个fragment instance分配id，都有了id之后就可以指明各个instance的输入输出，把它们串成一个有向无环图。
这段代码不难了，有以下两点需要解释：
- 根fragment没有输出，其它的每个fragment都会输出到另一个fragment中的exchange node去，因此需要把目标fragment的各个instance id记录下来，也就是记录本fragment要有几个sender来发送数据，各发到哪些instance去。另外，每个exchange node也要知道有多少个instance会给自己发数据，即sender数目。
- 每个instance的id是一个128位整数（由两个64位整数表示，类型为TUniqueId），其中的高64位与query id的高64位一样，低64位是query id的低64位加上其是第几个instance。
总结
读懂SimpleScheduler的代码对后续阅读Impala更底层的代码很有帮助。在这里可以弄明白很多基本概念，如Plan Fragment Instance、Scan Range、Exchange Node等。
SimpleScheduler的调度策略很直接，把包含Scan Node的Plan Fragment Instance优先调度到数据所在的机器上去运行，同时会考虑均摊对HDFS DataNode的负载。如果数据所在机器上没有impalad，则用round robin的方式选一个远程的impalad。另外上层的Plan Fragment会尽量在其依赖的Plan Fragment所在机器上去运行。